GE-Enforce polish: DDL-parity guard test, retire Collector PC Types page
All checks were successful
CI / backend (push) Successful in 1m40s
CI / naming (push) Successful in 2s
CI / frontend (push) Successful in 8s

Add test_geenforce_ddl_parity to lock the manifest models against their
Alembic baseline (catches model/migration drift for a chain that is still
amendable pre-deploy).

Retire the "Collector PC Types" settings card: GE-Enforce scope
computertypeid supersedes the pctypemap editor UI (ADR-012). The collector
still reads pctype_mapping(), so the backend map stays; only the editor
surface is removed, with a deprecation note in pctypemap.py.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-07-13 07:25:01 -04:00
parent 4e5b4228c1
commit 456a44104b
4 changed files with 151 additions and 43 deletions

View File

@@ -0,0 +1,134 @@
"""GE-Enforce DDL parity gate: models match the hand-written Alembic baseline.
The geenforce baseline (0001_geenforce_baseline.py) actually CREATES the plugin
tables (ADR-008 per-plugin chain), and it is hand-written - the models and the
migration can drift silently (a column added to a model but not the migration
ships a schema that upgrade() never builds, and vice versa). This guard builds
the schema two ways and diffs them column-by-column:
1. the SQLAlchemy models via metadata.create_all (desired shape), and
2. the migration's upgrade() run against a fresh engine (shipped shape).
Both go through the same SQLite dialect, so the reflected column types render
identically and the comparison stays low-flake. A drift (missing column, changed
type, changed nullability) fails here instead of on a real deploy insert.
"""
import importlib.util
from pathlib import Path
from sqlalchemy import create_engine, inspect
from alembic.migration import MigrationContext
from alembic.operations import Operations
_BASELINE = (Path(__file__).resolve().parent.parent.parent / 'plugins' /
'geenforce' / 'migrations' / 'versions' /
'0001_geenforce_baseline.py')
def _geenforce_models():
from plugins.geenforce.models import manifest
return [
manifest.ManifestScope,
manifest.ManifestEntry,
manifest.ManifestEntryPcType,
manifest.ManifestEntryHostname,
manifest.ManifestEntryMachineNumber,
manifest.ManifestInUseCheck,
manifest.ManifestInUseCheckProcess,
manifest.ManifestPublishedVersion,
manifest.ManifestPayload,
manifest.ManifestEnforcementReport,
manifest.ManifestEnforcementResult,
manifest.PcTypeAlias,
]
def _model_schema(db, models):
"""Build the geenforce tables from the models; return an inspector."""
engine = create_engine('sqlite://')
db.metadata.create_all(engine, tables=[model.__table__ for model in models])
return inspect(engine)
def _migration_schema():
"""Run the baseline migration's upgrade() on a fresh engine; return columns.
Returns {tablename: {columnname: reflected-column-dict}}. Reflection happens
while the connection is open, so we materialize the dicts before it closes.
"""
spec = importlib.util.spec_from_file_location('geenforce_baseline',
str(_BASELINE))
migration = importlib.util.module_from_spec(spec)
spec.loader.exec_module(migration)
engine = create_engine('sqlite://')
schema = {}
with engine.connect() as connection:
operations = Operations(MigrationContext.configure(connection))
operations._install_proxy()
try:
migration.upgrade()
finally:
operations._remove_proxy()
connection.commit()
inspector = inspect(connection)
for table in inspector.get_table_names():
schema[table] = {column['name']: column
for column in inspector.get_columns(table)}
return schema
def _colkey(column):
"""Normalize a reflected column to the facets we gate on."""
return (str(column['type']), bool(column['nullable']))
def test_models_match_migration_baseline(app, db):
"""Every geenforce model table+column matches what the migration builds."""
with app.app_context():
models = _geenforce_models()
model_inspector = _model_schema(db, models)
migration_schema = _migration_schema()
drift = []
for model in models:
table = model.__tablename__
if table not in migration_schema:
drift.append(f'{table}: table missing from migration')
continue
modelcols = {column['name']: column
for column in model_inspector.get_columns(table)}
migrationcols = migration_schema[table]
missing = set(modelcols) - set(migrationcols)
extra = set(migrationcols) - set(modelcols)
for name in sorted(missing):
drift.append(f'{table}.{name}: in model, not in migration')
for name in sorted(extra):
drift.append(f'{table}.{name}: in migration, not in model')
for name in sorted(set(modelcols) & set(migrationcols)):
if _colkey(modelcols[name]) != _colkey(migrationcols[name]):
drift.append(
f'{table}.{name}: model {_colkey(modelcols[name])} != '
f'migration {_colkey(migrationcols[name])}')
assert not drift, 'model/migration DDL drift:\n' + '\n'.join(drift)
def test_curated_and_payload_columns_present_both_sides(app, db):
"""Spot-guard the two real-work columns: manifestentries.appid and
manifestpayloads.payloadsha256 exist in BOTH the models and the migration
(they were added for the app-link + inline-payload features)."""
with app.app_context():
models = _geenforce_models()
model_inspector = _model_schema(db, models)
migration_schema = _migration_schema()
for table, column in (('manifestentries', 'appid'),
('manifestpayloads', 'payloadsha256')):
modelcols = {c['name'] for c in model_inspector.get_columns(table)}
assert column in modelcols, f'{table}.{column} missing from models'
assert column in migration_schema[table], \
f'{table}.{column} missing from migration'