"""Offline evidence replay; never issues HTTP, mutates DB or reruns fixtures.""" import base64 import hashlib import json import re from pathlib import Path import verify ROOT=Path(__file__).parent def sha(data): return hashlib.sha256(data).hexdigest() def strict(left,right): return not verify.diff(left,right) def all_records(): output=[] for lane in ['B','C']: directory=ROOT/'private'/f'{lane}-original' for phase in (['original','retest','final'] if lane=='B' else ['original','final']): suffix='' if phase=='original' else '-'+phase attempts=json.loads((directory/f'http-attempts{suffix}.json').read_text()) assert len(attempts)==(15 if phase=='original' else 4 if phase=='retest' else 2) for attempt in attempts: raw=base64.b64decode(attempt['body_base64'],validate=True) assert attempt['sent'] and attempt['complete'] and sha(raw)==attempt['body_sha256'] wire=json.loads((directory/f"wire-{attempt['label']}{suffix}.json").read_text()) assert wire['label']==attempt['label'] and wire['path']==attempt['path'] assert wire['body_sha256']==sha(wire['body'].encode()) assert re.match(r'HTTP/1\.\d (\d{3}) ',attempt['headers'][0]) assert int(attempt['headers'][0].split(' ')[1])==wire['status'] and wire['source_unchanged'] assert strict(wire['source_before'],wire['source_after']) if wire['body']: assert wire['body'].encode()==raw # Preserve shutdown-buffer gaps; the original transport body is independently saved. combined={**wire,'body':raw.decode('utf-8')} failures=verify.check_wire(combined) if lane=='B' and phase=='original' and attempt['label'] in ['list-positive','list-page2','list-status40','detail-positive']: assert len(failures)==4 and wire['status']==500 and 'extend/vendor/autoload.php' in json.loads(combined['body'])['message'] classification='tool-dependency-failure-preserved' elif lane=='C' and phase=='original' and attempt['label'] in ['list-positive','detail-positive']: assert len(failures)==10 classification='DM-response-contract-failure-preserved' else: assert not failures,(lane,phase,attempt['label'],failures) classification='nonDB-shared-unreachable' if wire['kind']=='unreachable' else 'tested-scenario-contract-pass' for entry in wire['native']: sql=entry['queryString'] assert sql.upper().startswith(('SELECT','SHOW','CALL SP_TABLEDEF')) assert not re.search(r';|\b(?:INSERT|UPDATE|DELETE|MERGE|DROP|ALTER|CREATE|TRUNCATE|OUTFILE)\b',sql,re.I) if entry['layer']=='actual PDO.prepare': assert entry['prepared'] if entry['layer']=='actual PDOStatement.execute': assert entry['success'] if sql.startswith('CALL '): assert entry['bound']['1']['value']=='ZHCT_APP' and entry['bound']['1']['type']==2 assert entry['bound']['2']['value'].lower() in ['ydy_config','ydy_meal_order_supplement','ydy_user','ydy_restaurants','ydy_equipment','ydy_meal_order'] and entry['bound']['2']['type']==2 output.append({'lane':lane,'phase':phase,'label':attempt['label'],'path':attempt['path'],'classification':classification,'shutdown_body_capture_gap':not bool(wire['body']),'native_prepare':sum(e['layer']=='actual PDO.prepare' for e in wire['native']),'native_execute':sum(e['layer']=='actual PDOStatement.execute' for e in wire['native']),'body_sha256':sha(raw),'errors':failures,'raw':raw,'body':json.loads(raw),'wire':wire}) return output def verified(): records=all_records();assert len(records)==38 for lane,pk in [('B',17),('C',4)]: directory=ROOT/'private'/f'{lane}-original' journal=json.loads((directory/'fixture-journal.json').read_text()) assert journal['committed'] and journal['insert_sent'] and journal['rowCount']==1 and journal['original']['id']==pk assert journal['approved_plan_sha256']=='44a590f033b8017933aa590df7631da47707e97192bbc6ad338f2a6ca5392065' assert strict(json.loads((directory/'batch-before.json').read_text()),json.loads((directory/'batch-after.json').read_text())) assert strict(json.loads((directory/'batch-before.json').read_text()),json.loads((directory/'batch-after-completed.json').read_text())) # Exact old MySQL positive bytes; no DTO numeric/time/value coercion. controls=[] for label in ['list-positive','detail-positive']: old=next(x for x in records if x['lane']=='B' and x['phase']=='retest' and x['label']==label) new=next(x for x in records if x['lane']=='B' and x['phase']=='final' and x['label']==label) assert old['raw']==new['raw'] controls.append({'label':label,'old_and_new_sha256':old['body_sha256'],'byte_identical':True}) assert new['wire']['reflection']['app\\p\\logic\\SupplementMealOrder']['sha256']=='47d088cb9ea11f19e63132f19c294acab1cc3679b84b83472bc7c9ff8aca5689' c=next(x for x in records if x['lane']=='C' and x['phase']=='final' and x['label']==label) assert c['wire']['reflection']['app\\p\\logic\\SupplementMealOrder']['sha256']=='47d088cb9ea11f19e63132f19c294acab1cc3679b84b83472bc7c9ff8aca5689' original=json.loads((ROOT/'source-lock.json').read_text());final=json.loads((ROOT/'source-lock-final.json').read_text()) assert len(original)==len(final)==957 assert [p for p in original if original[p]!=final[p]]==['p/logic/SupplementMealOrder.php'] public=[{k:v for k,v in row.items() if k not in ['raw','body','wire']} for row in records] return {'passed':True,'actual_http':38,'transport_bodies':38,'B_actual_http':21,'C_actual_http':17,'positive_new_method_paths':['POST /p/supplementMealOrder/list','POST /p/supplementMealOrder/detail'],'nonDB_unreachable_methods':10,'old_mysql_body_controls':controls,'prepared_http':sum(x['native_prepare'] for x in records),'executed_http':sum(x['native_execute'] for x in records),'source_php_pins':957,'only_changed_source':['p/logic/SupplementMealOrder.php'],'cases':public} if __name__=='__main__': result=verified() (ROOT/'verification.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n') print(json.dumps({k:v for k,v in result.items() if k!='cases'},ensure_ascii=False))