"""Offline audit of immutable HTTP and full-row snapshots; never opens a DB.""" import json, re, hashlib, sys from pathlib import Path ROOT = Path(__file__).parent checks = 0 def equal(a, b): if type(a) is not type(b): return False if isinstance(a, dict): return list(a) == list(b) and all(equal(a[k], b[k]) for k in a) if isinstance(a, list): return len(a) == len(b) and all(equal(x, y) for x, y in zip(a, b)) return a == b def assert_eq(a, b, reason): global checks if not equal(a, b): raise AssertionError(reason) checks += 1 def read(p): return json.loads(p.read_text()) def protect(s, uuid): c = dict(s) c['activity'] = [r for r in s['activity'] if r['uuid'] != uuid] return c def dto(rows, method): if method == 'info': row = dict(rows[0]); row['activity_status'] = '未开始'; row['participants'] = 0 return {'code': 0, 'message': '信息获取成功', 'data': row} if method == 'lists': rr = [dict(r, activity_status='未开始') for r in rows] return {'code': 0, 'message': '请求成功', 'data': {'rows': rr, 'total': len(rows)}} return {'code': 0, 'message': {'add':'新增活动成功','update':'编辑活动成功','able':'更改状态成功','delete':'删除活动成功'}[method], 'data': []} summaries = {} for lane in ('B', 'C'): folder = ROOT / 'private' / lane base = read(folder / 'baseline.json'); own = read(folder / 'owned.json'); uuid = own['uuid'] pre = read(folder / 'preflight.json'); proof = read(folder / 'tap-proof-v2.json') assert_eq(pre['marker_collision'], 0, 'Fresh marker collision') assert_eq(protect(proof['source'], uuid), base, 'Exact post-add protected state') for role in ('store','ai'): pp = proof['proof'][role] assert_eq(pp['native_class'], 'ActPDO', 'Effective actual native PDO') assert_eq(pp['identity'][0]['db'], pp['config_database'], 'Native DB identity') assert_eq(pp['class'], 'think\\db\\connector\\Mysql' if lane == 'B' else 'app\\cores\\database\\Dm', 'Original driver class') previous = base native_reads = native_dml = 0 for stage in range(1, 12): label = f'{stage:02d}' w = read(folder / f'wire-{label}.json'); client = read(folder / f'client-{label}.json'); response = read(folder / f'response-{label}.json') case = w['case']; method = case['method'] assert_eq(w['before'], previous, f'{lane}/{label} chronological full before') assert_eq(protect(w['after'], uuid), base, f'{lane}/{label} protected full typed state') assert_eq(client['status'], 200, 'Actual HTTP status') assert_eq(response['status'], 200, 'Original Response status') assert_eq(client['body'], response['body'], 'Original wire response bytes') assert_eq(w['body'], response['body'], 'Original server response bytes') assert_eq(hashlib.sha256(response['body'].encode()).hexdigest(), response['body_sha256'], 'Original response checksum') assert_eq(w['warnings'], [], 'Request-local warnings') assert_eq(w.get('business_error'), None, 'No source exception') assert_eq(w.get('observer_error'), None, 'No observer failure') actual = json.loads(response['body']) ownrows = [r for r in w['after']['activity'] if r['uuid'] == uuid] expected = dto([r for r in w['after']['activity'] if 'CDXACT11W1' in r['title']] if method == 'lists' else ownrows, method) if stage == 8 and lane == 'B': expected = {'code':1,'message':'修改失败','data':[]} assert_eq(actual, expected, f'{lane}/{label} complete independently derived DTO') if method in ('add','update','able'): row = ownrows[0] if stage == 1: assert_eq(row, own['full_created_row'], 'Actual first created row journal') assert_eq(row['store_id'], 10001, 'Tenant') assert_eq(row['create_by'], base['actor'][0]['name'], 'Original get_create_by') assert re.fullmatch(r'[0-9A-Fa-f-]{36}', row['uuid']) else: original = [r for r in w['before']['activity'] if r['uuid'] == uuid][0] merged = dict(original) for key, value in case['allowed']['data'].items(): merged[key] = int(value) if type(row[key]) is int else value assert_eq(row, merged, 'Full own row only exact approved fields change') for key, value in case['params'].items(): if key != 'uuid': physical = str(value) if lane=='C' and key in ('register_start','register_end','start_time','end_time'): column = next(c for c in pre['metadata']['activity_columns'] if c['column_name'].lower()==key) assert_eq(column['data_type'],'DATETIME','Actual datetime metadata') assert_eq(column['data_scale'],6.0,'Actual fractional scale') physical += '.' + '0'*int(column['data_scale']) assert_eq(str(row[key]), physical, f'{lane}/{stage}/{key} actual persisted request field') if stage >= 2: assert_eq(sorted(set(x['schema'] for x in w['native'])), ['ai','store'], 'Actual HTTP dual native connections') writes = [x for x in w['native'] if re.match(r'^\s*(INSERT|UPDATE|DELETE)\b',x['sql'],re.I)] assert_eq(len(writes), 1 if case['allowed'] else 0, 'Exact native write attempts') native_dml += len(writes) for x in writes: assert_eq(x['schema'], 'ai', 'Only AI activity DML') assert_eq(x['execute_parameters'], None, 'No implicit binds') assert_eq(x['success'], True, 'Original driver execute success') assert_eq(x['affected'], 0 if lane=='B' and stage==8 else 1, 'Actual affected rows') assert 'yoshop_activity' in x['sql'] and uuid in [str(b['value']) for b in x['binds'].values()] else: assert_eq(w['native'], [], 'Original INSERT capture limitation preserved') assert_eq(len(w['after']['activity']),len(base['activity'])+1,'Actual API insert once') previous = w['after'] native_reads += sum(not re.match(r'^\s*(INSERT|UPDATE|DELETE)\b',x['sql'],re.I) for x in w['native']) assert_eq(previous, base, 'All natural state returns exactly to original after delete') assert_eq(read(folder/'final.json')['source'],base,'Independent final source exact baseline') assert_eq(read(folder/'delete-lease.json')['references'],[],'Zero registrations before delete') assert_eq(read(folder/'cleanup.json')['own_port_closed'],True,'Own port closed') assert_eq(read(folder/'cleanup.json')['original_parent_alive'],True,'Original parent preserved') budget = {p.stem.replace('budget-',''):int(p.read_text()) for p in folder.glob('budget-*.json')} assert_eq(budget['HTTP'],11,'Actual requests11') assert_eq(budget['observer'],436 if lane=='B' else 580,'Exact independent observer SELECT count') assert_eq(budget['DML-update'],3,'Native UPDATE attempts3') assert_eq(budget['DML-delete'],1,'Native DELETE1') assert 'DML-insert' not in budget total = budget['observer'] + budget['business-read'] if lane=='B': assert_eq(budget['total-read'],total,'Unified observed read counter') summaries[lane]={'HTTP':11,'actual_API_INSERT':1,'captured_INSERT':0,'captured_UPDATE_attempts':3,'captured_DELETE':1,'captured_business_reads':budget['business-read'],'observer_SELECT':budget['observer'],'known_readonly_total':total,'first_add_business_reads':'UNKNOWN_UNCAPTURED','original_600_scope_status':'EXCEEDED_KNOWN_AT_LEAST_682' if lane=='C' else 'SUPERSEDED_FOR_REMAINING_CHAIN; INITIAL_ADD_READS_UNKNOWN','B_remaining_chain_total_read_ceiling_raised_by_Root':800 if lane=='B' else None,'original_repeatable_code':1 if lane=='B' else 0,'full_final_state_exact':True,'native_initial_add_limited':True} cross=[] for stage in (2,3,5,7,9): bb=json.loads(read(ROOT/'private'/'B'/f'response-{stage:02d}.json')['body'])['data'] cc=json.loads(read(ROOT/'private'/'C'/f'response-{stage:02d}.json')['body'])['data'] if stage==3: bb,cc=bb['rows'][0],cc['rows'][0] for key in bb: if key not in ('id','uuid','create_by','create_time') and not equal(bb[key],cc[key]): cross.append({'stage':stage,'field':key,'B':bb[key],'C':cc[key]}) assert_eq(sorted(set(d['field'] for d in cross)),['end_time','register_end','register_start','start_time'],'Preserved exact cross-DB time precision differences') mutations=[(1,1.0),([],{}),(None,''),(0,False),({'x':1},{'x':1.0}),({'x':1,'y':2},{'y':2,'x':1}),('2099-01-01 00:00:00','2099-01-01 00:00:00.000000')] for a,b in mutations: assert_eq(equal(a,b),False,'Full typed comparator mutation rejected') result={'status':'LIMITED_INITIAL_INSERT_NATIVE_CAPTURE; ORIGINAL_NOOP_AND_DATETIME_COMPATIBILITY_RED; C_READ_BUDGET_EXCEEDED','checks':checks,'type_mutations_rejected':len(mutations),'lanes':summaries,'cross_DTO_datetime_differences':cross} print(json.dumps(result,ensure_ascii=False,indent=2))