shopdb-flask/tests/test_plugins/test_geenforce_manifest.py

"""GE-Enforce first slice: import a scope, publish it, serve it to a client.

Exercises the vertical the plan calls the first slice (via a small synthetic
scope rather than the real gea-shopfloor-cmm): draft import -> publish ->
GET /api/geenforce/manifest with a geenforce.fetch service token; ETag/304;
draft edits never change the served bytes; publish + rollback change them.
"""

import json

import pytest

from plugins.geenforce.models import ManifestScope
from plugins.geenforce import service


SCOPE = {
    'Version': '2.6',
    'Applications': [
        {'Name': 'Alpha', 'Type': 'MSI', 'Installer': 'apps/alpha.msi',
         'DetectionMethod': 'Registry', 'DetectionPath': 'HKLM:\\SOFTWARE\\Alpha'},
        {'Name': 'Beta', 'Type': 'PS1', 'Script': 'scripts/beta.ps1',
         'DetectionMethod': 'Always'},
    ],
}


def _seed_and_publish(app, scopename='gea-shopfloor-cmm', manifest=None):
    with app.app_context():
        service.replace_scope_draft(scopename, 'runtime', manifest or SCOPE)
        service.publish_scope(scopename, 'runtime', notes='initial')
        service.db.session.commit()


def _mint_fetch_token(client, auth_headers):
    resp = client.post('/api/apitokens',
                       json={'name': 'geenforce svc', 'scopes': ['geenforce.fetch']},
                       headers=auth_headers)
    assert resp.status_code == 201, resp.get_json()
    return resp.get_json()['data']['secret']


def test_import_publish_serve(client, db, app, auth_headers):
    _seed_and_publish(app)
    secret = _mint_fetch_token(client, auth_headers)

    resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                      headers={'X-API-Key': secret})
    assert resp.status_code == 200, resp.data
    body = json.loads(resp.data)
    assert [e['Name'] for e in body['Applications']] == ['Alpha', 'Beta']
    assert resp.headers.get('ETag')
    assert resp.headers.get('X-Manifest-Version') == '1'


def test_etag_304(client, db, app, auth_headers):
    _seed_and_publish(app)
    secret = _mint_fetch_token(client, auth_headers)
    first = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                       headers={'X-API-Key': secret})
    etag = first.headers['ETag']
    again = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                       headers={'X-API-Key': secret, 'If-None-Match': etag})
    assert again.status_code == 304


def test_draft_edit_does_not_change_served_bytes(client, db, app, auth_headers):
    _seed_and_publish(app)
    secret = _mint_fetch_token(client, auth_headers)
    before = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                        headers={'X-API-Key': secret}).data

    # Edit the DRAFT (add an entry) but do NOT publish.
    changed = {'Version': '2.6', 'Applications': SCOPE['Applications'] + [
        {'Name': 'Gamma', 'Type': 'PS1', 'Script': 'scripts/gamma.ps1'}]}
    with app.app_context():
        service.replace_scope_draft('gea-shopfloor-cmm', 'runtime', changed)
        service.db.session.commit()

    after = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                       headers={'X-API-Key': secret}).data
    assert before == after  # served bytes come from the published snapshot only


def test_publish_then_rollback(client, db, app, auth_headers):
    _seed_and_publish(app)
    secret = _mint_fetch_token(client, auth_headers)
    v1 = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                    headers={'X-API-Key': secret}).data

    changed = {'Version': '2.6', 'Applications': SCOPE['Applications'] + [
        {'Name': 'Gamma', 'Type': 'PS1', 'Script': 'scripts/gamma.ps1'}]}
    with app.app_context():
        service.replace_scope_draft('gea-shopfloor-cmm', 'runtime', changed)
        service.publish_scope('gea-shopfloor-cmm', 'runtime', notes='add gamma')
        service.db.session.commit()

    v2 = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                    headers={'X-API-Key': secret})
    assert v2.headers['X-Manifest-Version'] == '2'
    assert b'Gamma' in v2.data

    with app.app_context():
        service.rollback_scope('gea-shopfloor-cmm', 'runtime', 1)
        service.db.session.commit()

    rolled = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                        headers={'X-API-Key': secret})
    assert rolled.data == v1  # byte-identical: published snapshots are frozen text


def test_unauthenticated_rejected(client, db, app):
    _seed_and_publish(app)
    resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm')
    assert resp.status_code == 401


def test_wrong_scope_rejected(client, db, app, auth_headers):
    _seed_and_publish(app)
    resp = client.post('/api/apitokens',
                       json={'name': 'wrong', 'scopes': ['collector.ingest']},
                       headers=auth_headers)
    secret = resp.get_json()['data']['secret']
    manifest = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
                          headers={'X-API-Key': secret})
    assert manifest.status_code == 401


def test_unpublished_scope_404(client, db, app, auth_headers):
    with app.app_context():
        service.replace_scope_draft('gea-shopfloor-keyence', 'runtime', SCOPE)
        service.db.session.commit()  # draft only, never published
    secret = _mint_fetch_token(client, auth_headers)
    resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-keyence',
                      headers={'X-API-Key': secret})
    assert resp.status_code == 404


def test_admin_list_and_preview(client, db, app, auth_headers):
    _seed_and_publish(app)
    listing = client.get('/api/geenforce/scopes', headers=auth_headers)
    assert listing.status_code == 200
    scopes = listing.get_json()['data']
    cmm = next(s for s in scopes if s['scopename'] == 'gea-shopfloor-cmm')
    assert cmm['entrycount'] == 2
    assert cmm['publishedversion'] == 1

    preview = client.get(f"/api/geenforce/scopes/{cmm['scopeid']}/preview",
                         headers=auth_headers)
    assert preview.status_code == 200
    assert [e['Name'] for e in preview.get_json()['data']['manifest']['Applications']] \
        == ['Alpha', 'Beta']