import copy import datetime import hashlib import json import re from decimal import Decimal from pathlib import Path from compare import strict_diff,validate_body from sql_audit import all_audits P=Path(__file__).resolve().parent SMALL=['staff','nutrition_meals','nutrition_lack','nurtrition','staff_store_user','staff_face'] APPROVED={'B':'db74684022319c160b79816201e437fda0e4f8146be7220d6d8ca9f741a20931','C':'b030d2a72730d5cc9eb8b57196146a3ea0d81733549297a6845fbbe60047ce5d'} def eq(a,b,label): errors=strict_diff(a,b) assert not errors,(label,errors[:3]) def trace_only(raw): body=json.loads(raw) assert re.fullmatch('[a-f0-9]{13}',body['trace_id']) result,count=re.subn(r'"trace_id":"[a-f0-9]{13}"','"trace_id":"VALID_TRACE"',raw) assert count==1 return result def fixture_check(lane,journal,proposal): assert journal['committed'] is True and len(journal['owned'])==len(journal['statements'])==13 assert journal['approved_plan_sha256']==APPROVED[lane] assert hashlib.sha256((P/f'fixture-proposal-{lane}.json').read_bytes()).hexdigest()==APPROVED[lane] ids={} for plan,statement,owned in zip(proposal['plans'],journal['statements'],journal['owned']): assert statement['queryString']==plan['sql'] and statement['sent'] is True and statement['executed'] is True and statement['rowCount']==1 values={key:(ids[value] if isinstance(value,str) and value.startswith('@') else value) for key,value in plan['values'].items()} binds=[{'position':i+1,'value':str(value),'pdo_type':2} for i,value in enumerate(values.values())] eq(binds,statement['bindValue'],'Every exact actual INSERT binding') row=owned['original'] assert owned['reference']==plan['generated_reference'] and owned['table']==plan['table'] and owned['id']==int(row['id'])>0 assert set(row)==set(values)|{'id'} for key,value in values.items(): if type(value) is int:assert int(row[key])==value elif key in ['height','weight']:assert Decimal(str(row[key]))==Decimal(value) elif key in ['create_time','update_time']:assert re.fullmatch(re.escape(value)+r'(?:\.0+)?',row[key]) else:assert type(row[key]) is type(value) and row[key]==value ids[owned['reference']]=owned['id'] eq(ids,journal['refs'],'Actual generated reference ledger') for table,old in journal['before'].items(): own=[row for row in journal['owned'] if row['table']==table] keys={row['id'] for row in own} # Existing tables without new rows can have non-id primary key names. after=journal['after'][table] remaining=[row for row in after if not (keys and int(row['id']) in keys)] eq(old,remaining,'Every previous row preserved '+table) assert len(after)==len(old)+len(own) def map_ids(body,case,journal): body=copy.deepcopy(body);body['trace_id']='VALID_TRACE' if body['status']!=200:return body name=case['name'];module=case['context'] def owned(reference):return next(row['original'] for row in journal['owned'] if row['reference']==reference) def replace(row,key,value,label): assert type(row[key]) is int and row[key]==int(value),(name,key) row[key]=label if name=='STAFF_DETAIL' or name.startswith('STAFF_LIST'): rows=[body['data']['detail']] if name=='STAFF_DETAIL' else body['data']['list']['data'] physical=owned('@staff_'+module) for row in rows: replace(row,'id',physical['id'],'STAFF_'+module) if 'user_id' in row:replace(row,'user_id',physical['user_id'],'ACTIVE_USER_'+module) elif name=='NUTRITION_DETAIL': row=body['data']['detail']['nutrition_target'];physical=owned('@nurtrition_store') replace(row,'id',physical['id'],'NURTRITION_store') replace(row,'staff_id',physical['staff_id'],'STAFF_store') replace(row,'user_id',physical['user_id'],'OPERATOR_store') return body def verify_all(): sources={'original':json.loads((P/'source-frozen-sha.json').read_text()),'final':json.loads((P/'source-final-sha.json').read_text())} assert len(sources['original']['files'])==len(sources['final']['files'])==6168 assert {key for key in sources['original']['files'] if sources['original']['files'][key]!=sources['final']['files'][key]}==set(sources['final']['accepted_changes']) assert len(sources['final']['accepted_changes'])==2 assert hashlib.sha256((P/'case-proposal.json').read_bytes()).hexdigest()=='51c59365391ea18d43c633b654a492a4cedddb9c5d78652bfbb95f3375fa8263' wires={};fixtures={};out={'true_HTTP':60,'complete_wire':60,'original':{},'final':{},'B_old_raw_trace_only_equal':0,'strict_effective_BC_pairs':0,'new_preparation_INSERT_each':13,'HTTP_DML':0,'HTTP_DDL':0} for lane in 'BC': directory=P/'private-evidence'/lane fixture=json.loads((directory/'fixture-journal.json').read_text());fixtures[lane]=fixture fixture_check(lane,fixture,json.loads((P/f'fixture-proposal-{lane}.json').read_text())) dictionary=json.loads((directory/f'preflight-stage2-{lane}.json').read_text())['enum_dictionary'] wires[lane]={};records=[] for phase,casefile,pattern in [('original','cases.json','wire-PILOT_RR_*.json'),('final','cases-FINAL.json','wire-PILOT_RR_FINAL_*.json')]: cases=json.loads((directory/casefile).read_text());assert len(cases)==(26 if phase=='original' else 4) phase_records=[] for scenario,case in cases.items(): wire=json.loads((directory/f'wire-{scenario}.json').read_text());response=json.loads((directory/f'response-{scenario}.json').read_text()); eq(case,wire['case'],'Frozen exact case') assert response['received'] is True if 'received' in response else response['response_received'] is True assert response['method']=='POST' and response['path']==case['path'] assert response['body_sha256']==hashlib.sha256(case['body_raw'].encode()).hexdigest() assert response['raw']==wire['raw'] and (directory/f'response-{scenario}.body').read_bytes()==wire['raw'].encode() assert wire['http_status']==200 tenant=2147001111 if case['context']=='store' else 2147001112 small={key:[row for row in fixture['after'][key] if int(row['store_id'])==tenant] for key in SMALL} eq(small,wire['before'],'Independent original before each case');eq(small,wire['original'],'Independent original after each case') for file,sha in wire['source_sha256'].items():assert sources[phase]['files'][file]==sha body=json.loads(wire['raw']);errors=validate_body(case,body,wire['original'],dictionary,datetime.date(2026,10,11)) phase_records.append({'scenario':scenario,'kind':case['kind'],'business_status':body['status'],'differences':errors}) base_scenario=scenario.replace('PILOT_RR_FINAL_','PILOT_RR_') wires[lane][base_scenario]=wire out[phase][lane]=phase_records end=json.loads((directory/('final-original.json' if phase=='original' else 'final-FINAL-original.json')).read_text()) eq(fixture['after'],end['original'],'Complete42 final group scope') eq(end['enum_before'],end['enum_after'],'Complete179 natural dictionary native values/order') assert len(end['enum_before'])==179 original_red=[r for r in out['original'][lane] if r['differences']] assert len(original_red)==(0 if lane=='B' else 4) if lane=='C':assert sum(len(r['differences']) for r in original_red)==7 assert all(not row['differences'] for row in out['final'][lane]) cleanup=json.loads((P/f'resource-cleanup-{lane}.json').read_text()) assert cleanup['final42_full_rows_exact'] and cleanup['retained13_full_rows_exact'] and cleanup['final179_enum_unchanged'] and cleanup['all6168_final_source_pins_verified'] and cleanup['all_Auth_cache_files_absent'] assert all(row['absent'] for row in cleanup['own_original_final_workers'].values()) if lane=='C':assert cleanup['IMG_final']['original1775_full20_sha256']=='1fd0eef856df45a66f6c2f34c11022d9e8a65c960df641edcdf0a583ec5ff2a3' for scenario,b in wires['B'].items(): c=wires['C'][scenario] eq(map_ids(json.loads(b['raw']),b['case'],fixtures['B']),map_ids(json.loads(c['raw']),c['case'],fixtures['C']),'Strict complete effective B/C body '+scenario) out['strict_effective_BC_pairs']+=1 for file in (P/'private-evidence/B').glob('wire-PILOT_RR_FINAL_*.json'): final=json.loads(file.read_text());old=json.loads((file.parent/file.name.replace('PILOT_RR_FINAL_','PILOT_RR_')).read_text()) assert trace_only(final['raw'])==trace_only(old['raw']) out['B_old_raw_trace_only_equal']+=1 assert out['B_old_raw_trace_only_equal']==4 and out['strict_effective_BC_pairs']==26 out['native_sql']=all_audits() return out if __name__=='__main__': result=verify_all();(P/'comparison-final.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n') print(json.dumps({key:value for key,value in result.items() if key not in ['original','final','native_sql']}))