#!/usr/bin/env python3 """Offline full DTO/source evidence verification. No DB/HTTP, no natural values printed.""" import copy import hashlib import json import pathlib import sys HERE=pathlib.Path(__file__).resolve().parent MAIN=['EQUIPMENT_READ','PUSH_LIST_1','PUSH_LIST_2','VERSION_CODE','VERSION_CODE_EMPTY','WEIGH_DISHES','MOBILE_RECIPE_OWNED','MOBILE_RECIPE_EMPTY'] DINING=['STAFF_DINING_ALL','STAFF_DINING_MEAL1'] CURRENT={line.split(None,1)[1]:line.split(None,1)[0] for line in (HERE/'SOURCE_SHA256.txt').read_text().splitlines()} TDD=json.loads((HERE/'CANDIDATE_TDD.json').read_bytes()) DTDD=json.loads((HERE/'DINING_TDD.json').read_bytes()) ORIGINAL=dict(CURRENT) for path,sha in TDD['source_sha256'].items(): ORIGINAL[path]=sha['original'] ORIGINAL[DTDD['source']]=DTDD['original_sha256'] MIDDLE=dict(CURRENT);MIDDLE[DTDD['source']]=DTDD['original_sha256'] def strict(a,b,path='$'): assert type(a) is type(b),path+':type' if isinstance(a,dict): assert set(a)==set(b),path+':keys' for key in a: strict(a[key],b[key],path+'.'+str(key)) elif isinstance(a,list): assert len(a)==len(b),path+':length' for i,(x,y) in enumerate(zip(a,b)): strict(x,y,path+f'[{i}]') else: assert a==b,path+':value' def read(root,file): return json.loads((pathlib.Path(root)/file).read_bytes()) def load(root): root=pathlib.Path(root) names=['report','responses','source','cleanup','dining-report','dining-responses','dining-source', 'green-report','green-responses','green-source','dining-green-report','dining-green-responses','dining-green-source', 'h7-native-verification','h7-native-before','h7-native-after','boundary-version-overflow'] return {name:read(root,name+'-private.json') for name in names if (root/(name+'-private.json')).is_file()} def stage(data,prefix,names,sha,dining=False): report=data[prefix+'report'];responses=data[prefix+'responses'];sources=data[prefix+'source'] strict(report['source_sha256'],sha) cases={x['case']:x for x in report['cases']} assert report['http_dml']==report['new_fixture_dml']==report['ddl']==0 assert report['actor_unchanged'] is True assert report['boundary']['private_source_reflection'] is True and report['boundary']['real_queue'] is False assert report['boundary']['external_stubs'] is True and report['boundary']['redis_disabled'] is True for name in names: case=cases[name];response=responses[name];source=sources[name] assert case['method']=='POST' and case['http_status']==200 and type(case['code']) is int and case['code']==0 assert case['source_pass'] is True and case['source_window_unchanged'] is True and case['differences']==[] wire=response['raw_wire'].encode() assert len(wire)==response['wire_bytes'] and hashlib.sha256(wire).hexdigest()==response['wire_sha256'] body=json.loads(wire);strict(body,response['body']) before,after=source['before'],source['after'] assert before['source_digest']==after['source_digest'] and len(before['source_digest'])==64 strict(before['source_sets'],after['source_sets']) if dining: oracle=source['group_oracle'] assert oracle['representative_selection_unresolved'] is True and case['representative_selection_unresolved'] is True strict(oracle['expected'],body) # Other definite fields must be original source values, not normalized wholesale. modified=copy.deepcopy(before['body']) for i,row in enumerate(body['data']['rows']): expected=modified['data']['rows'][i] assert row['uuid']==expected['uuid'] if isinstance(row['meal_times'],dict): menu=before['source_sets']['menus'][row['uuid']] for meal,time in row['meal_times'].items(): assert type(time) is str and len(time)==19 matching=[x for x in menu if str(x['meal_times'])==meal and x['menu_date']==row['menu_date'] and x['create_time']==time] assert matching,'selected timestamp not real source member' expected['meal_times'][meal]=time;expected['meal_msg'][meal]=time[11:16] strict(modified,body) else: strict(before['body'],body) if name=='VERSION_CODE_EMPTY': assert type(body['data']) is int and body['data']==1 return [cases[name] for name in names] def verify(b,c): bmain=stage(b,'',MAIN,ORIGINAL) bdining=stage(b,'dining-',DINING,ORIGINAL) cmain=stage(c,'green-',MAIN,MIDDLE) cdining=stage(c,'dining-green-',DINING,CURRENT,True) assert all(x['positive'] is True for x in bmain+cmain if x['case'] not in ['VERSION_CODE_EMPTY','MOBILE_RECIPE_EMPTY']) assert bdining[0]['positive'] is True and cdining[0]['positive'] is True overflow=c['boundary-version-overflow'] assert overflow['http_status']==404 and overflow['body']['code']=='2' and overflow['positive_claimed'] is False assert overflow['type']==2147483646 red={x['case']:x for x in c['report']['cases']} assert all(red[name]['source_pass'] is False for name in ['EQUIPMENT_READ','PUSH_LIST_1','PUSH_LIST_2','MOBILE_RECIPE_OWNED']) native=[] for lane,data in [('B',b),('C',c)]: cleanup=data['cleanup'] assert cleanup['port_listening'] is False and cleanup['worker_state_exists'] is False assert cleanup['private_cache_files_remaining']==cleanup['token_artifacts_remaining']==0 assert cleanup['new_pdo_owned_actor_and_consumed_chain_verified'] is True assert cleanup['new_fixture_dml']==cleanup['ddl']==cleanup['business_rows_deleted']==0 assert cleanup['retained_rows']==7 and cleanup['fixture_meal_type']==4 h=data['h7-native-verification'];before=data['h7-native-before'];after=data['h7-native-after'] assert h['status']=='pass' and h['native_cli_version']=='2.2.8' and h['actual_native_requests']==6 and h['unique_scenarios']==3 assert h['new_http_methods']==h['new_fixture_dml']==h['ddl']==0 and h['owned_consumed_unchanged'] is True strict(h['source_sha256'],MIDDLE) assert after['unchanged'] is True;strict(before['own_rows'],after['own_rows']);strict(before['consumed_source'],after['consumed_source']) assert len(h['cases'])==3 assert [x['case'] for x in h['cases']]==['H7_NATIVE_HARDWARE4','H7_NATIVE_CHECK4','H7_NATIVE_SEARCH4'] for case in h['cases']: assert case['source_pass'] is True and case['wire_bytes']>0 and len(case['wire_sha256'])==64 body=case['json_body'];own=before['own_rows'] assert type(body['code']) is int and body['code']==0 if case['case']=='H7_NATIVE_HARDWARE4': assert body['message'] is True and len(body['data'])==1 image=body['data'][0] assert type(image['id']) is int and image['id']==own['date_menu_hardware_img']['id'] assert image['uuid']==own['date_menu_hardware_img']['uuid'] assert image['dishes'][0]['date_menu_uuid']==own['initial_menu']['uuid'] assert image['dishes'][0]['dishes_uuid']==own['dishes']['uuid'] assert image['dishes'][0]['weight']==own['initial_menu']['weight']=='12.34' assert image['dishes'][0]['name']==own['dishes']['name'] assert type(image['dishes'][1]) is list and image['dishes'][1]==[] else: staff=body['data']['rows'][0] assert staff['staff_uuid']==own['staff']['uuid'] and staff['staff_name']==own['staff']['name'] assert type(staff['sex']) is int and staff['sex']==own['staff']['sex'] assert staff['is_flag'] is True for report in h['reports'].values(): assert report['requests']==3 and report['assertions']==6 and report['failures']==0 native.append(h) for left,right in zip(native[0]['cases'],native[1]['cases']): a,b=copy.deepcopy(left['json_body']),copy.deepcopy(right['json_body']) if left['case']=='H7_NATIVE_HARDWARE4': assert type(a['data'][0]['id']) is int and type(b['data'][0]['id']) is int a['data'][0]['id']=b['data'][0]['id']='' strict(a,b) return {'status':'pass','method_candidates':6,'scope':'bounded partial; no full system claim', 'main_contract_cases_per_engine':10,'native_h7_unique_scenarios_per_engine':3, 'native_h7_actual_requests_per_engine':6,'native_h7_added_method_count':0, 'business_dml':0,'ddl':0,'dining_representative_selection_unresolved':True, 'full_natural_response_equal_claimed':False,'sql_fingerprint_gate_closed':False} if __name__=='__main__': if len(sys.argv)!=3: raise SystemExit('usage: verify_next6.py B-private-runtime C-private-runtime') print(json.dumps(verify(load(sys.argv[1]),load(sys.argv[2])),ensure_ascii=False,indent=2))