Add GE-Enforce P2 admin CRUD API: scopes, entries, reorder, simulate, publish
Full HTTP admin surface behind the manifest editor (geenforce.manage for edits, geenforce.publish for shipping): - Scopes: POST/GET/PUT/DELETE /scopes[/<id>] (create imaging PC types, edit the ComputerType/MeasuringToolType mapping + metadata, delete). - Entries: POST /scopes/<id>/entries, PUT/DELETE /entries/<id>. Payloads use the manifest Applications[] shape; populate_entry (refactored out of build_entry) updates an entry in place, resetting omitted fields and replacing children. - Reorder: PUT /scopes/<id>/entries/reorder enforces the ordering contract (body must list exactly the scope's entry ids). - Simulate: GET /scopes/<id>/simulate?pctype&subtype&hostname&machinenumber& cmmversion returns which entries apply and which filter excluded the rest, reusing the engine-mirror filters. The "what would this PC get" tool. - Publish lifecycle: POST /scopes/<id>/publish (records publishedby from JWT), GET /scopes/<id>/versions, GET .../versions/<n> (frozen manifest), POST /scopes/<id>/rollback. Entry type validated against ENTRY_TYPES; 8 CRUD tests. JWT+permission gated so the authz sweep covers them. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
@@ -20,9 +20,15 @@ from shopdb.api import (
|
||||
)
|
||||
|
||||
from ..models import (
|
||||
ManifestScope, ManifestPublishedVersion, ManifestEnforcementReport,
|
||||
ManifestScope, ManifestEntry, ManifestPublishedVersion,
|
||||
ManifestEnforcementReport, ENTRY_TYPES, PHASES,
|
||||
)
|
||||
from ..serializer import scope_to_manifest, entry_to_dict
|
||||
from ..importer import build_entry, populate_entry
|
||||
from ..filters import (
|
||||
entry_applies, matches_pctype, matches_hostname, matches_machinenumber,
|
||||
matches_cmmversion,
|
||||
)
|
||||
from ..serializer import scope_to_manifest
|
||||
from .. import service
|
||||
|
||||
geenforce_bp = Blueprint('geenforce', __name__)
|
||||
@@ -149,6 +155,292 @@ def preview_scope(scopeid):
|
||||
'manifest': scope_to_manifest(scope)})
|
||||
|
||||
|
||||
# -- scope CRUD (geenforce.manage) --------------------------------------------
|
||||
|
||||
def _scope_summary(scope):
|
||||
current = scope.publishedversions.filter_by(iscurrent=True).first()
|
||||
return {
|
||||
'scopeid': scope.scopeid,
|
||||
'scopename': scope.scopename,
|
||||
'phase': scope.phase,
|
||||
'manifestversion': scope.manifestversion,
|
||||
'description': scope.description,
|
||||
'computertypeid': scope.computertypeid,
|
||||
'measuringtooltypeid': scope.measuringtooltypeid,
|
||||
'iscommon': scope.iscommon,
|
||||
'entrycount': len(scope.entries),
|
||||
'publishedversion': current.versionnumber if current else None,
|
||||
}
|
||||
|
||||
|
||||
def _entry_payload(entry):
|
||||
"""Manifest-entry dict with entryid + sortorder for the editor."""
|
||||
data = {'entryid': entry.entryid, 'sortorder': entry.sortorder}
|
||||
data.update(entry_to_dict(entry))
|
||||
return data
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def create_scope():
|
||||
payload = request.get_json(silent=True) or {}
|
||||
scopename = (payload.get('scopename') or '').strip()
|
||||
phase = (payload.get('phase') or 'runtime').strip()
|
||||
if not scopename:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'scopename is required', http_code=400)
|
||||
if phase not in PHASES:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
f'phase must be one of {PHASES}', http_code=400)
|
||||
if ManifestScope.query.filter_by(scopename=scopename, phase=phase).first():
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'scope already exists', http_code=400)
|
||||
scope = ManifestScope(
|
||||
scopename=scopename, phase=phase,
|
||||
manifestversion=str(payload.get('manifestversion', '1.0')),
|
||||
description=payload.get('description'),
|
||||
computertypeid=payload.get('computertypeid'),
|
||||
measuringtooltypeid=payload.get('measuringtooltypeid'),
|
||||
iscommon=bool(payload.get('iscommon', scopename == 'common')))
|
||||
db.session.add(scope)
|
||||
db.session.commit()
|
||||
return success_response(_scope_summary(scope), http_code=201)
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>', methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def get_scope(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
data = _scope_summary(scope)
|
||||
data['entries'] = [_entry_payload(e) for e in scope.entries]
|
||||
return success_response(data)
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def update_scope(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
payload = request.get_json(silent=True) or {}
|
||||
for field in ('description', 'computertypeid', 'measuringtooltypeid'):
|
||||
if field in payload:
|
||||
setattr(scope, field, payload[field])
|
||||
if 'manifestversion' in payload:
|
||||
scope.manifestversion = str(payload['manifestversion'])
|
||||
if 'iscommon' in payload:
|
||||
scope.iscommon = bool(payload['iscommon'])
|
||||
db.session.commit()
|
||||
return success_response(_scope_summary(scope))
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def delete_scope(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
db.session.delete(scope)
|
||||
db.session.commit()
|
||||
return success_response({'deleted': scopeid})
|
||||
|
||||
|
||||
# -- entry CRUD (geenforce.manage) --------------------------------------------
|
||||
|
||||
def _validate_entry(payload):
|
||||
"""Return an error string, or None if the entry payload is valid."""
|
||||
if not (payload.get('Name') or '').strip():
|
||||
return 'entry Name is required'
|
||||
if payload.get('Type') not in ENTRY_TYPES:
|
||||
return f'entry Type must be one of {ENTRY_TYPES}'
|
||||
return None
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/entries', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def create_entry(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
payload = request.get_json(silent=True) or {}
|
||||
invalid = _validate_entry(payload)
|
||||
if invalid:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, invalid, http_code=400)
|
||||
nextorder = max([e.sortorder for e in scope.entries], default=-1) + 1
|
||||
entry = build_entry(payload, nextorder)
|
||||
scope.entries.append(entry)
|
||||
db.session.commit()
|
||||
return success_response(_entry_payload(entry), http_code=201)
|
||||
|
||||
|
||||
@geenforce_bp.route('/entries/<int:entryid>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def update_entry(entryid):
|
||||
entry = db.session.get(ManifestEntry, entryid)
|
||||
if not entry:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such entry', http_code=404)
|
||||
payload = request.get_json(silent=True) or {}
|
||||
invalid = _validate_entry(payload)
|
||||
if invalid:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, invalid, http_code=400)
|
||||
populate_entry(entry, payload)
|
||||
db.session.commit()
|
||||
return success_response(_entry_payload(entry))
|
||||
|
||||
|
||||
@geenforce_bp.route('/entries/<int:entryid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def delete_entry(entryid):
|
||||
entry = db.session.get(ManifestEntry, entryid)
|
||||
if not entry:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such entry', http_code=404)
|
||||
db.session.delete(entry)
|
||||
db.session.commit()
|
||||
return success_response({'deleted': entryid})
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/entries/reorder', methods=['PUT'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def reorder_entries(scopeid):
|
||||
"""Set entry order from a list of entryids (the ordering contract)."""
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
order = (request.get_json(silent=True) or {}).get('order') or []
|
||||
owned = {e.entryid: e for e in scope.entries}
|
||||
if set(order) != set(owned):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'order must list exactly this scope\'s entry ids',
|
||||
http_code=400)
|
||||
for position, entryid in enumerate(order):
|
||||
owned[entryid].sortorder = position
|
||||
db.session.commit()
|
||||
return success_response({'order': order})
|
||||
|
||||
|
||||
# -- simulate: "what would this PC get" ---------------------------------------
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/simulate', methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def simulate_scope(scopeid):
|
||||
"""Which draft entries apply to a given machine profile, and why the rest
|
||||
are filtered out (reuses the engine-mirror filters)."""
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
profile = {
|
||||
'pctype': request.args.get('pctype') or scope.scopename,
|
||||
'subtype': request.args.get('subtype'),
|
||||
'hostname': request.args.get('hostname'),
|
||||
'machinenumber': request.args.get('machinenumber'),
|
||||
'cmmversion': request.args.get('cmmversion'),
|
||||
}
|
||||
applied, filtered = [], []
|
||||
for entry in scope.entries:
|
||||
entry_dict = entry_to_dict(entry)
|
||||
if entry_applies(entry_dict, profile):
|
||||
applied.append(entry.name)
|
||||
else:
|
||||
reasons = []
|
||||
if not matches_pctype(entry_dict, profile['pctype'], profile['subtype']):
|
||||
reasons.append('PCTypes')
|
||||
if not matches_hostname(entry_dict, profile['hostname']):
|
||||
reasons.append('TargetHostnames')
|
||||
if not matches_machinenumber(entry_dict, profile['machinenumber']):
|
||||
reasons.append('TargetMachineNumbers')
|
||||
if not matches_cmmversion(entry_dict, profile['cmmversion']):
|
||||
reasons.append('_CmmVersion')
|
||||
filtered.append({'name': entry.name, 'filteredby': reasons})
|
||||
return success_response({'profile': profile, 'applied': applied,
|
||||
'filtered': filtered})
|
||||
|
||||
|
||||
# -- publish lifecycle (geenforce.publish) ------------------------------------
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/publish', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.publish')
|
||||
def publish_scope_route(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
from flask_jwt_extended import get_jwt_identity
|
||||
notes = (request.get_json(silent=True) or {}).get('notes')
|
||||
try:
|
||||
publishedby = int(get_jwt_identity())
|
||||
except (TypeError, ValueError):
|
||||
publishedby = None
|
||||
version = service.publish_scope(scope.scopename, scope.phase,
|
||||
notes=notes, publishedby=publishedby)
|
||||
db.session.commit()
|
||||
return success_response({'versionnumber': version}, http_code=201)
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/versions', methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def list_versions(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
versions = ManifestPublishedVersion.query.filter_by(
|
||||
scopeid=scopeid).order_by(
|
||||
ManifestPublishedVersion.versionnumber.desc()).all()
|
||||
return success_response([{
|
||||
'versionnumber': v.versionnumber,
|
||||
'iscurrent': v.iscurrent,
|
||||
'publishedat': v.publishedat.isoformat() + 'Z' if v.publishedat else None,
|
||||
'publishedby': v.publishedby,
|
||||
'notes': v.notes,
|
||||
} for v in versions])
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/versions/<int:versionnumber>',
|
||||
methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.manage')
|
||||
def get_version(scopeid, versionnumber):
|
||||
version = ManifestPublishedVersion.query.filter_by(
|
||||
scopeid=scopeid, versionnumber=versionnumber).first()
|
||||
if not version:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such version',
|
||||
http_code=404)
|
||||
import json as _json
|
||||
return success_response({'versionnumber': version.versionnumber,
|
||||
'manifest': _json.loads(version.manifestjson)})
|
||||
|
||||
|
||||
@geenforce_bp.route('/scopes/<int:scopeid>/rollback', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_permission('geenforce.publish')
|
||||
def rollback_scope_route(scopeid):
|
||||
scope = db.session.get(ManifestScope, scopeid)
|
||||
if not scope:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'No such scope', http_code=404)
|
||||
versionnumber = (request.get_json(silent=True) or {}).get('versionnumber')
|
||||
if versionnumber is None:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'versionnumber is required', http_code=400)
|
||||
try:
|
||||
service.rollback_scope(scope.scopename, scope.phase, int(versionnumber))
|
||||
db.session.commit()
|
||||
except ValueError as exc:
|
||||
db.session.rollback()
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400)
|
||||
return success_response({'versionnumber': int(versionnumber)})
|
||||
|
||||
|
||||
def _current_published_version(scopename, phase):
|
||||
scope = ManifestScope.query.filter_by(
|
||||
scopename=scopename, phase=phase).first()
|
||||
|
||||
@@ -24,33 +24,40 @@ _KEY_TO_ATTR = {mk: attr for attr, mk in _SCALAR_FIELDS}
|
||||
_FLAG_TO_ATTR = {mk: attr for attr, mk in _BOOL_FLAGS}
|
||||
|
||||
|
||||
def build_entry(entry_dict, sortorder):
|
||||
"""Build a ManifestEntry (+ children) from one Applications[] entry."""
|
||||
entry = ManifestEntry(sortorder=sortorder,
|
||||
name=entry_dict.get('Name'),
|
||||
entrytype=entry_dict.get('Type'))
|
||||
if '_comment' in entry_dict:
|
||||
entry.comment = entry_dict['_comment']
|
||||
# Scalar model attributes that a populate resets (so an update clears fields the
|
||||
# new payload omits). name/entrytype are required and set explicitly.
|
||||
_RESETTABLE_ATTRS = [attr for _, attr in _SCALAR_FIELDS
|
||||
if attr not in ('name', 'entrytype')] + ['comment', 'regvalue']
|
||||
|
||||
|
||||
def populate_entry(entry, entry_dict):
|
||||
"""Set every field + child on an existing (or new) ManifestEntry from one
|
||||
Applications[] entry dict. Existing children are replaced. Returns entry."""
|
||||
entry.name = entry_dict.get('Name')
|
||||
entry.entrytype = entry_dict.get('Type')
|
||||
entry.comment = entry_dict.get('_comment')
|
||||
for attr in _RESETTABLE_ATTRS:
|
||||
if attr in ('comment', 'regvalue'):
|
||||
continue
|
||||
setattr(entry, attr, None)
|
||||
for key, attr in _KEY_TO_ATTR.items():
|
||||
if attr in ('name', 'entrytype'):
|
||||
continue
|
||||
if key in entry_dict and entry_dict[key] is not None:
|
||||
setattr(entry, attr, entry_dict[key])
|
||||
# RegValue stored as its raw JSON literal so DWord vs string typing survives.
|
||||
if 'RegValue' in entry_dict:
|
||||
entry.regvalue = json.dumps(entry_dict['RegValue'])
|
||||
entry.regvalue = (json.dumps(entry_dict['RegValue'])
|
||||
if 'RegValue' in entry_dict else None)
|
||||
for key, attr in _FLAG_TO_ATTR.items():
|
||||
if entry_dict.get(key):
|
||||
setattr(entry, attr, True)
|
||||
# Multi-value filters -> child rows (preserve order).
|
||||
for i, value in enumerate(entry_dict.get('PCTypes') or []):
|
||||
entry.pctypes.append(ManifestEntryPcType(sortorder=i, pctypevalue=value))
|
||||
for i, value in enumerate(entry_dict.get('TargetHostnames') or []):
|
||||
entry.hostnames.append(
|
||||
ManifestEntryHostname(sortorder=i, hostnamepattern=value))
|
||||
for i, value in enumerate(entry_dict.get('TargetMachineNumbers') or []):
|
||||
entry.machinenumbers.append(
|
||||
ManifestEntryMachineNumber(sortorder=i, machinenumber=str(value)))
|
||||
setattr(entry, attr, bool(entry_dict.get(key)))
|
||||
# Multi-value filters -> child rows (replace, preserve order).
|
||||
entry.pctypes = [ManifestEntryPcType(sortorder=i, pctypevalue=value)
|
||||
for i, value in enumerate(entry_dict.get('PCTypes') or [])]
|
||||
entry.hostnames = [ManifestEntryHostname(sortorder=i, hostnamepattern=value)
|
||||
for i, value in enumerate(entry_dict.get('TargetHostnames') or [])]
|
||||
entry.machinenumbers = [
|
||||
ManifestEntryMachineNumber(sortorder=i, machinenumber=str(value))
|
||||
for i, value in enumerate(entry_dict.get('TargetMachineNumbers') or [])]
|
||||
# InUseCheck object + Processes[].
|
||||
inuse = entry_dict.get('InUseCheck')
|
||||
if inuse:
|
||||
@@ -62,9 +69,17 @@ def build_entry(entry_dict, sortorder):
|
||||
exepath=proc.get('ExePath'),
|
||||
gracefulclosetimeoutsec=proc.get('GracefulCloseTimeoutSec')))
|
||||
entry.inusecheck = check
|
||||
else:
|
||||
entry.inusecheck = None
|
||||
return entry
|
||||
|
||||
|
||||
def build_entry(entry_dict, sortorder):
|
||||
"""Build a new ManifestEntry (+ children) from one Applications[] entry."""
|
||||
entry = ManifestEntry(sortorder=sortorder)
|
||||
return populate_entry(entry, entry_dict)
|
||||
|
||||
|
||||
def build_scope(scopename, phase, manifest_dict, iscommon=False):
|
||||
"""Build a ManifestScope (+ entries) from a parsed manifest dict."""
|
||||
scope = ManifestScope(
|
||||
|
||||
170
tests/test_plugins/test_geenforce_crud.py
Normal file
170
tests/test_plugins/test_geenforce_crud.py
Normal file
@@ -0,0 +1,170 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user