import copy,hashlib,json,re,struct from decimal import Decimal from pathlib import Path P=Path(__file__).resolve().parent def load(p): return json.loads(p.read_text(),parse_float=Decimal) def strict(a,b,path='$'): assert type(a) is type(b),(path,type(a).__name__,type(b).__name__,a,b) if isinstance(a,dict): assert a.keys()==b.keys(),(path,'keys',a.keys(),b.keys()) for k in a: strict(a[k],b[k],path+'.'+k) elif isinstance(a,list): assert len(a)==len(b),(path,'length',len(a),len(b)) for i,(x,y) in enumerate(zip(a,b)): strict(x,y,path+f'[{i}]') else: assert a==b,(path,a,b) def dto(form,id): return {'id':id,'mac':form['mac'],'mode':'free','modeLabel':'CDX精确跳绳','stoppedAt':int(form['stoppedAt']),'dateKey':form['dateKey'],'dateLabel':'2097 marker','timeLabel':'12:34:56','count':int(form['count']),'duration':int(form['duration']),'calories':Decimal('12.25'),'fatBurnEfficiency':Decimal('.125'),'avgFreq':100,'maxFreq':120,'mostJump':90,'ropeTrips':2,'target':{'seconds':60},'data':form['data']} def parse_insert(sql,bind): if ' SET ' in sql: pairs=re.findall(r'`([a-z_]+)`\s*=\s*:([A-Za-z0-9_]+)',sql) else: match=re.fullmatch(r'INSERT INTO [`"]?yoshop_rope_skipping[`"]?\s*\(([^)]+)\)\s*VALUES\s*\(([^)]+)\)\s*',sql,re.I) assert match,sql pairs=[(k.strip(' `"').lower(),v.strip().lstrip(':')) for k,v in zip(match[1].split(','),match[2].split(','))] assert len(pairs)==21 and len({k for k,v in pairs})==21 return {k:bind[':'+v] for k,v in pairs} def verify_lane(lane,wires_override=None): base=P/'private'/('C' if lane=='C' else '') cases=load(P/('cases-retry.json' if lane=='B' else 'cases.json')) wires={n:load(base/f'wire-{n}.json') for n in cases} if wires_override is None else wires_override initial=load(base/'batch-before.json');final=load(base/('batch-after-retry.json' if lane=='B' else 'batch-after.json')) for key in initial: if key not in ['rope','rope_sentinel']: strict(initial[key],final[key],'protected.'+key) assert initial['rope']==[] and len(final['rope'])==4 physical=load(base/'final-independent.json');assert len(physical['rows'])==4 rows={r['mac'][-1]:r for r in physical['rows']};assert set(rows)==set('ABCD') source=load(P/'source-frozen-sha.json')['files'];assert len(source)==6168 assertion_count=0 forms={label:load(P/'cases.json')['ADD_'+label]['form'] for label in 'ABCD'} dtos={label:dto(forms[label],int(rows[label]['id'])) for label in 'ABCD'} assert [int(rows[l]['id']) for l in 'ABCD']==sorted(int(rows[l]['id']) for l in 'ABCD') for label,row in rows.items(): form=forms[label] expected={'id':int(row['id']),'mac':form['mac'],'mode':'free','mode_label':'CDX精确跳绳','date_key':form['dateKey'],'date_label':'2097 marker','time_label':'12:34:56','count':int(form['count']),'duration':int(form['duration']),'calories':Decimal('12.25'),'fat_burn_efficiency':Decimal('.125'),'avg_freq':100,'max_freq':120,'most_jump':90,'rope_trips':2,'target_seconds':60,'data':json.dumps(form['data'],ensure_ascii=False,separators=(',',':')),'user_id':46914 if lane=='B' else 1928,'store_id':2147001111,'is_delete':0,'create_time':int(row['create_time']),'update_time':int(row['update_time']),'exact_stopped_at':form['stoppedAt']} assert set(row)==set(expected)|{'stopped_at'},('physical_full_columns',set(row)) for key,val in expected.items(): if key in ['calories','fat_burn_efficiency']: assert Decimal(str(row[key]))==val elif isinstance(val,int): assert int(row[key])==val else: strict(row[key],val,'physical.'+label+'.'+key) assert row['create_time']==row['update_time'] if lane=='B': assert int(row['stopped_at'])==int(form['stoppedAt']) assertion_count+=24 assert len(physical['native_types'])==4 for native in physical['native_types']: label=native['mac'][-1] assert native['exact_stopped_at']==forms[label]['stoppedAt'] if lane=='C': assert native['native_type']=='double' assert native['native_double_big_endian_hex']==struct.pack('>d',float(int(forms[label]['stoppedAt']))).hex() else: assert native['native_type']=='integer' and native['native_value']==int(forms[label]['stoppedAt']) for name,wire in wires.items(): case=cases[name];strict(wire['case'],case,'frozen_case.'+name) assert wire['http_status']==200,(name,'HTTP actual',wire['http_status']) raw=wire['raw'];response=json.loads(raw,parse_float=Decimal) assert set(response)=={'status','message','data','trace_id'} assert re.fullmatch('[0-9a-f]{13}',response['trace_id']) expected={'status':200,'message':'success','data':[],'trace_id':response['trace_id']} if name.startswith('ADD_'): expected['message']='跳绳数据上传成功' assert wire['actual_formal_inserts']==1 before=wire['before']['rope'];after=wire['after']['rope'];assert len(after)==len(before)+1 and after[:-1]==before assert after[-1]['mac']==case['form']['mac'] and after[-1]['exact_stopped_at']==case['form']['stoppedAt'] else: assert wire['actual_formal_inserts']==0 strict(wire['before']['rope'],wire['after']['rope'],'no_DML.'+name) if name.startswith('LATEST_'): expected['data']=dtos['A' if name[-1] in 'AB' else 'C'] elif name.startswith('LIST_') or name=='EMPTY_LIST': is_empty=name=='EMPTY_LIST';size=50 if name in ['LIST_ALL','EMPTY_LIST'] else 1;page=2 if name in ['LIST_P2','LIST_P3'] else 1 data=[] if not is_empty and name!='LIST_P3': labels=[['C','A'],['D','B']] if name=='LIST_ALL' else ([['C','A']] if name=='LIST_P1' else [['D','B']]) data=[{'dateKey':dtos[group[0]]['dateKey'],'count':sum(dtos[l]['count'] for l in group),'duration':sum(dtos[l]['duration'] for l in group),'list':[dtos[l] for l in group]} for group in labels] expected['data']={'list':{'total':0 if is_empty else 2,'per_page':size,'current_page':page,'last_page':0 if is_empty else (1 if size==50 else 2),'data':data}} elif name=='CROSS_TOKEN': expected.update(status=401,message='很抱歉，用户信息不存在或已删除') elif name=='MISSING_TOKEN': expected.update(status=401,message='缺少必要的参数token, 请先登录') elif name.startswith('INVALID_'): expected.update(status=500,message='结束时间不能为空' if name=='INVALID_MISSING' else '结束时间格式不正确') strict(response,expected,'complete_DTO.'+name) literals=re.findall(r'"stoppedAt":([^,}\s]+)',raw) def stopped(value): if isinstance(value,dict): return ([str(value['stoppedAt'])] if 'stoppedAt' in value else [])+[x for v in value.values() for x in stopped(v)] if isinstance(value,list): return [x for v in value for x in stopped(v)] return [] assert literals==stopped(expected),(name,'exact_JSON_integer_literals',literals,stopped(expected)) for included,sha in wire['included_source_sha256'].items(): assert source[included]==sha for key in ['users','staff','store']: strict(wire['before'][key],wire['after'][key],'protected_per_case.'+name+'.'+key) assertion_count+=len(response)+len(wire['included_source_sha256'])+len(literals)+5 events=[json.loads(line,parse_float=Decimal) for line in (base/'pdo-native.jsonl').read_text().splitlines()] executes=[e for e in events if e['layer']=='actual PDOStatement.execute'] writes=[e for e in executes if e['operation']=='INSERT'];assert len(writes)==4 for write in writes: assert write['success'] and write['rowCount']==1 fields=parse_insert(write['queryString'],write['bindValue']);form=cases[write['scenario']]['form'] assert fields['stopped_at']['value']==int(form['stoppedAt']) and fields['stopped_at']['php_type']=='integer' assert fields['stopped_at']['pdo_type']==(1 if lane=='B' else 2) assert fields['user_id']['value']==(46914 if lane=='B' else 1928) and fields['store_id']['value']==2147001111 assert fields['mac']['value']==form['mac'] and fields['date_key']['value']==form['dateKey'] assert 'id' not in fields and fields['data']['value']==json.dumps(form['data'],ensure_ascii=False,separators=(',',':')) for event in executes: assert event['operation'] in ['SELECT','INSERT','metadata_SHOW','metadata_CALL'] and event['success'] if event['operation']=='metadata_CALL': b=list(event['bindValue'].values());assert len(b)==2 and all(x['pdo_type']==2 and x['php_type']=='string' for x in b) assert (b[0]['value']=='AIZHCT_APP' and b[1]['value'].startswith('YOSHOP_')) or (b[0]['value']=='ZHCT_APP' and b[1]['value']=='YDY_USER') if lane=='B': failed=load(base/'wire-ADD_A.json');assert failed['actual_formal_inserts']==0 and failed['before']['rope']==failed['after']['rope']==[] assert json.loads(failed['raw'])['message']=='SQL outside exact frozen scope' assert not any(e['operation']=='INSERT' for e in executes if e['scenario']=='ADD_A') before_s=initial['rope_sentinel'][0];after_s=final['rope_sentinel'][0] assert int(after_s['n'])-int(before_s['n'])==4 assert Decimal(str(after_s['sum_id'] or 0))-Decimal(str(before_s['sum_id'] or 0))==sum(int(r['id']) for r in rows.values()) preflight=load(base/('preflight-'+lane+'.json')) for table,metadata in physical['metadata'].items(): original=preflight['schema'][table] for old,new in [('columns','columns'),('foreign_keys','FK'),('triggers','triggers'),('indexes','indexes')]: strict(original[old],metadata[new],'metadata_unchanged.'+table+'.'+old) return {'lane':lane,'complete_DTO_cases':len(cases),'actual_HTTP':21 if lane=='B' else 20,'formal_INSERT':len(writes),'actual_native_execute':len(executes),'actual_native_prepare':sum(e['layer']=='actual PDO.prepare' for e in events),'assertions_lower_bound':assertion_count,'whole_source_files':len(source),'physical_rows_exact':4,'existing_projections_unchanged':True,'constraints_indexes_unchanged':True,'native_stopped_at_bind_type':1 if lane=='B' else 2,'ids':{label:int(row['id']) for label,row in rows.items()}} def all_verify(): out=[verify_lane(l) for l in ['B','C']] source=load(P/'source-frozen-sha.json')['files'];main=P.parents[3]/'zhctproject/ai_api' drift=[name for name,sha in source.items() if hashlib.sha256((main/name).read_bytes()).hexdigest()!=sha] assert not drift,('Current_root_source_drift',drift) return {'lanes':out,'main_whole_source_drift':drift,'business_changes_by_slice':0,'new_interface_keys':0} if __name__=='__main__': result=all_verify();(P/'verification.json').write_text(json.dumps(result,indent=2)+'\n');print(json.dumps(result))