shopdb-flask/tests/test_plugins/test_geenforce_crud.py

"""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_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_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