kangshifu-board/scripts/verify-industry-policy-runtime.py

"""Run the frozen release in a real interpreter without changing installed runtimes."""
import argparse
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import subprocess


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--python', type=Path, required=True)
    parser.add_argument('--package', type=Path, required=True)
    parser.add_argument('--snapshot', type=Path, required=True)
    parser.add_argument('--out', type=Path, required=True)
    args = parser.parse_args()
    package, snapshot = args.package.resolve(), args.snapshot.resolve()
    generation = json.loads(snapshot.read_bytes())['generation']
    directory = args.out.resolve()
    directory.mkdir(parents=True, exist_ok=True)
    report = {'ok': False, 'checkedAt': datetime.now(timezone.utc).isoformat(),
              'generation': generation, 'productionVerified': False, 'productionWrites': 0, 'paidCalls': 0}

    def run(name, source):
        result = subprocess.run([str(args.python.resolve()), '-B', '-X', 'utf8', '-c', source],
                                cwd=package, capture_output=True, encoding='utf-8', timeout=180)
        (directory / (name + '.stdout.txt')).write_text(result.stdout, 'utf-8')
        (directory / (name + '.stderr.txt')).write_text(result.stderr, 'utf-8')
        if result.returncode:
            raise RuntimeError(name + ' failed; inspect saved stderr')
        return result.stdout

    prefix = 'import sys;sys.path.insert(0,' + repr(str(package / 'server/voc-library')) + ');'
    try:
        report['runtime'] = json.loads(run('interpreter', 'import sys,platform,sqlite3,json;print(json.dumps({"python":platform.python_version(),"platform":platform.platform(),"sqlite":sqlite3.sqlite_version,"nativePython39Verified":sys.version_info[:2]==(3,9)}))'))
        tests = prefix + '''import unittest,json
suite=unittest.defaultTestLoader.discover('server/voc-library',pattern='test_*.py')
result=unittest.TextTestRunner(verbosity=2).run(suite)
print(json.dumps({'testsRun':result.testsRun,'failures':len(result.failures),'errors':len(result.errors),'ok':result.wasSuccessful()}))
sys.exit(0 if result.wasSuccessful() else 1)
'''
        report['unitTests'] = json.loads(run('python-tests', tests))
        argv = ['verify-industry-install.py', '--snapshot', str(snapshot), '--monitoring-store',
                str(package / 'data/monitoring-registry'), '--expected-generation', generation]
        report['preflight'] = json.loads(run('preflight', prefix + 'import runpy;sys.argv=' + repr(argv) + ';runpy.run_path("scripts/verify-industry-install.py",run_name="__main__")'))
        report['policyProbe'] = json.loads(run('policy-probe', prefix + 'import json;from industry_monitor import run;r=run(' + repr({'snapshot': str(snapshot), 'monitoringStore': str(package/'data/monitoring-registry')}) + ',' + repr({'op': 'policies', 'generation': generation, 'policyQuery': {'application': 'sodium-claims', 'asOf': '2026-11-07', 'perServingMg': '2414', 'servingGrams': '96'}}) + ');a=r["applications"];assert a["calculation"]["mgPer100g"]=="2514.58";assert a["calculation"]["productVerdict"] is None;assert a["officialChecks"]["verifiedDocuments"]==9;print(json.dumps({"workflows":len(a["workflows"]),"officialMetadata":a["officialChecks"]["verifiedDocuments"],"calculation":a["calculation"]},ensure_ascii=False))'))
        report['snapshotSha256'] = hashlib.sha256(snapshot.read_bytes()).hexdigest()
        report['ok'] = report['unitTests']['ok'] and report['preflight']['ok']
    finally:
        (directory/'verification.json').write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n', 'utf-8')
    print(json.dumps(report, ensure_ascii=False))


if __name__ == '__main__':
    main()