#!/usr/bin/env python3 """Pure contract gate, written before the 2B candidate; not a database acceptance test.""" import copy import sys import hashlib import json import re from pathlib import Path r = Path(__file__).parent b = json.loads((r / 'private-evidence/B-fresh.json').read_text()) c = json.loads((r / 'private-evidence/C-fresh.json').read_text()) names = [x['column_name'].upper() for x in b['columns']] def contract(p): def resolve(value): if isinstance(value, dict): assert set(value) == {'payload_ref'} and value['payload_ref'] in p['payloads'] return p['payloads'][value['payload_ref']] return value assert p['status'] == 'PROPOSAL_ONLY_NOT_AUTHORIZED_NOT_RUN' assert len(p['columns']) == 77 and p['columns'] == names assert p['marker'] == 'CDXSS2B' and p['max_rows_each'] == 1 and p['lease_ids'] == [1, 2] assert p['business_DDL_DML_HTTP'] == 0 trigger = p['lanes']['C']['create_trigger'] assert trigger.index(':= RTRIM(:NEW.PRIMARY_DEPARTMENT_UUID') < trigger.index('IF UPDATING AND') assert trigger.index(':= RTRIM(:NEW.DINGTALK_SYNC_MD5') < trigger.index('IF UPDATING AND') assert set(re.findall(r'^ -- FIELD ([A-Z0-9_]+)$', trigger, re.M)) == set(names) - {'UPDATE_TIME', 'SPORT_EVENT_UUID'} assert len(re.findall(r'^ -- FIELD ', trigger, re.M)) == 75 assert 'DBMS_LOB.COMPARE' not in trigger and 'V_OFFSET := V_OFFSET + 256' in trigger assert "NOT UPDATING('UPDATE_TIME')" in trigger and "UPDATING('SPORT_EVENT_UUID')" in trigger assert 'V_OLD_CHUNK VARCHAR(2048)' in trigger and 'V_NEW_CHUNK VARCHAR(2048)' in trigger assert p['lanes']['B']['create_table'] == 'CREATE TABLE cdx_ss7_2b261011 LIKE ydy_staff' for lane, spec in p['lanes'].items(): assert spec['table'] == ('cdx_ss7_2b261011' if lane == 'B' else 'ZHCT_APP.CDX_SS7_2B261011') assert len(spec['seed']['binds']) == 76 and 'UPDATE_TIME' not in spec['seed']['columns'] assert len(spec['cases']) == p['budget']['cases_each'] covered = {x['field_coverage'] for x in spec['cases'] if x.get('field_coverage')} assert covered == set(names) - {'UPDATE_TIME'} assert spec['cases'][-1]['field_coverage'] == 'ID' for case in spec['cases']: assert len(case['reset']['binds']) == 78 for stmt in (case['reset'], case['mutation']): assert stmt['SQL'].startswith('UPDATE ' + spec['table'] + ' SET ') assert stmt['SQL'].count('?') == len(stmt['binds']) == len(stmt['PDO_types']) binds = [resolve(v) for v in stmt['binds']] assert all(type(v) in (str, int, type(None)) for v in binds) assert stmt['PDO_types'] == [0 if v is None else 1 if type(v) is int else 2 for v in binds] assert case['reset']['binds'][-1] == 1 assert case['expected_time'] in ('advance_to_own_DB_window', 'preserve_anchor', 'explicit_value', 'invalid_source_observe_only') assert len(spec['cleanup']) == (1 if lane == 'B' else 3) assert all('OR REPLACE' not in x for x in [spec['create_table'], spec.get('create_trigger') or '']) labels = {x['label'] for x in p['lanes']['B']['cases']} for field in ('CARD_ID', 'POSITION', 'DINGTALK_USERID', 'PRIMARY_DEPARTMENT_UUID'): assert 'NULL_EMPTY_' + field in labels and 'EMPTY_NULL_' + field in labels for label in ('ALL76_SELF', 'CHAR_TAIL_ONLY_NOOP', 'USERID_CASE', 'USERID_TAIL_BYTES', 'VARCHAR_CASE', 'VARCHAR_TAIL_BYTES', 'EXPLICIT_TIME_AND_CHANGE', 'SELF_TIME_AND_CHANGE', 'CLOB_NULL_INVALID', 'CLOB_LONG_SAME', 'CLOB_LONG_LAST_CASE', 'CLOB_UTF8_CHUNK_LAST_CASE'): assert label in labels long_case = next(x for x in p['lanes']['B']['cases'] if x['label'] == 'CLOB_LONG_LAST_CASE') assert len(resolve(long_case['mutation']['binds'][0])) == len(resolve(long_case['mutation']['binds'][0]).encode()) == 68001 unicode_case = next(x for x in p['lanes']['B']['cases'] if x['label'] == 'CLOB_UTF8_CHUNK_LAST_CASE') assert len(resolve(unicode_case['mutation']['binds'][0])) == 257 and len(resolve(unicode_case['mutation']['binds'][0]).encode()) == 1025 old_c_definitions = [line.rstrip(',') for line in c['DDL'][0]['ddl'].splitlines() if re.match(r'^"[A-Z0-9_]+" ', line)] assert len(old_c_definitions) == 70 assert all(line in p['lanes']['C']['create_table'] for line in old_c_definitions) assert p['budget']['DML_attempts_max_each'] == 1 + 2 * p['budget']['cases_each'] assert p['budget']['DDL_B'] == 2 and p['budget']['DDL_C'] == 6 and p['budget']['SELECT_max_each'] == 650 # Verify that compact reviewer can reconstruct every actual reset/mutation binding without hidden choices. for lane, spec in p['lanes'].items(): q = (lambda f: '\u0060' + f + '\u0060') if lane == 'B' else (lambda f: '"' + f + '"') for case in spec['cases']: reset_values = dict(p['baseline_values'], **case['reset_overrides']) expected = [reset_values[f] for f in p['columns']] + [1] assert case['reset']['binds'] == expected assert resolve(case['reset']['binds'][p['columns'].index('UPDATE_TIME')]) == '2001-01-01 00:00:00' set_values = case['set_values'] assert case['mutation']['binds'] == list(set_values.values()) + [1] expected_sql = 'UPDATE ' + spec['table'] + ' SET ' + ','.join([q(f) + '=?' for f in set_values] + [q(f) + '=' + q(f) for f in case['self_columns']]) + ' WHERE ' + q('ID') + '=?' assert case['mutation']['SQL'] == expected_sql # Static expression references (not merely the comment manifest). for field in set(names) - {'UPDATE_TIME', 'SPORT_EVENT_UUID'}: assert ':NEW."' + field + '"' in trigger and ':OLD."' + field + '"' in trigger sport = next(x for x in b['columns'] if x['column_name'] == 'sport_event_uuid') assert sport['data_type'] == 'mediumtext' and sport['character_octet_length'] == 16777215 assert all(len(v.encode()) < sport['character_octet_length'] for v in p['payloads'].values()) return {'pure_contract': True, 'columns': 77, 'scalar_static_coverage': 75, 'actual_safe_field_cases': 76, 'cases_each': p['budget']['cases_each'], 'max_DML_each': p['budget']['DML_attempts_max_each'], 'database_execution': False} if __name__ == '__main__': raw = (r / 'probe2b-full77-time-proposal.json').read_bytes() p = json.loads(raw) result = contract(p) result['plan_SHA'] = hashlib.sha256(raw).hexdigest() if '--test-mutations' in sys.argv: def case(plan, label, lane='C'): return next(x for x in plan['lanes'][lane]['cases'] if x['label'] == label) mutations = [ lambda d: d['lanes']['C'].update(create_trigger=d['lanes']['C']['create_trigger'].replace(':OLD."WEIGHT"', ':OLD."HEIGHT"')), lambda d: d['lanes']['C'].update(create_trigger=d['lanes']['C']['create_trigger'].replace("NOT UPDATING('UPDATE_TIME')", 'TRUE')), lambda d: d['lanes']['C'].update(create_trigger=d['lanes']['C']['create_trigger'].replace('V_OFFSET := V_OFFSET + 256', 'V_OFFSET := V_OFFSET + 512')), lambda d: d['lanes']['C'].update(create_table=d['lanes']['C']['create_table'].replace('"CREATE_TIME" DATETIME(6)', '"CREATE_TIME" DATETIME(0)')), lambda d: case(d, 'FIELD_NAME')['reset']['binds'].__setitem__(d['columns'].index('UPDATE_TIME'), None), lambda d: case(d, 'FIELD_DINGTALK_LAST_SEEN_FULL_BATCH_ID')['mutation']['PDO_types'].__setitem__(0, 1), lambda d: d['payloads'].__setitem__('LOB_LONG_a', d['payloads']['LOB_LONG_a'][:-1]), lambda d: case(d, 'NULL_EMPTY_PRIMARY_DEPARTMENT_UUID')['mutation']['binds'].__setitem__(0, None), lambda d: d['lanes']['B']['cases'].pop(), lambda d: case(d, 'SELF_TIME_AND_CHANGE')['mutation'].update(SQL=case(d, 'SELF_TIME_AND_CHANGE')['mutation']['SQL'].replace('\"UPDATE_TIME\"=\"UPDATE_TIME\"', '\"UPDATE_TIME\"=CURRENT_TIMESTAMP')), ] for mutate in mutations: changed = copy.deepcopy(p) mutate(changed) try: contract(changed) except AssertionError: continue raise AssertionError('Plan mutation was not detected') result['mutations_rejected'] = len(mutations) print(json.dumps(result))