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

"""Verify a frozen detail-evidence candidate, including a disposable publisher exercise."""
import argparse
import ast
import hashlib
import json
from pathlib import Path
import sqlite3
import subprocess
import tempfile

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


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


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--package', required=True, type=Path)
    p.add_argument('--out', required=True, type=Path)
    args = p.parse_args()
    package = args.package.resolve()
    release = json.loads((package / 'RELEASE.json').read_bytes())
    assert release['overallComplete'] is False and release['productionDeployed'] is False
    assert release['privateDetailOriginalsIncluded'] is False
    assert not list(package.rglob('detail-originals'))
    manifest = json.loads((package / 'deploy/industry-ui-files.json').read_bytes())
    assert len(manifest['files']) == 10
    for entry in manifest['files']:
        assert sha((package / 'payload' / entry['path']).read_bytes()) == entry['sha256']
    syntax_count = 0
    for path in package.rglob('*.py'):
        ast.parse(path.read_text('utf-8-sig'), feature_version=(3, 9))
        syntax_count += 1
    assets = list((package / 'data/monitoring-registry/assets').iterdir())
    for path in assets:
        assert sha(path.read_bytes()) == path.stem
    with sqlite3.connect(package / 'data/monitoring-registry/monitoring.sqlite') as c:
        assert c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
        detail_rows = c.execute('SELECT product_id,projection_sha FROM detail_evidence').fetchall()
    assert len(detail_rows) == 4
    for product, digest in detail_rows:
        projection = json.loads((package / 'data/monitoring-registry/assets' / (digest + '.json')).read_bytes())
        assert projection['productId'] == product
        assert set(projection).issubset({'schema', 'generation', 'productId', 'title', 'platformFamily', 'marketplace',
            'marketplaceProof', 'returnedShopId', 'shopIdentityPolicy', 'requestId', 'endpointId', 'observedAtRaw',
            'timezoneKnown', 'rawSha256', 'proofPointers', 'fields', 'state', 'rawVerifiedAtImport', 'orderBasis', 'policy'})
    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 {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 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,3);
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}));
'''
    # Keep this temporary test fixture for audit; never point it at a live site.
    temporary = Path(tempfile.mkdtemp(prefix='r7-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-r6/package'), str(temporary), str(baseline)],
                               check=True, capture_output=True, text=True, encoding='utf-8')
    publisher = json.loads(completed.stdout)
    result = {'ok': True, 'uiFilesVerified': 10, 'python39SyntaxFiles': syntax_count,
              'nativePython39Verified': False, 'auxiliaryIntegrity': 'ok', 'assetsVerified': len(assets),
              'detailObservations': 4, 'privateDetailOriginalsIncluded': False,
              'publisher': publisher, 'paidCalls': 0, 'productionWrites': 0, 'productionDeployed': False}
    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()