shopdb-flask/tests/test_plugins/test_geenforce_parity.py

"""GE-Enforce P1 parity gate (Gate A): import + export is behaviorally lossless.

Runs the DB-free parity harness (parse -> in-memory rows -> render -> compare)
over a synthetic, site-neutral manifest that exercises every entry type,
detection method, targeting filter, RegValue typing, InUseCheck, and preinstall
flag. The synthetic fixture is what CI gates on (no real site data is vendored
into the framework repo). If the real GE-Enforce reference share is present
(dev), it is also checked.
"""

import os

import pytest

from plugins.geenforce.parity import run_parity, load_fixtures
from plugins.geenforce.importer import discover_share, load_manifest_file


# A site-neutral manifest covering the whole schema surface.
SYNTHETIC = {
    'Version': '2.6',
    '_comment': 'Synthetic parity fixture (not a real site manifest).',
    'Applications': [
        {
            '_comment': 'MSI with Registry detection, version gate, InUseCheck.',
            'Name': 'Sample MSI', 'Type': 'MSI',
            'Installer': 'apps/sample.msi',
            'InstallArgs': '/qn /norestart ALLUSERS=1',
            '_CmmVersion': '2019',
            'DetectionMethod': 'Registry',
            'DetectionPath': 'HKLM:\\SOFTWARE\\Sample',
            'DetectionName': 'DisplayVersion', 'DetectionValue': '1.2.3.4',
            'InUseCheck': {
                'Behavior': 'CloseAndReopen',
                'Processes': [
                    {'Name': 'sample', 'ExePath': 'C:\\sample.exe',
                     'GracefulCloseTimeoutSec': 15},
                    {'Name': 'other'},
                ],
            },
        },
        {
            'Name': 'Sample EXE', 'Type': 'EXE', 'Installer': 'apps/sample.exe',
            'InstallArgs': '/quiet', 'WaitTimeoutSec': 60,
            'DetectionMethod': 'FileVersion',
            'DetectionPath': 'C:\\sample.exe', 'DetectionValue': '1.0.0.0',
        },
        {
            'Name': 'Sample PS1', 'Type': 'PS1', 'Script': 'scripts/run.ps1',
            'Args': '-Force', 'DetectionMethod': 'Always',
        },
        {
            'Name': 'Sample File', 'Type': 'File', 'Source': 'configs/app.json',
            'Destination': 'C:\\ProgramData\\app.json',
            'DetectionMethod': 'Hash', 'DetectionPath': 'C:\\ProgramData\\app.json',
            'DetectionValue': 'a' * 64,
            'PCTypes': ['gea-shopfloor-collections', 'gea-shopfloor-cmm'],
        },
        {
            'Name': 'Sample Registry DWord', 'Type': 'Registry',
            'RegPath': 'HKLM:\\SOFTWARE\\Sample', 'RegName': 'Enabled',
            'RegValue': 1, 'RegType': 'DWord',
            'DetectionMethod': 'ValueMatches',
            'DetectionPath': 'HKLM:\\SOFTWARE\\Sample', 'DetectionName': 'Enabled',
        },
        {
            'Name': 'Sample Registry String', 'Type': 'Registry',
            'RegPath': 'HKLM:\\SOFTWARE\\Sample', 'RegName': 'Mode',
            'RegValue': 'on', 'RegType': 'String',
        },
        {
            'Name': 'Sample INF', 'Type': 'INF', 'Installer': 'configs/driver.inf',
            'DetectionMethod': 'pnputil', 'DetectionPattern': 'SampleDriver',
        },
        {
            'Name': 'Sample bay-gated', 'Type': 'CMD', 'Installer': 'scripts/bay.cmd',
            'TargetMachineNumbers': ['3201', '3202'],
        },
        {
            'Name': 'Sample host-gated', 'Type': 'BAT', 'Installer': 'scripts/host.bat',
            'TargetHostnames': ['WJS-*', 'WJPC0615'],
        },
        {
            'Name': 'Sample always-installs (no detection)', 'Type': 'PS1',
            'Script': 'scripts/every.ps1',
        },
        {
            'Name': 'Sample preinstall-flagged', 'Type': 'EXE',
            'Installer': 'apps/pre.exe', 'PreEnrollment': True,
            'PCTypesStrict': True, 'KillAfterDetection': True,
            'PCTypes': ['gea-shopfloor-nocollections'],
        },
    ],
}

REFERENCE_SHARE = '/home/camp/pxe-images/tsgwp00525-v2/shared/dt/shopfloor'
REFERENCE_PREINSTALL = '/home/camp/projects/pxe/playbook/preinstall/preinstall.json'


