#!/usr/bin/env python3 """One-time root integration; preserved private wire, source and SQL verification.""" import hashlib import json from pathlib import Path import re import subprocess import sys import integrate_20261010 as tool SYL = tool.STORE + 'parallel-store-syllabus-evaluate-readonly-20261010/' CAT = tool.STORE + 'parallel-store-device-catalog-read6-20261010/' SIGN = tool.AI + 'parallel-ai-market-sign-config-20261010/' def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def native_equal(a, b): if type(a) is not type(b): return False if isinstance(a, dict): return a.keys() == b.keys() and all(native_equal(a[k], b[k]) for k in a) if isinstance(a, list): return len(a) == len(b) and all(native_equal(x, y) for x, y in zip(a, b)) return a == b def manifest(base, values): root = tool.ROOT / base for name, digest in values.items(): p = root / name assert p.is_file() and not p.is_symlink() and sha(p) == digest, name def run(base, executable, script, *args): result = subprocess.run([executable, script, *args], cwd=tool.ROOT/base, check=True, capture_output=True, text=True) return result.stdout def private_files(root, files): assert root.is_dir() and not root.is_symlink() and root.stat().st_mode & 0o777 == 0o700 for name, record in files.items(): p = root/name assert p.is_file() and not p.is_symlink() and p.stat().st_mode & 0o777 == 0o600 assert sha(p) == record['sha256'], name def verify_syllabus(): m = tool.load(SYL+'manifest.json') manifest(SYL, m['files_sha256']) p = Path(m['private_terminal_evidence']['path']) assert str(p) == '/private/tmp/cdx-syllabus-private.D389Qf' private_files(p, m['private_terminal_evidence']['files']) r = json.loads(run(SYL, 'node', 'compare-final-private.js', str(p))) assert r['passed'] and r['actual_HTTP'] == 15 and r['CLI_assertions'] == 53 and r['CLI_failed'] == 1 assert len(r['checks']) == 32 and all(r['checks'].values()) assert len(r['mutation_checks']) == 110 and all(r['mutation_checks'].values()) # Historical read-only source is sealed. A later DM-only add fix may extend main. assert sha(p/'SyllabusEvaluate.php') == m['logic_final_SHA256'] assert m['fixture_PDO_INSERTs'] == 6 and m['business_API_DML'] == m['DDL'] == 0 resource = tool.load(SYL+'resource-cleanup.json') assert all(resource[l]['stopped'] and resource[l]['runtime_removed'] for l in ['B','C']) assert resource['B']['port_18321_closed'] and resource['C']['port_18322_closed'] assert resource['Auth_token_config_cache_source_archive_original_CLI_removed'] def verify_catalog(): manifest(CAT, tool.load(CAT+'MANIFEST.json')) args = ['private-evidence/B', 'private-evidence/C'] r = json.loads(run(CAT, sys.executable, 'verify_catalog.py', *args)) assert r['status'] == 'pass' and r['positive_partial'] == 5 and r['limited_empty_only'] == 1 r = json.loads(run(CAT, sys.executable, 'test_verify_catalog.py', *args)) assert r['tests'] == 29 and not r['failures'] for lane in ['B','C']: p = tool.ROOT/CAT/'private-evidence'/lane assert p.stat().st_mode & 0o777 == 0o700 assert all(x.stat().st_mode & 0o777 == 0o600 for x in p.iterdir()) candidates = tool.load(CAT+'CANDIDATE_TDD.json')['source_sha256'] for relative, records in candidates.items(): assert sha(tool.ROOT.parent/'zhctproject/store'/relative) == records['candidate'] def verify_sign(): m = tool.load(SIGN+'manifest.json') manifest(SIGN, m['hashes']) r = json.loads(run(SIGN, sys.executable, 'compare.py')) assert r['status'] == 'PASS' and not r['differences'] and not r['original_projection_errors'] assert r['strict_full_body_pairs'] == 33 and r['final_full_original_projection_reads'] == 6 assert r['actual_native_business_HTTP_total'] == 110 and r['actual_native_named_assertions_total'] == 470 sql = json.loads(run(SIGN, sys.executable, 'sql_audit.py')) assert sql == {'B': {'business_DML':26,'all_exact_scope':True}, 'C': {'business_DML':40,'all_exact_scope':True}} run(SIGN, sys.executable, '-m', 'unittest', 'discover', '-p', 'test_*.py') root = tool.ROOT/SIGN for lane in ['B','C']: p = root/'private-evidence'/lane records = {x['name']:x for x in m['private_credential_free_terminal_evidence'][lane] if x['name'] in ['private-wire.jsonl','private-sql.jsonl']} private_files(p, records) wires = [json.loads(line) for line in (p/'private-wire.jsonl').read_text().splitlines()] by_scenario = {w['scenario']:w for w in wires} assert len(by_scenario) == len(wires) total = 0 for prefix in ['evidence','green-evidence','final-evidence']: evidence = tool.load(SIGN+prefix+'-'+lane+'.json') for case in evidence['cases']: wire = by_scenario[case['scenario']] # Recorder preserves lower-case dispatched URLs; source labels use Sign. assert wire['method'] == case['method'] and wire['path'].lower() == case['path'].lower() assert wire['http_status'] == case['http_status'] and hashlib.sha256(wire['raw'].encode()).hexdigest() == case['raw_sha256'] body = json.loads(wire['raw']) assert re.fullmatch('[0-9a-f]{13}', body['trace_id']) body['trace_id'] = 'VALID_DYNAMIC_TRACE' assert native_equal(body, case['body']) total += 1 assert total == (38 if lane == 'B' else 72) clean = tool.load(SIGN+'resource-cleanup-'+lane+'.json') assert clean['worker_stopped'] and clean['port_closed'] and clean['cache_files'] == 0 assert clean['token_entries_verified_absent'] == 6 and clean['copied_source_and_config_symlinks_removed'] for name, relative in [('SignActionCandidate.php','app/common/model/SignAction.php'), ('UploadFileCandidate.php','app/terminal/model/UploadFile.php'), ('StoreSignCandidate.php','app/store/service/Sign.php'), ('TerminalSignCandidate.php','app/terminal/service/Sign.php')]: assert sha(root/name) == sha(tool.ROOT.parent/'zhctproject/ai_api'/relative) assert m['own_DML'] == 70 and m['global_DML'] == m['DDL'] == 0 def item(repo, method, path, base, **extra): return dict(repo=repo,method=method,path=path,module='数据库兼容分批实测', terminal='actual-private-HTTP-dispatch-see-source-report',role='normal-auth-owned-or-source-readonly', risk='owned-write-or-read-see-report',scope='customer-local-MySQL-and-nonproduction-DM-only', evidence=base+'REPORT.md',request_type='actual-form-or-query-see-report', parity='partial-evidence-complete-sampled-wire-native-types-and-independent-original-projection', cleanup='workers-auth-config-cache-released-marked-DB-fixtures-retained-private-terminal-evidence-preserved',**extra) def batches(): syllabus=[item('store','POST','/p/syllabusEvaluate/'+a,SYL, boundary_empty_min_max='partial-evidence-nonempty-own-list-info-CLOB-date-and-empty-missing-id') for a in ['lists','info']] catalog=[] for method,path in [('POST','/api/consume/dishes'),('POST','/api/consume/dishesRecognizeImgList'), ('POST','/api/consume/orderList'),('POST','/api/consume/card'),('GET','/p/api/getFaceDataList'),('POST','/p/api/rotation')]: x=item('store',method,path,CAT,from_repo='unreviewed',from_source='apifox-2026-05-08') if path.endswith('/rotation'): x.update(dm_state='partial-evidence-empty-result-only',success_case='partial-evidence-B-nonempty-C-empty-not-positive-C-branch', dm_readback='partial-evidence-source-absent-no-DML',parity='partial-evidence-empty-shape-only-positive-DM-unverified') catalog.append(x) sign=[item('ai_api','GET' if a in ['list','detail'] else 'POST',f'/{app}/market.Sign/{a}',SIGN, boundary_empty_min_max='partial-evidence-own-positive-empty-required-auth-and-missing-id-see-report', duplicate_retry='not-tested') for app in ['store','terminal'] for a in ['list','detail','add','edit','delete']] return [(40,syllabus,verify_syllabus),(41,catalog,verify_catalog),(42,sign,verify_sign)] if __name__ == '__main__': plans=batches() for n,_,verify in plans: assert not Path(__file__).with_name(f'integration-20261011-batch{n}.json').exists(), 'Preserve original transition audit' verify() for n,items,verify in plans: tool.main(items,Path(__file__).with_name(f'integration-20261011-batch{n}.json'),verify)