import copy import datetime import hashlib import json import re from pathlib import Path from zoneinfo import ZoneInfo ROOT = Path(__file__).resolve().parent TZ = ZoneInfo('Asia/Shanghai') def equal(a, b): if type(a) is not type(b): return False if isinstance(a, dict): return a.keys() == b.keys() 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 cast(row, columns): result = {} for col in columns: key = col['column_name']; value = row[key] if col['data_type'].lower() in {'int', 'integer', 'bigint', 'tinyint', 'smallint'}: assert value is not None value = int(value) result[key] = value return result def original_projection(row, table, meta, hidden=()): result = cast(row, meta['tables'][table]['columns']) for key in hidden: result.pop(key, None) return result def action_projection(row, meta): result = original_projection(row, 'sign_action', meta, ['store_id', 'is_delete', 'create_by', 'create_time', 'update_time']) if meta.get('_date_candidate') and meta['lane'] == 'C': for key in ['start_time', 'end_time']: if re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.0{1,6}', result[key]): datetime.datetime.strptime(result[key][:19], '%Y-%m-%d %H:%M:%S') result[key] = result[key][:19] return result def expected_detail(case, meta): source = case['independent_original']; ident = int(case['params']['signId']) raw = next(x for x in source['action'] if int(x['id']) == ident) result = action_projection(raw, meta) addresses = [] for row in source['address']: if int(row['action_id']) != ident or int(row['is_delete']) != 0: continue value = original_projection(row, 'sign_address', meta, ['is_delete', 'create_by', 'create_time', 'update_time']) for field, target in [('qrurl_id', 'qrurl'), ('receive_url_id', 'receive_url')]: if int(row[field]) == 0: value[target] = '' continue file = next(x for x in source['file'] if int(x['file_id']) == int(row[field])) assert file['storage'] == 'local' value[target] = 'http://' + case['module'] + '.sign-config.invalid/uploads/' + file['file_path'] addresses.append(value) result.update(addressList=addresses, cover_img=[], cover_img_url='', rule_img=[], rule_img_url='') return result def verify(case, meta): body = case['body']; assert body['trace_id'] == 'VALID_DYNAMIC_TRACE' assert type(body['status']) is int and case['http_status'] == 200 assert body['status'] == case['expected_status'], 'Business status: ' + case['scenario'] if case['expected_message'] is not None: assert body['message'] == case['expected_message'] assert type(body['message']) is str if body['status'] != 200: assert body['data'] == []; return action = case['action']; source = case['independent_original'] if action == 'add': tenant = 2147001571 if case['module'] == 'store' else 2147001572 rows = [r for r in source['action'] if int(r['store_id']) == tenant and int(r['is_delete']) == 0] assert len(rows) == 1 assert equal(body['data'], {'signId': str(rows[0]['id'])}) form = case['params']['form'] for field in ['title', 'start_time', 'end_time']: assert action_projection(rows[0], meta)[field] == form[field] assert int(rows[0]['status']) == form['status'] addresses = [r for r in source['address'] if int(r['action_id']) == int(rows[0]['id'])] assert [r['address'] for r in addresses] == [r['address'] for r in form['addressList']] assert len(addresses) == 2 files = [r for r in source['file'] if int(r['store_id']) == tenant] assert len(files) == 4 for row in addresses: assert int(row['qrurl_id']) > 0 and int(row['receive_url_id']) > 0 for field in ['qrurl_id', 'receive_url_id']: f = next(f for f in files if int(f['file_id']) == int(row[field])) assert f['file_name'] == str(row['action_id']) + '_' + str(row['id']) + '.png' assert int(f['channel']) == 30 and int(f['file_type']) == 10 elif action == 'detail': assert equal(body['data'], {'detail': expected_detail(case, meta)}), 'Complete detail original projection: ' + case['scenario'] elif action == 'list': tenant = 2147001571 if case['module'] == 'store' else 2147001572 search = (case['params'] or {}).get('search', '') rows = [r for r in source['action'] if int(r['store_id']) == tenant and int(r['is_delete']) == 0 and search in r['title']] rows.sort(key=lambda r: int(r['id']), reverse=True) expected = {'list': {'total': len(rows), 'per_page': 15, 'current_page': 1, 'last_page': (len(rows) + 14) // 15, 'data': [action_projection(r, meta) for r in rows[:15]]}} assert equal(body['data'], expected), 'Complete list original projection: ' + case['scenario'] else: assert body['data'] == [] row = next(r for r in source['action'] if int(r['id']) == int(case['params']['signId'])) if action == 'delete': assert int(row['is_delete']) == 1 if action == 'edit': form = case['params']['form']; assert row['title'] == form['title'] and int(row['status']) == form['status'] addresses = [r for r in source['address'] if int(r['action_id']) == int(row['id'])] assert [(int(r['id']), r['address']) for r in addresses] == [(r['id'], r['address']) for r in form['addressList']] def normalize(case): value = copy.deepcopy(case['body']); source = case['independent_original']; module = case['module'] ids = {} for i, row in enumerate(source['action']): ids[('action', int(row['id']))] = 'ACTION_' + str(row['store_id']) for row in source['action']: addresses = [r for r in source['address'] if int(r['action_id']) == int(row['id'])] for i, addr in enumerate(addresses): ids[('address', int(addr['id']))] = ids[('action', int(row['id']))] + '_ADDRESS_' + str(i) for addr in source['address']: for field in ['qrurl_id', 'receive_url_id']: ids[('file', int(addr[field]))] = ids[('address', int(addr['id']))] + '_' + field def walk(v, key=None): if isinstance(v, list): return [walk(x, key) for x in v] if isinstance(v, dict): typ = 'address' if 'action_id' in v else 'action' out = {} for k, x in v.items(): if k == 'id': out[k] = ids[(typ, x)] elif k == 'signId': assert type(x) is str out[k] = {'native_type': 'string', 'logical_id': ids[('action', int(x))]} elif k == 'action_id': out[k] = ids[('action', x)] elif k in ['qrurl_id', 'receive_url_id']: out[k] = ids[('file', x)] if x != 0 else 0 elif k in ['qrurl', 'receive_url']: field = 'qrurl_id' if k == 'qrurl' else 'receive_url_id' if v[field] == 0: assert x == '' out[k] = '' continue f = next(r for r in source['file'] if int(r['file_id']) == v[field]) assert x.endswith('/' + f['file_path']) out[k] = ids[('file', v[field])] + '_SOURCE_PATH_VERIFIED' else: out[k] = walk(x, k) return out return v return walk(value) def main(): red = {k: json.loads((ROOT / ('red-evidence-' + k + '.json')).read_text()) for k in ['B', 'C']} green = {k: json.loads((ROOT / ('green-evidence-' + k + '.json')).read_text()) for k in ['B', 'C']} final = {k: json.loads((ROOT / ('final-evidence-' + k + '.json')).read_text()) for k in ['B', 'C']} meta = {k: json.loads((ROOT / ('structure-' + k + '.json')).read_text()) for k in red} errors = []; normalized = {}; historical_failures = [] for lane in ['B', 'C']: assert red[lane]['native_totals'] == {'requests': 32, 'assertions': 138, 'failures': 3 if lane == 'B' else 8} assert green[lane]['native_totals']['failures'] == 0 and green[lane]['errors'] == [] assert final[lane]['native_totals'] == {'requests': 2 if lane == 'B' else 4, 'assertions': 8 if lane == 'B' else 16, 'failures': 0} for phase, e in [('red', red[lane]), ('green', green[lane]), ('final', final[lane])]: assert e['protected_unchanged'] and e['final_protected_unchanged'] and e['before'] == e['after'] == red[lane]['before'] metadata = copy.deepcopy(meta[lane]) metadata['_date_candidate'] = e['boundary']['hashes']['app/common/model/SignAction.php'] == '996cc4428bcd336e8ca2a06dfb54976445b3c9f7eaeb0a1290e0ec1606740cd4' normalized[(lane, phase)] = {} for case in e['cases']: if phase == 'red' and case['native_cli_exit'] != 0: historical_failures.append({'lane': lane, 'scenario': case['scenario'], 'message': case['body']['message']}) continue try: verify(case, metadata) normalized[(lane, phase)][case['scenario'].replace('PILOT_SG_R_', '').replace('PILOT_SG_G_', '')] = normalize(case) except (AssertionError, KeyError, StopIteration, TypeError) as err: errors.append({'lane': lane, 'phase': phase, 'scenario': case['scenario'], 'error': str(err)}) pairs = [] for scenario in normalized[('B', 'red')]: if scenario in normalized[('C', 'green')]: pairs.append((scenario, normalized[('B', 'red')][scenario], normalized[('C', 'green')][scenario])) for scenario in normalized[('B', 'green')]: assert scenario in normalized[('C', 'green')] pairs.append((scenario, normalized[('B', 'green')][scenario], normalized[('C', 'green')][scenario])) differences = [s for s, b, c in pairs if not equal(b, c)] for lane in ['B', 'C']: a = json.loads((ROOT / ('final-audit-' + lane + '.json')).read_text()) assert a['global_DML'] == a['DDL'] == a['final_read_DML'] == 0 assert len(a['business_DML']) == (26 if lane == 'B' else 40) and a['own_fixture_DML'] == 2 assert all(v['own_scope'] for v in a['business_DML']) and len(a['local_files']) == 8 assert all(v['own_action_address'] for v in a['external_leaves'] if v['kind'] != 'token') assert a['protected_original'] == red[lane]['before'] out = {'status': 'PASS' if not errors and not differences else 'FAIL', 'original_projection_errors': errors, 'differences': differences, 'strict_full_body_pairs': len(pairs), 'final_full_original_projection_reads': sum(len(e['cases']) for e in final.values()), 'actual_native_business_HTTP_total': sum(e['native_totals']['requests'] for phase in [red, green, final] for e in phase.values()), 'actual_native_named_assertions_total': sum(e['native_totals']['assertions'] for phase in [red, green, final] for e in phase.values()), 'historical_actual_failures_retained': historical_failures, 'request_counts': {k: {'initial': red[k]['native_totals'], 'repair': green[k]['native_totals'], 'final_all_rows': final[k]['native_totals']} for k in red}, 'scope': 'Ten configuration methods, partial DB scenarios; synthetic network and QR leaves, no real QR/WeChat/OSS acceptance.'} (ROOT / 'comparison.json').write_text(json.dumps(out, ensure_ascii=False, indent=2) + '\n') print(json.dumps(out, ensure_ascii=False)); assert out['status'] == 'PASS' if __name__ == '__main__': main()