def test_synthetic_manifest_round_trips_losslessly():
    """The synthetic manifest imports + exports with full behavioral parity."""
    results, ok = run_parity([('synthetic', 'runtime', SYNTHETIC)])
    result = results[0]
    assert result['entries_identical'] == result['entries_total'], result['firstdiff']
    assert result['profiles_same'] == result['profiles_total'], result['firstdiff']
    assert ok


def test_regvalue_numeric_type_preserved():
    """A DWord RegValue of 1 round-trips as int 1, not the string '1'."""
    from plugins.geenforce.importer import build_scope
    from plugins.geenforce.serializer import scope_to_manifest

    rebuilt = scope_to_manifest(build_scope('synthetic', 'runtime', SYNTHETIC))
    dword = next(e for e in rebuilt['Applications']
                 if e['Name'] == 'Sample Registry DWord')
    assert dword['RegValue'] == 1
    assert isinstance(dword['RegValue'], int)


@pytest.mark.skipif(not os.path.isdir(REFERENCE_SHARE),
                    reason='GE-Enforce reference share not present')
def test_reference_share_round_trips_losslessly():
    """Every real on-share manifest round-trips (the live Gate A)."""
    manifests = list(discover_share(REFERENCE_SHARE))
    if os.path.isfile(REFERENCE_PREINSTALL):
        manifests.append(('preinstall', 'preinstall',
                          load_manifest_file(REFERENCE_PREINSTALL)))
    results, ok = run_parity(manifests)
    failed = [r for r in results if not r['passed']]
    assert ok, f"parity failures: {[(r['scopename'], r['firstdiff']) for r in failed]}"


@pytest.mark.skipif(not os.path.isdir(REFERENCE_SHARE),
                    reason='GE-Enforce reference share not present')
def test_reference_manifests_fit_column_limits():
    """Guard the truncation blind spot: the DB-free parity harness can't catch a
    value that exceeds its column length (Python strings are unbounded), so a
    260-char DetectionPath would pass parity yet truncate on a real insert.
    Assert every real-manifest string field fits its declared column, deriving
    limits from the model so this never drifts.
    """
    from plugins.geenforce.models import (
        ManifestEntry, ManifestEntryPcType, ManifestEntryHostname,
        ManifestEntryMachineNumber)

    def limit(model, attr):
        return model.__table__.columns[attr].type.length

    # manifest key -> (model, column attribute)
    entry_fields = {
        'Name': 'name', 'Installer': 'installer', 'Script': 'scriptpath',
        'Args': 'scriptargs', 'Source': 'sourcepath', 'Destination': 'destination',
        'RegPath': 'regpath', 'RegName': 'regname', 'RegType': 'regtype',
        'DetectionPath': 'detectionpath', 'DetectionName': 'detectionname',
        'DetectionValue': 'detectionvalue', 'DetectionPattern': 'detectionpattern',
        '_CmmVersion': 'cmmversion', 'LogFile': 'logfile',
        'UpdateWindow': 'updatewindow', 'ApplyMode': 'applymode',
    }
    overflows = []
    manifests = list(discover_share(REFERENCE_SHARE))
    if os.path.isfile(REFERENCE_PREINSTALL):
        manifests.append(('preinstall', 'preinstall',
                          load_manifest_file(REFERENCE_PREINSTALL)))
    for scopename, _phase, manifest in manifests:
        for entry in manifest.get('Applications', []):
            for key, attr in entry_fields.items():
                value = entry.get(key)
                if isinstance(value, str) and len(value) > limit(ManifestEntry, attr):
                    overflows.append(f'{scopename}/{entry.get("Name")}.{key}')
            for value in entry.get('PCTypes', []):
                if len(value) > limit(ManifestEntryPcType, 'pctypevalue'):
                    overflows.append(f'{scopename}/{entry.get("Name")}.PCTypes')
            for value in entry.get('TargetHostnames', []):
                if len(value) > limit(ManifestEntryHostname, 'hostnamepattern'):
                    overflows.append(f'{scopename}/{entry.get("Name")}.TargetHostnames')
            for value in entry.get('TargetMachineNumbers', []):
                if len(str(value)) > limit(ManifestEntryMachineNumber, 'machinenumber'):
                    overflows.append(f'{scopename}/{entry.get("Name")}.TargetMachineNumbers')
    assert not overflows, f'fields exceed column limits (would truncate): {overflows}'


def test_fixtures_cover_every_pctype():
    """The machine-profile fixtures include one of each imaging pctype."""
    labels = {f['pctype'] for f in load_fixtures()}
    for pctype in ('gea-shopfloor-collections', 'gea-shopfloor-nocollections',
                   'gea-shopfloor-common', 'gea-shopfloor-cmm',
                   'gea-shopfloor-genspect', 'gea-shopfloor-heattreat',
                   'gea-shopfloor-waxtrace', 'gea-shopfloor-partmarker',
                   'gea-shopfloor-display'):
        assert pctype in labels