Files
shopdb-flask/tests/test_plugins/test_geenforce_crud.py
cproudlock 0dcd186820
All checks were successful
CI / backend (push) Successful in 1m43s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 8s
Fix defects found in session review of GE-Enforce plugin
Consolidated fixes from a three-dimension adversarial review.

Data-loss (HIGH): the manifest entry editor stripped fields the form did not
expose, because PUT /entries is a full reset-then-apply. The form now captures
everything - InUseCheck processes as structured name/ExePath/timeout rows (not
just names), LogFile, and the three preinstall flags as checkboxes; the dead
payload-source control (never wired) is removed. New regression test proves an
edit preserves ExePath/timeout/LogFile/PreEnrollment/PCTypesStrict.

Update-entry crash (found by that regression test): replacing an entry's
one-to-one InUseCheck (unique entryid) collided with the old row mid-flush ->
IntegrityError -> 400. update_entry now frees the old InUseCheck (delete+flush)
before populate re-inserts it.

Export truncation (MEDIUM): export_scope_to_share used a plain truncating open,
so a failed/partial write left the live on-share manifest (every PC reads it)
empty. Now writes a temp file in the same dir and os.replace() atomically.

Report dedup case bug (MEDIUM, confirmed by scratch test): the iscurrent demote
matched hostname case-sensitively while the read path uses ilike, so a PC
reporting different casing left two iscurrent rows and double-counted. Demote is
now case-insensitive; regression test added.

Simulator fidelity (MEDIUM): PCTypesStrict was captured but ignored by the
filter mirror, so the simulator wrongly matched a collections-only strict entry
to a nocollections PC via the shared Standard alias group. matches_pctype now
honors PCTypesStrict (disables alias expansion); test added.

