#!/usr/bin/env python3 """Offline evidence gate. No HTTP, DB, normalization of scalar types, or writes.""" import copy import hashlib import json import pathlib import sys CASES = ['HARDWARE4', 'CHECK4', 'SEARCH4', 'PREFIX4', 'HARDWARE_FALLBACK1', 'CHECK1', 'RESTORED_CHECK4'] TABLES = ['staff', 'department', 'staff_department', 'dishes', 'recipes', 'date_menu_hardware_img', 'initial_menu'] HARDWARE = '/p/date_menu/hardwareInfo' CHECK = '/p/dateMenu/checkMenuList' FROZEN_SOURCES = {line.split(None,1)[1]:line.split(None,1)[0] for line in (pathlib.Path(__file__).resolve().parent/'SOURCE_SHA256.txt').read_text().splitlines() if line} def strict(a, b, path='$'): assert type(a) is type(b), f'{path}: type' if isinstance(a, dict): assert set(a) == set(b), f'{path}: keys' for k in a: strict(a[k], b[k], path+'.'+k) elif isinstance(a, list): assert len(a) == len(b), f'{path}: length' for i, (x, y) in enumerate(zip(a, b)): strict(x, y, path+f'[{i}]') else: assert a == b, f'{path}: value' def load(directory): p = pathlib.Path(directory) return {key: json.loads((p/(name if (p/name).is_file() else name.replace('-private', ''))).read_bytes()) for key, name in { 'report': 'report-private.json', 'responses': 'responses-private.json', 'source': 'source-private.json', 'cleanup': 'resource-cleanup-private.json'}.items()} def verify(b, c): outcomes = [] for lane, data in [('B', b), ('C', c)]: r, responses, sources, cleanup = (data[k] for k in ['report', 'responses', 'source', 'cleanup']) assert r['lane'] == lane and r['unexpected'] is None strict(r['source_sha256'], FROZEN_SOURCES) assert [x['case'] for x in r['cases']] == CASES assert set(responses) == set(sources) == set(CASES) assert r['fixture_restored'] is True and r['actor_unchanged'] is True and r['restaurant_unchanged'] is True assert r['http_dml'] == 0 assert [(x['from'], x['to'], x['rows']) for x in r['updates']] == [(4, 1, 1), (1, 4, 1)] assert all(type(x['owned_pk']) is int and x['owned_pk'] == r['retained_ids']['initial_menu'] for x in r['updates']) assert set(r['retained_ids']) == set(TABLES) assert all(type(v) is int and v > 0 for v in r['retained_ids'].values()) if lane == 'B': assert r['root_authorized_consumed_scope'] is True and r['consumed_source_unchanged'] is True assert r['global_natural_unchanged_proven'] is False and r['natural_staff_fulltable_unchanged'] is False assert r['initial_menu_global_count_telemetry']['acceptance_scope'] is False else: assert r['natural_unchanged'] is True strict(r['natural_before'], r['natural_after_excluding_owned']) assert r['boundary']['private_source_reflection'] is True and r['boundary']['external_stubs'] is True assert r['boundary']['real_queue'] is False and r['boundary']['file_cache'] is True and r['boundary']['redis_disabled'] is True ends = [x for x in r['events'] if x['kind'] == 'request_end'] assert len(ends) == 7 and all(x['fatal'] is False and x['http_status'] == 200 and x['observed_queries'] > 0 for x in ends) assert {x['scenario'] for x in ends} == {'PILOT_HC_'+s for s in CASES} queries = [x for x in r['events'] if x['kind'] == 'query_observed'] assert not any(x['operation'] in ['insert', 'update', 'delete', 'replace', 'merge'] for x in queries) for i, name in enumerate(CASES): case, resp, proof = r['cases'][i], responses[name], sources[name] hardware = name.startswith('HARDWARE') path = HARDWARE if hardware else CHECK assert case['source_pass'] is True and case['difference'] is None and case['expected_nonempty'] is True assert case['http_status'] == resp['http_status'] == 200 and case['code'] == 0 assert case['method'] == resp['method'] == 'POST' and case['path'] == resp['path'] == path assert proof['meal_type'] == case['meal_type'] == (1 if name in ['HARDWARE_FALLBACK1', 'CHECK1'] else 4) assert type(resp['wire_bytes']) is int and resp['wire_bytes'] > 0 assert len(resp['wire_sha256']) == 64 if 'raw_wire' in resp: raw = resp['raw_wire'].encode() assert len(raw) == resp['wire_bytes'] and hashlib.sha256(raw).hexdigest() == resp['wire_sha256'] strict(json.loads(raw), resp['body']) strict(proof['body'], resp['body']) own = proof['own_rows'] assert set(own) == set(TABLES) for table in TABLES: assert type(own[table]['id']) is int and own[table]['id'] == r['retained_ids'][table] if table != 'date_menu_hardware_img': assert own[table]['create_by'] == r['marker'] assert own['staff_department']['staff_uuid'] == own['staff']['uuid'] assert own['staff_department']['department_uuid'] == own['department']['uuid'] assert own['initial_menu']['staff_uuid'] == own['date_menu_hardware_img']['staff_uuid'] == own['staff']['uuid'] assert own['initial_menu']['hardware_img_uuid'] == own['date_menu_hardware_img']['uuid'] assert own['recipes']['dishes_uuid'] == own['initial_menu']['dishes_uuid'] == own['dishes']['uuid'] if lane == 'B': assert case['owned_request_window_unchanged'] is True and case['consumed_relation_window_unchanged'] is True strict(proof['consumed_before'], proof['consumed_after']) assert proof['consumed_before']['child_uuids'] == [own['department']['uuid']] body = resp['body'] assert type(body['code']) is int and body['code'] == 0 if hardware: 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'] and image['img'] == 'CDX_TEST_METADATA/no-real-file' assert type(image['dishes']) is list and len(image['dishes']) == 2 and type(image['dishes'][1]) is list and image['dishes'][1] == [] dish = image['dishes'][0] assert type(dish['weight']) is str and dish['weight'] == own['initial_menu']['weight'] == '12.34' assert type(dish['code']) is int and dish['code'] == 1 assert type(dish['in_recipes']) is int and dish['in_recipes'] == 1 assert type(dish['un_edit']) is int and dish['un_edit'] == 1 assert dish['name'] == own['dishes']['name'] and dish['dishes_uuid'] == own['dishes']['uuid'] else: assert body['message'] == '查询成功' and type(body['data']['total']) is int and body['data']['total'] == 1 assert len(body['data']['rows']) == 1 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'] == 1 and type(staff['meal_times']) is str and staff['meal_times'] == '1' assert type(staff['is_flag']) is bool and staff['is_flag'] is (case['meal_type'] == 4) assert cleanup['lane'] == lane and cleanup['port'+('18391' if lane == 'B' else '18392')+'_listening'] is False assert cleanup['worker_state_exists'] is False and cleanup['private_cache_files_remaining'] == 0 assert cleanup['credential_tokens_persisted'] is False and cleanup['business_rows_deleted'] == 0 and cleanup['owned_chain_retained'] is True outcomes.append({'lane': lane, 'positive_cases': 7, 'observed_queries': len(queries), 'lexical_warnings': sum(x['fingerprint_lexical_warning'] is True for x in queries), 'contracted_natural_snapshot_unchanged': r['natural_unchanged']}) strict(b['report']['source_sha256'], c['report']['source_sha256']) for name in CASES: left = copy.deepcopy(b['responses'][name]['body']) right = copy.deepcopy(c['responses'][name]['body']) if name.startswith('HARDWARE'): # Only after positive integer PK + own original row + shared UUID proof above. left['data'][0]['id'] = right['data'][0]['id'] = '' strict(left, right) return {'status': 'pass', 'real_http_methods': ['POST '+HARDWARE, 'POST '+CHECK], 'paired_positive_cases': 7, 'lanes': outcomes, 'comparison': 'strict JSON type/value/shape/order; only proven image PK mapped', 'raw_wire_literal_equality_claimed': False, 'full_natural_database_unchanged_claimed': False, 'sql_fingerprint_gate_closed': False, 'full_system_compatible_claimed': False} if __name__ == '__main__': if len(sys.argv) != 3: raise SystemExit('usage: verify.py B-private-runtime C-private-runtime') print(json.dumps(verify(load(sys.argv[1]), load(sys.argv[2])), ensure_ascii=False, indent=2))