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

"""Verify unified r9 contents and exercise deployment only against a disposable r8 site."""
import argparse
import ast
from contextlib import closing
import hashlib
import json
from pathlib import Path
import sqlite3
import subprocess
import tempfile

ROOT = Path(__file__).resolve().parents[1]


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--package', type=Path, required=True)
    parser.add_argument('--out', type=Path, required=True)
    args = parser.parse_args()
    package = args.package.resolve()
    release = json.loads((package / 'RELEASE.json').read_bytes())
    assert release['release'] == 'ksf-industry-monitor-20261107-r9'
    assert not release['overallComplete'] and not release['productionDeployed']
    assert release['innovationWorkspaceImplemented'] and release['requestGuardImplemented']
    assert release['existingVocCoreModulesReplaced'] == 0
    assert not release['credentialsIncluded'] and not release['privateDetailOriginalsIncluded']
    assert not list(package.rglob('detail-originals'))
    assert not list(package.rglob('snapshot.json'))
    manifest = json.loads((package / 'deploy/industry-ui-files.json').read_bytes())
    assert len(manifest['files']) == 14 and len(release['backendFiles']) == 15
    assert set(e['path'] for e in manifest['files']) == set(release['staticFiles'])
    for entry in manifest['files']:
        assert sha((package / 'payload' / entry['path']).read_bytes().replace(b'\r\n', b'\n')) == entry['sha256']
    for name in release['backendFiles']:
        assert (package / name).is_file(), name
    syntax = list(package.rglob('*.py'))
    for path in syntax:
        ast.parse(path.read_text('utf-8-sig'), feature_version=(3, 9))
    assets = list((package / 'data/monitoring-registry/assets').iterdir())
    for path in assets:
        assert sha(path.read_bytes()) == path.stem, path
    database = package / 'data/monitoring-registry/monitoring.sqlite'
    with closing(sqlite3.connect(database.resolve().as_uri() + '?mode=ro', uri=True)) as connection:
        assert connection.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
        assert connection.execute('SELECT count(*) FROM detail_evidence').fetchone()[0] == 4
        assert connection.execute('SELECT count(*) FROM detail_endpoint_proofs').fetchone()[0] == 1
        assert connection.execute('SELECT count(DISTINCT document_id) FROM policy_source_checks').fetchone()[0] == 25
        assert connection.execute("SELECT count(*) FROM pipeline_workers WHERE state<>'ARCHIVED_COPY'").fetchone()[0] == 0
    scope = json.loads((package / 'handoff/collector-scope.json').read_bytes())
    assert not scope['dispatchAllowed'] and scope['budget']['remainingDispatchMinor'] == 0
    assert scope['primaryShop']['products'] == 165 and len(scope['shops']) == 8
    assert scope['budget']['identifiedSettledRequests'] == 76
    assert scope['budget']['aggregateOnlyRows'] == 78
    baseline = ROOT / 'work/industry-monitoring-20261107-r1/release-baseline/index.html'
    assert sha(baseline.read_bytes()) == manifest['homepageSha256']
    program = r'''
import fs from 'node:fs';
import path from 'node:path';
import assert from 'node:assert/strict';
import crypto from 'node:crypto';
import {pathToFileURL} from 'node:url';
const [candidate,base,temporary,index]=process.argv.slice(1);
const {publishFiles}=await import(pathToFileURL(path.join(candidate,'scripts/publish-industry-ui.mjs')));
const target=path.join(temporary,'site');
fs.cpSync(path.join(base,'payload'),target,{recursive:true});
fs.copyFileSync(index,path.join(target,'index.html'));
const manifest=JSON.parse(fs.readFileSync(path.join(candidate,'deploy/industry-ui-files.json')));
const normalized=p=>crypto.createHash('sha256').update(fs.readFileSync(p,'utf8').replaceAll('\r\n','\n')).digest('hex');
const expected=manifest.files.filter(e=>!fs.existsSync(path.join(target,e.path)) || normalized(path.join(target,e.path))!==e.sha256).length;
const options={sourceDir:path.join(candidate,'payload'),manifestPath:path.join(candidate,'deploy/industry-ui-files.json'),targetDir:target};
const dry=publishFiles(options);assert.equal(dry.readOnly,true);assert.equal(dry.changes.length,expected);
const applied=publishFiles({...options,apply:true});assert.equal(applied.filesystemVerified,true);
const repeated=publishFiles({...options,apply:true});assert.equal(repeated.changes.length,0);
fs.appendFileSync(path.join(target,'src/pages/industry-detail-view.js'),'\n// concurrent change');
let rejected=false;try{publishFiles({...options,apply:true});}catch(e){rejected=e.code==='UNKNOWN_TARGET_CONTENT';}
assert.equal(rejected,true);
console.log(JSON.stringify({ok:true,dryChanges:dry.changes.length,applyVerified:true,idempotent:true,unknownConcurrentFileRejected:true,networkRequests:0}));
'''
    temporary = Path(tempfile.mkdtemp(prefix='r9-publisher-', dir=ROOT / 'work/industry-monitoring-20261107-r1'))
    completed = subprocess.run(['node', '--input-type=module', '-e', program, str(package),
        str(ROOT / 'artifacts/ksf-industry-monitor-20261107-r8/package'), str(temporary), str(baseline)],
        capture_output=True, text=True, encoding='utf-8')
    if completed.returncode:
        raise AssertionError(completed.stderr)
    native = json.loads((package/'evidence/r9-native39/verification.json').read_bytes())
    assert native['ok'] and native['runtime']['nativePython39Verified']
    assert native['unitTests'] == {'testsRun':92, 'failures':0, 'errors':0, 'ok':True}
    assert native['generation'] == scope['generation']
    assert native['productionVerified'] is False
    result = {'ok': True, 'staticFiles': 14, 'backendFiles': 15, 'python39SyntaxFiles': len(syntax),
        'nativePython39Verified': True, 'nativePython39Platform': native['runtime']['platform'],
        'productionNativeRuntimeVerified': False, 'assetsVerified': len(assets), 'auxiliaryIntegrity': 'ok',
        'detailObservations': 4, 'appendedEndpointProofs': 1, 'officialPolicyMetadataVerified': 9, 'privateOriginalsIncluded': False,
        'publisher': json.loads(completed.stdout), 'overallComplete': False,
        'productionDeployed': False, 'productionWrites': 0, 'paidCalls': 0}
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', 'utf-8')
    print(json.dumps(result, ensure_ascii=False))


if __name__ == '__main__':
    main()