Hardening: removed the dead/unscoped GEENFORCE_API_KEY env fallback (never wired
into config; tokens are the only path); create/update entry return 400 on a
duplicate Name instead of 500; parity now asserts scope-level Version/Site; a
new test guards real-manifest field lengths against column limits (the DB-free
parity harness can't see truncation); error handling added to the previously
unguarded editor + reports API calls.

Full suite green; naming + frontend build green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-12 20:46:09 -04:00

227 lines
10 KiB
Python

"""GE-Enforce P2 admin CRUD: scopes, entries, reorder, simulate, publish cycle.
All admin endpoints are JWT + geenforce.manage/publish gated (the admin user
holds both). Entry payloads use the manifest Applications[] shape.
"""
def _create_scope(client, auth_headers, name='gea-shopfloor-test'):
resp = client.post('/api/geenforce/scopes',
json={'scopename': name, 'phase': 'runtime',
'description': 'test scope'},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
return resp.get_json()['data']['scopeid']
def _add_entry(client, auth_headers, scopeid, entry):
resp = client.post(f'/api/geenforce/scopes/{scopeid}/entries',
json=entry, headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
return resp.get_json()['data']['entryid']
def test_scope_crud(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers)
got = client.get(f'/api/geenforce/scopes/{scopeid}', headers=auth_headers)
assert got.status_code == 200
assert got.get_json()['data']['scopename'] == 'gea-shopfloor-test'
upd = client.put(f'/api/geenforce/scopes/{scopeid}',
json={'description': 'updated', 'computertypeid': 3},
headers=auth_headers)
assert upd.get_json()['data']['computertypeid'] == 3
dele = client.delete(f'/api/geenforce/scopes/{scopeid}', headers=auth_headers)
assert dele.status_code == 200
assert client.get(f'/api/geenforce/scopes/{scopeid}',
headers=auth_headers).status_code == 404
def test_duplicate_scope_rejected(client, db, auth_headers):
_create_scope(client, auth_headers, 'gea-shopfloor-dup')
resp = client.post('/api/geenforce/scopes',
json={'scopename': 'gea-shopfloor-dup', 'phase': 'runtime'},
headers=auth_headers)
assert resp.status_code == 400
def test_entry_crud_and_reorder(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers)
a = _add_entry(client, auth_headers, scopeid,
{'Name': 'Alpha', 'Type': 'MSI', 'Installer': 'apps/a.msi'})
b = _add_entry(client, auth_headers, scopeid,
{'Name': 'Beta', 'Type': 'PS1', 'Script': 'scripts/b.ps1'})
# Update Alpha: change type + add detection.
upd = client.put(f'/api/geenforce/entries/{a}',
json={'Name': 'Alpha', 'Type': 'EXE', 'Installer': 'apps/a.exe',
'DetectionMethod': 'FileVersion',
'DetectionPath': 'C:\\a.exe', 'DetectionValue': '1.0'},
headers=auth_headers)
assert upd.status_code == 200
assert upd.get_json()['data']['Type'] == 'EXE'
assert upd.get_json()['data']['DetectionMethod'] == 'FileVersion'
# Reorder: Beta first.
reo = client.put(f'/api/geenforce/scopes/{scopeid}/entries/reorder',
json={'order': [b, a]}, headers=auth_headers)
assert reo.status_code == 200
entries = client.get(f'/api/geenforce/scopes/{scopeid}',
headers=auth_headers).get_json()['data']['entries']
assert [e['Name'] for e in entries] == ['Beta', 'Alpha']
# Delete Beta.
assert client.delete(f'/api/geenforce/entries/{b}',
headers=auth_headers).status_code == 200
entries = client.get(f'/api/geenforce/scopes/{scopeid}',
headers=auth_headers).get_json()['data']['entries']
assert [e['Name'] for e in entries] == ['Alpha']
def test_entry_edit_preserves_full_fidelity(client, db, auth_headers):
"""Editing an entry with the complete payload keeps InUseCheck ExePath +
timeout, LogFile, and preinstall flags (the UI now sends all of them)."""
scopeid = _create_scope(client, auth_headers)
rich = {
'Name': 'PC-DMIS', 'Type': 'MSI', 'Installer': 'apps/pcdmis.msi',
'LogFile': 'C:\\Logs\\pcdmis.log',
'PreEnrollment': True, 'PCTypesStrict': True,
'InUseCheck': {'Behavior': 'CloseAndReopen', 'Processes': [
{'Name': 'PCDLRN', 'ExePath': 'C:\\PCDLRN.exe',
'GracefulCloseTimeoutSec': 15}]},
}
entryid = _add_entry(client, auth_headers, scopeid, rich)
# Re-send the same full payload (what the editor now does) and confirm
# nothing is stripped.
upd = client.put(f'/api/geenforce/entries/{entryid}', json=rich,
headers=auth_headers)
assert upd.status_code == 200
entry = upd.get_json()['data']
assert entry['LogFile'] == 'C:\\Logs\\pcdmis.log'
assert entry['PreEnrollment'] is True
assert entry['PCTypesStrict'] is True
proc = entry['InUseCheck']['Processes'][0]
assert proc['ExePath'] == 'C:\\PCDLRN.exe'
assert proc['GracefulCloseTimeoutSec'] == 15
def test_invalid_entry_type_rejected(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers)
resp = client.post(f'/api/geenforce/scopes/{scopeid}/entries',
json={'Name': 'Bad', 'Type': 'NOTATYPE'},
headers=auth_headers)
assert resp.status_code == 400
def test_reorder_rejects_foreign_ids(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers)
a = _add_entry(client, auth_headers, scopeid,
{'Name': 'Alpha', 'Type': 'MSI'})
resp = client.put(f'/api/geenforce/scopes/{scopeid}/entries/reorder',
json={'order': [a, 9999]}, headers=auth_headers)
assert resp.status_code == 400
def test_simulate_cmm_version_gate(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers, 'gea-shopfloor-cmm')
_add_entry(client, auth_headers, scopeid,
{'Name': 'CLM (untagged)', 'Type': 'MSI', 'Installer': 'apps/clm.msi'})
_add_entry(client, auth_headers, scopeid,
{'Name': 'PC-DMIS 2016', 'Type': 'MSI', 'Installer': 'apps/16.msi',
'_CmmVersion': '2016'})
_add_entry(client, auth_headers, scopeid,
{'Name': 'PC-DMIS 2019', 'Type': 'MSI', 'Installer': 'apps/19.msi',
'_CmmVersion': '2019'})
sim = client.get(f'/api/geenforce/scopes/{scopeid}/simulate?cmmversion=2019',
headers=auth_headers)
assert sim.status_code == 200
data = sim.get_json()['data']
assert data['applied'] == ['CLM (untagged)', 'PC-DMIS 2019']
filtered = {f['name']: f['filteredby'] for f in data['filtered']}
assert filtered['PC-DMIS 2016'] == ['_CmmVersion']
def test_duplicate_entry_name_is_400_not_500(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers)
_add_entry(client, auth_headers, scopeid, {'Name': 'Alpha', 'Type': 'MSI'})
dup = client.post(f'/api/geenforce/scopes/{scopeid}/entries',
json={'Name': 'Alpha', 'Type': 'PS1'}, headers=auth_headers)
assert dup.status_code == 400
def test_simulate_pctypesstrict_disables_alias(client, db, auth_headers):
"""A collections-only strict entry must NOT match a nocollections PC via the
shared Standard alias group (mirrors the preinstall UDC entry)."""
scopeid = _create_scope(client, auth_headers, 'preinstall')
_add_entry(client, auth_headers, scopeid,
{'Name': 'UDC (strict)', 'Type': 'EXE', 'Installer': 'apps/udc.exe',
'PCTypes': ['gea-shopfloor-collections'], 'PCTypesStrict': True})
_add_entry(client, auth_headers, scopeid,
{'Name': 'UDC (loose)', 'Type': 'EXE', 'Installer': 'apps/udc.exe',
'PCTypes': ['gea-shopfloor-collections']})
sim = client.get(
f'/api/geenforce/scopes/{scopeid}/simulate?pctype=gea-shopfloor-nocollections',
headers=auth_headers).get_json()['data']
# loose entry matches via the Standard alias group; strict entry does not.
assert sim['applied'] == ['UDC (loose)']
filtered = {f['name']: f['filteredby'] for f in sim['filtered']}
assert filtered['UDC (strict)'] == ['PCTypes']
def test_simulate_machine_number_gate(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers, 'gea-shopfloor-collections')
_add_entry(client, auth_headers, scopeid,
{'Name': 'Okuma cred', 'Type': 'PS1', 'Script': 'scripts/o.ps1',
'TargetMachineNumbers': ['3201', '3202']})
_add_entry(client, auth_headers, scopeid,
{'Name': 'Everyone', 'Type': 'PS1', 'Script': 'scripts/e.ps1'})
match = client.get(
f'/api/geenforce/scopes/{scopeid}/simulate?machinenumber=3201',
headers=auth_headers).get_json()['data']
assert set(match['applied']) == {'Okuma cred', 'Everyone'}
nomatch = client.get(
f'/api/geenforce/scopes/{scopeid}/simulate?machinenumber=9999',
headers=auth_headers).get_json()['data']
assert nomatch['applied'] == ['Everyone']
def test_publish_versions_rollback(client, db, auth_headers):
scopeid = _create_scope(client, auth_headers, 'gea-shopfloor-cmm')
_add_entry(client, auth_headers, scopeid, {'Name': 'Alpha', 'Type': 'MSI'})
pub1 = client.post(f'/api/geenforce/scopes/{scopeid}/publish',
json={'notes': 'v1'}, headers=auth_headers)
assert pub1.status_code == 201
assert pub1.get_json()['data']['versionnumber'] == 1
_add_entry(client, auth_headers, scopeid, {'Name': 'Beta', 'Type': 'PS1'})
pub2 = client.post(f'/api/geenforce/scopes/{scopeid}/publish',
json={'notes': 'v2'}, headers=auth_headers)
assert pub2.get_json()['data']['versionnumber'] == 2
versions = client.get(f'/api/geenforce/scopes/{scopeid}/versions',
headers=auth_headers).get_json()['data']
assert [v['versionnumber'] for v in versions] == [2, 1]
assert versions[0]['iscurrent'] is True
# v2 manifest has both entries.
v2doc = client.get(f'/api/geenforce/scopes/{scopeid}/versions/2',
headers=auth_headers).get_json()['data']['manifest']
assert [e['Name'] for e in v2doc['Applications']] == ['Alpha', 'Beta']
# Rollback to v1.
rb = client.post(f'/api/geenforce/scopes/{scopeid}/rollback',
json={'versionnumber': 1}, headers=auth_headers)
assert rb.status_code == 200
versions = client.get(f'/api/geenforce/scopes/{scopeid}/versions',
headers=auth_headers).get_json()['data']
current = next(v for v in versions if v['iscurrent'])
assert current['versionnumber'] == 1