#!/usr/bin/env python3 """Offline comparison uses captured wire and independently queried physical rows.""" import copy, hashlib, json, re from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo ROOT=Path(__file__).resolve().parent META=json.loads((ROOT/'structure-B.json').read_text())['tables'] TZ=ZoneInfo('Asia/Shanghai') def equal(a,b,path='$'): if type(a) is not type(b): raise AssertionError(f'{path}: {type(a).__name__} != {type(b).__name__}') if isinstance(a,dict): if a.keys()!=b.keys(): raise AssertionError(f'{path}: keys {a.keys() ^ b.keys()}') for k in a: equal(a[k],b[k],path+'.'+k) elif isinstance(a,list): if len(a)!=len(b): raise AssertionError(f'{path}: length {len(a)} != {len(b)}') for i,(x,y) in enumerate(zip(a,b)): equal(x,y,f'{path}[{i}]') elif a!=b: raise AssertionError(f'{path}: {a!r} != {b!r}') def cast(row,table): result={} for c in META[table]['columns']: k=c['column_name'];v=row[k] if v is None: result[k]=None elif k in ('create_time','update_time'): result[k]=datetime.fromtimestamp(int(v),TZ).strftime('%Y-%m-%d %H:%M:%S') elif 'int' in c['data_type']: result[k]=int(v) else: result[k]=v return result def file_oracle(row,module): out=cast(row,'upload_file');url=f'http://{module}.comment.invalid/uploads/{row["file_path"]}' out['preview_url']=url;out['external_url']=url return out def projected_comment(row,original,module,detail): out=cast(row,'comment');cid=int(row['comment_id']);tenant=int(row['store_id']) files={int(r['file_id']):r for r in original['upload_file'] if int(r['store_id'])==tenant} if detail: imgs=[r for r in original['comment_image'] if int(r['comment_id'])==cid and int(r['store_id'])==tenant] imgs.sort(key=lambda r:int(r['id'])) values=[] for im in imgs: v=cast(im,'comment_image');v['file']=file_oracle(files[int(im['image_id'])],module);values.append(v) out['images']=values;out['imageIds']=[int(r['image_id']) for r in imgs];out['imageList']=[r['file'] for r in values] else: user=next(r for r in original['user'] if int(r['user_id'])==int(row['user_id'])) avatar=file_oracle(files[int(user['avatar_id'])],module) out['user']={'user_id':int(user['user_id']),'nick_name':user['nick_name'],'avatar_id':int(user['avatar_id']),'avatar_url':avatar['preview_url'],'avatar':avatar} og=next(r for r in original['order_goods'] if int(r['order_goods_id'])==int(row['order_goods_id'])) image=file_oracle(files[int(og['image_id'])],module) out['orderGoods']={'order_goods_id':int(og['order_goods_id']),'goods_id':int(og['goods_id']),'goods_name':og['goods_name'],'image_id':int(og['image_id']),'goods_props':json.loads(og['goods_props']),'order_id':int(og['order_id']),'goods_image':image['preview_url'],'image':image} return out def expected_data(wire,case): original=wire['original'];module=case['module'];tenant=2147001111 if module=='store' else 2147001112 rows=[r for r in original['comment'] if int(r['store_id'])==tenant] action=case['action'];params=case['params'] or {} if action in ('edit','delete'): return [] if action=='detail': row=next(r for r in rows if int(r['comment_id'])==int(params['commentId'])) return {'detail':projected_comment(row,original,module,True)} rows=[r for r in rows if int(r['is_delete'])==0] for k in ('score','status'): if k in params: rows=[r for r in rows if int(r[k])==params[k]] if 'userId' in params: rows=[r for r in rows if int(r['user_id'])==params['userId']] if 'goodsName' in params: goods={int(r['goods_id']):r for r in original['goods']} rows=[r for r in rows if any(params['goodsName'] in goods[int(r['goods_id'])][f] for f in ('goods_name','goods_no'))] if 'orderNo' in params: orders={int(r['order_id']):r for r in original['order']} rows=[r for r in rows if params['orderNo'] in orders[int(r['order_id'])]['order_no']] rows.sort(key=lambda r:(int(r['sort']),-int(r['create_time']))) total=len(rows);requested=params.get('page',1);page=max(1,min(requested,max(1,(total+14)//15))) return {'list':{'total':total,'per_page':15,'current_page':page,'last_page':(total+14)//15,'data':[projected_comment(r,original,module,False) for r in rows[(requested-1)*15:requested*15]]}} def cases_for(lane_dir): cases=json.loads((lane_dir/'run-journal.json').read_text())['cases'] result={c['scenario']:c for c in cases} fixture=json.loads((lane_dir/'fixture-journal.json').read_text()) for phase in ('CONTROL','GREEN'): for module in ('store','terminal'): user=fixture['owned'][module]['user'] for name,p in [('LIST',{}),('GOODS',{'goodsName':'CDXCM11F782'}),('ORDER',{'orderNo':'CDXCM11F782'}),('USER',{'userId':user}),('SCORE',{'score':20,'status':0}),('EMPTY_FILTER',{'goodsName':'MISSING_CDXCM11F782'}),('PAGE2',{'page':2}),('LITERAL',{'goodsName':"comment.order 中文' \\"})]: key=f'PILOT_CM_{phase}_{module.upper()}_{name}' result[key]={'module':module,'name':name,'action':'list','params':p,'expected_status':200} return result def normalize(body,original,fixture): out=copy.deepcopy(body);out['trace_id']='VALIDATED_TRACE' maps={k:{} for k in ('user_id','goods_id','order_id','order_goods_id','comment_id','file_id','image_id','avatar_id','id')} for module,own in fixture['owned'].items(): for table,key in [('user','user_id'),('goods','goods_id'),('order','order_id'),('order_goods','order_goods_id'),('upload_file','file_id')]: maps[key][own[table]]=module+'.'+table for i,cid in enumerate(own['comment']):maps['comment_id'][cid]=module+'.comment.'+str(i) maps['image_id'][own['upload_file']]=module+'.upload_file';maps['avatar_id'][own['upload_file']]=module+'.upload_file' for im in original['comment_image']: key=int(im['id']);owner='store' if int(im['store_id'])==2147001111 else 'terminal' maps['id'][key]=owner+'.owned_image_relation' def walk(v): if isinstance(v,dict): for k,x in list(v.items()): if k=='imageIds': assert type(x) is list and all(type(i) is int and i in maps['image_id'] for i in x) v[k]=[maps['image_id'][i] for i in x] elif k in maps and type(x) is int and x in maps[k]:v[k]=maps[k][x] elif k in ('create_time','update_time'): assert type(x) is str and re.fullmatch(r'\d{4}-\d\d-\d\d \d\d:\d\d:\d\d',x) v[k]='ORIGINAL_ROW_VALIDATED_TIME' else:walk(x) elif isinstance(v,list): for x in v:walk(x) walk(out);return out def verify(): reports={};normalized={};wire_map={};total=0 for lane in ('B','C'): d=ROOT/'private-evidence'/lane;fixture=json.loads((d/'fixture-journal.json').read_text());cases=cases_for(d) wires=[json.loads(l) for l in (d/'private-wire.jsonl').read_text().splitlines()] output=[] for w in wires: s=w['scenario'];c=cases[s];body=json.loads(w['raw']);total+=1 assert w['http_status']==200,s assert type(body) is dict and set(body)=={'status','message','data','trace_id'},s assert type(body['status']) is int and type(body['message']) is str and re.fullmatch('[0-9a-f]{13}',body['trace_id']),s historical_red=lane=='C' and not s.startswith('PILOT_CM_GREEN_') and c['action']=='list' and c['name']!='AUTH' if historical_red: assert body['status']==500 and '-2007' in body['message'] and '[comment]' in body['message'],s else: assert body['status']==c['expected_status'],s if body['status']==200: expected=expected_data(w,c);equal(body['data'],expected,s+'.data') message='更新成功' if c['action']=='edit' else '删除成功' if c['action']=='delete' else 'success' equal(body['message'],message,s+'.message') else:equal(body['data'],[],s+'.error_data') normalized[(lane,s)]=normalize(body,w['original'],fixture) if 'GREEN' in s: expected_sha={'app/store/model/Comment.php':'e06a302523b39e4e7b6a9240a24ce3d12444657e76ba3493d2b362c486cc153f','app/terminal/model/Comment.php':'ce38dd56d15984386473eba7596d2073c118e7afc02629915acbb618e15988ff','app/common/database/DmBuilder.php':'58e1e79ce69881cc305c6fd33adec0e9a3d030ac15b4c00ae464cbaa4cce2b59'} equal(w['source_sha256'],expected_sha,s+'.source_sha256') wire_map[(lane,s)]=body output.append({'scenario':s,'http_status':w['http_status'],'business_status':body['status'],'original_projection_equal':False if historical_red else body['status']==200,'historical_list_sql_red':historical_red,'raw_sha256':hashlib.sha256(w['raw'].encode()).hexdigest(),'elapsed_ms':w['elapsed_ms']}) reports[lane]=output pairs=[] for (lane,s),b in normalized.items(): if lane=='B' and ('C',s) in normalized: equal(b,normalized[('C',s)],s+'.BC_full_body');pairs.append(s) unchanged=[] for module in ('STORE','TERMINAL'): for name in ('LIST','GOODS','ORDER','USER','SCORE','EMPTY_FILTER','PAGE2','LITERAL'): old=f'PILOT_CM_CONTROL_{module}_{name}';new=f'PILOT_CM_GREEN_{module}_{name}' a=copy.deepcopy(wire_map[('B',old)]);b=copy.deepcopy(wire_map[('B',new)]);a['trace_id']=b['trace_id']='VALIDATED_TRACE' equal(a,b,new+'.B_original_full_body_unchanged');unchanged.append(new) return {'real_HTTP':total,'cases':reports,'BC_complete_body_pairs':pairs,'B_original_full_body_unchanged':unchanged,'historical_C_list_reds':16} if __name__=='__main__': report=verify();(ROOT/'comparison.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n');print(json.dumps({'HTTP':report['real_HTTP'],'full_BC_pairs':len(report['BC_complete_body_pairs']),'B_body_unchanged':len(report['B_original_full_body_unchanged']),'historical_C_red':16},ensure_ascii=False))