#!/usr/bin/env python3 """Offline full DTO, original red, exact native binds, and retained row audit.""" import copy import hashlib import json import re from pathlib import Path ROOT=Path(__file__).parent PRIVATE=ROOT/'private-evidence' WIDE=ROOT.parent/'store-gift-wide-http-20261011/private-evidence' def load(path):return json.loads(path.read_text()) def equal(a,b): assert type(a)is type(b),(type(a),type(b)) if isinstance(a,dict): assert a.keys()==b.keys() for key in a:equal(a[key],b[key]) elif isinstance(a,list): assert len(a)==len(b) for x,y in zip(a,b):equal(x,y) else:assert a==b,(a,b) def check_wire(wire,result,baseline,lane,phase,label): assert wire['HTTP']==result['HTTP']==200 assert wire['body']==result['body'] and hashlib.sha256(wire['body'].encode()).hexdigest()==wire['body_sha256']==result['body_sha256'] params=wire['params'];page=int(params['page']);size=int(params['page_size']) excluded='exclude' in label;source_red=lane=='B'and phase=='red'and not excluded expected=copy.deepcopy(baseline);data=expected['data'] if excluded or source_red:data.update(total=0,per_page=size,current_page=1,last_page=0,data=[]) else:data.update(total=5,per_page=size,current_page=page,last_page=(5+size-1)//size,data=baseline['data']['data'][(page-1)*size:page*size]) body=json.loads(wire['body']);equal(body,expected) for row in body['data']['data']: assert len(row)==15 and 'gift_biz_order_id_exact'not in row assert type(row['biz_order_id'])is(int if row['biz_order_no'][-1]!='4'else str) assert type(row['create_time'])is str and len(row['create_time'])==19 queries=[r for r in wire['prepared']if re.match(r'^\s*SELECT\b',r['queryString'],re.I)and'user_gift_log'in r['queryString'].lower()] assert len(queries)==4 kinds=[] for entry in wire['prepared']: assert entry['ok'] and (re.match(r'^\s*(SELECT|SHOW|EXPLAIN)\b',entry['queryString'],re.I)or entry['queryString']=='CALL SP_TABLEDEF(?, ?)') for query in queries: sql=query['queryString'];assert' BETWEEN 'in sql fields=query['bindings'];begin=next(v for k,v in fields.items()if'create_time_between_1'in k);end=next(v for k,v in fields.items()if'create_time_between_2'in k) if phase=='green'and lane=='B': expected_bounds=(978192000,978451199)if'include'in label else(978278400,978364799) assert begin=={'value':expected_bounds[0],'type':1}and end=={'value':expected_bounds[1],'type':1} else: native_type=1 if lane=='B' else 2 assert begin=={'value':params['start_date']+' 00:00:00','type':native_type}and end=={'value':params['end_date']+' 23:59:59','type':native_type} assert any(v['value']=='%CDXGW11B6A91C%'for v in fields.values())and any(v['value']==10001 for v in fields.values()) if'COUNT(*)'in sql:kinds.append('count') elif'SUM(money)'in sql:kinds.append('income'if' > 'in sql else'expense') else:kinds.append('rows');assert('gift_biz_order_id_exact'in sql)==(lane=='C') assert sorted(kinds)==sorted(['count','income','expense','rows']) if phase=='green':assert str(wire['actual_ai_connection_type']).lstrip('\\')==('mysql'if lane=='B'else'app\\cores\\database\\Dm') return {'domain_date_pass':not source_red,'source_old_bug':source_red,'rows_query_executed':True,'native_queries':len(wire['prepared']),'full_DTO_exact':True} def verify(): summary={'passed':True,'HTTP_total':13,'successful_HTTP':12,'bootstrap_tool_HTTP500':1,'DML':0,'DDL':0,'original_MySQL_date_failures':3,'cases':{},'readonly_total':{}} for lane in ('B','C'): baseline=load(WIDE/('body-B-red-account-positive.json'if lane=='B'else'body-C-green-account-positive.json')) fresh=load(PRIVATE/(lane+'-fresh.json')) for phase in ('post-http','post-green'): after=load(PRIVATE/(lane+'-'+phase+'.json'));equal(after['rows'],fresh['rows']);equal(after['owner_hashes'],fresh['owner_hashes']);equal(after['rows_after'],fresh['rows']) probes=fresh['date_probes'];assert[len(probes[key]['original_string_binding'])for key in('include','exclude','single_day')]==([0,0,0]if lane=='B'else[5,0,5]) if lane=='B':assert[len(probes[key]['int_column_unix_binding'])for key in('include','exclude','single_day')]==[5,0,5] native=0 phases={'red':['include-p1-bootstrap-recovery','include-p2','single-day','exclude-day']if lane=='B'else['include-p1','include-p2','single-day','exclude-day'],'green':['green-include-p1','green-include-p2','green-single-day']if lane=='B'else['green-single-day']} for phase,labels in phases.items(): for label in labels: directory=PRIVATE/(lane+'-'+phase);wire=load(directory/(label+'.wire.json'));result=load(directory/(label+'.result.json')) summary['cases'][lane+'-'+label]=check_wire(wire,result,baseline,lane,phase,label);native+=len(wire['prepared']) total=native+sum(len(load(PRIVATE/(lane+'-'+phase+'.json'))['journal'])for phase in('fresh','post-http','post-green'))+2 assert total==(134 if lane=='B' else 147)and total<=150 summary['readonly_total'][lane]=total c_red=load(PRIVATE/'C-red/single-day.wire.json');c_green=load(PRIVATE/'C-green/green-single-day.wire.json');assert c_red['body']==c_green['body']and c_red['body_sha256']==c_green['body_sha256'] proof=load(PRIVATE/'B-red/include-p1.bootstrap-failure.redacted.json');receipt=load(PRIVATE/'B-red/bootstrap-secret-removal.json') assert proof['HTTP']==500 and proof['SQL']==proof['DML']==0 and proof['entire_debug_HTML_redacted']and proof['original_bytes_reconstructible']is False assert receipt['deleted']and receipt['recoverable']is False and proof['stable_derivative_secret_matches']==0 assert not(PRIVATE/'B-red/include-p1.result.json').exists() assert load(PRIVATE/'tdd-original.json')['failures']==6 and load(PRIVATE/'tdd-candidate.json')['assertions']==270 and load(PRIVATE/'tdd-candidate.json')['passed'] assert load(PRIVATE/'gift85-regression.json')['assertions']==85 and load(PRIVATE/'gift85-regression.json')['passed'] summary['C_single_day_complete_body_same_sha256']=c_red['body_sha256'];summary['secret_original_response_deleted_irrecoverable']=True return summary if __name__=='__main__':print(json.dumps(verify(),ensure_ascii=False,indent=2))