import contextlib import copy import importlib.util import io import json import pathlib import sqlite3 import tempfile import time from unittest.mock import patch ROOT = pathlib.Path('/home/ubuntu/wt/issue-15873') SOURCE = ROOT / 'scripts/fixtures/issue-15873/seed.py' spec = importlib.util.spec_from_file_location('fixture_seed', SOURCE) seed = importlib.util.module_from_spec(spec) spec.loader.exec_module(seed) all_data = seed.fixtures() before = json.loads(pathlib.Path('/tmp/15873-fixtures-before.json').read_text()) changes = json.loads(pathlib.Path('/tmp/15873-fixture-field-diff.json').read_text()) assert len(changes) == 13 assert before['G'] == all_data['G'] and before['S'] == all_data['S'] for name in ('A', 'L', 'C'): try: seed.validate(before[name]) except AssertionError: pass else: raise AssertionError(('baseline unexpectedly valid', name)) print('Baseline regression checks: A/L/C fail new invariants; G/S unchanged.') class MemoryConnection(sqlite3.Connection): def close(self): pass memory = sqlite3.connect(':memory:', factory=MemoryConnection) for table, rows in all_data['G'].items(): if not rows: continue columns = list(rows[0]) primary = ['thread_id', 'turn_item_id'] if table == 'orchestration_v2_turn_item_positions' else columns[:1] defs = [f'"{c}"' for c in columns] defs.append('PRIMARY KEY (' + ','.join(f'"{c}"' for c in primary) + ')') memory.execute(f'CREATE TABLE "{table}" ({",".join(defs)})') with patch.object(seed.sqlite3, 'connect', return_value=memory), contextlib.redirect_stdout(io.StringIO()): seed.seed(ROOT / '.t3/capture/userdata/statev2.sqlite') seed.seed(ROOT / '.t3/pass2/userdata/statev2.sqlite') for table, rows in all_data['G'].items(): if not rows: continue count = memory.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0] assert count == sum(len(d[table]) for d in all_data.values()), (table, count) for name, data in all_data.items(): seed.validate(data) thread = data['orchestration_v2_projection_threads'][0]['thread_id'] maximum = memory.execute('SELECT MAX(ordinal) FROM orchestration_v2_projection_runs WHERE thread_id=?', (thread,)).fetchone()[0] assert maximum == {'G':3, 'S':3, 'A':2, 'L':2, 'C':1}[name] row, payload = memory.execute('SELECT last_run_ordinal, payload_json FROM orchestration_v2_projection_provider_threads WHERE thread_id=?', (thread,)).fetchone() assert row == json.loads(payload)['lastRunOrdinal'] == maximum for run, ordinal, payload in memory.execute('SELECT run_id, app_run_ordinal, payload_json FROM orchestration_v2_projection_checkpoints WHERE thread_id=?', (thread,)): expected = None if run is None else memory.execute('SELECT ordinal FROM orchestration_v2_projection_runs WHERE run_id=?', (run,)).fetchone()[0] assert ordinal == json.loads(payload)['appRunOrdinal'] == expected for table in ('orchestration_v2_projection_messages', 'orchestration_v2_projection_turn_items'): columns = ['updated_at'] if table.endswith('turn_items') else ['created_at','updated_at'] for row in memory.execute(f'SELECT {",".join(columns)},payload_json FROM {table} WHERE thread_id=?',(thread,)): p=json.loads(row[-1]) for column, value in zip(columns,row[:-1]): assert value == p[{'created_at':'createdAt','updated_at':'updatedAt'}[column]] for table, column, payload_key in ( ('orchestration_v2_projection_provider_threads','last_run_ordinal','lastRunOrdinal'), ('orchestration_v2_projection_checkpoints','app_run_ordinal','appRunOrdinal'), ('orchestration_v2_projection_messages','updated_at','updatedAt'), ('orchestration_v2_projection_turn_items','updated_at','updatedAt')): for corrupt_payload in (False, True): broken=copy.deepcopy(data) row=broken[table][-1] if corrupt_payload: p=json.loads(row['payload_json']); p[payload_key]='incorrect'; row['payload_json']=json.dumps(p) else: row[column]='incorrect' try: seed.validate(broken) except AssertionError: pass else: raise AssertionError(('missed mutation',name,table,corrupt_payload)) print('All 5 fixtures: in-memory seed twice, row counts/idempotence, run/checkpoint/provider ordinals and message/item timestamps passed; 40 corruptions rejected.') sqlite3.Connection.close(memory) (ROOT / '.t3').mkdir(exist_ok=True) rejected = 0 with tempfile.TemporaryDirectory(prefix='15873-guard-') as outside, tempfile.TemporaryDirectory(prefix='15873-guard-', dir=ROOT / '.t3') as inside: outside=pathlib.Path(outside); inside=pathlib.Path(inside) (inside/'escape').symlink_to(outside, target_is_directory=True) (inside/'statev2.sqlite').symlink_to(outside/'statev2.sqlite') invalid=[ outside/'.t3/userdata/statev2.sqlite', ROOT.parent/'other-worktree/.t3/userdata/statev2.sqlite', ROOT/'statev2.sqlite', ROOT/'.t3/capture/userdata/state.sqlite', ROOT/'.t3/../statev2.sqlite', pathlib.Path.home()/'.t3/userdata/statev2.sqlite', inside/'escape/userdata/statev2.sqlite', inside/'statev2.sqlite', ] for path in invalid: with patch.object(seed.sqlite3,'connect',side_effect=AssertionError('database opened')) as connect, patch.object(seed,'fixtures',side_effect=AssertionError('fixtures generated')): try: seed.seed(path) except ValueError: rejected += 1 else: raise AssertionError(('accepted unsafe destination',path)) connect.assert_not_called() with patch.object(seed.pathlib.Path,'home',return_value=ROOT), patch.object(seed.sqlite3,'connect',side_effect=AssertionError('live database opened')) as connect: try: seed.seed(ROOT/'.t3/capture/userdata/statev2.sqlite') except ValueError: rejected += 1 else: raise AssertionError('live home exception lost') connect.assert_not_called() print(f'Guard: {rejected} unsafe paths rejected before sqlite open; capture/pass2 accepted through in-memory seeding; no live DB opened.') original = json.loads((SOURCE.parent/'runtime.json').read_text()) malformed = ' '.join(['msg_zzzz call_zzzz rs_zzzz 00000000-0000-0000-0000-zzzzzzzzzzzz'] * 750) original['orchestration_v2_projection_messages'][0]['payload_json'] = json.dumps(dict(json.loads(original['orchestration_v2_projection_messages'][0]['payload_json']), text=malformed)) original['orchestration_v2_projection_turn_items'][0]['payload_json'] = json.dumps(dict(json.loads(original['orchestration_v2_projection_turn_items'][0]['payload_json']), text=malformed)) started = time.perf_counter() with patch.object(seed.pathlib.Path,'read_text',return_value=json.dumps(original)): stress=seed.fixtures() elapsed=time.perf_counter()-started print(f'3000 malformed identity tokens in linked message/item payloads: all 5 fixtures validated in {elapsed:.4f}s.') pathlib.Path('/tmp/15873-validation-results.json').write_text(json.dumps({'fixtures':5,'rejected_mutations':40,'rejected_paths':rejected,'malformed_tokens':3000,'stress_seconds':elapsed,'changed_fields':len(changes)},indent=2))