Open History to see the last commit
1#!/usr/bin/env python3
2"""python3 scripts/seed-issue-21397.test.py — runs the seeder against synthetic SQLite files only."""
3import datetime
4import hashlib
5import json
6import os
7import pathlib
8import sqlite3
9import subprocess
10import sys
11import tempfile
12import unittest
13
14SCRIPT = pathlib.Path(__file__).resolve().with_name('seed-issue-21397.py')
15FIXTURE = json.loads(SCRIPT.with_name('fixtures').joinpath('issue-21397.json').read_text())
16# Windows cannot delete a file SQLite still holds open, so cleanup must close connections first.
17OPEN_DB_PROBE = '''
18import runpy, shutil, sqlite3, sys
19opened, connect, rmtree = [], sqlite3.connect, shutil.rmtree
20def tracked(*args, **kwargs):
21 opened.append(connect(*args, **kwargs))
22 return opened[-1]
23def checked(*args, **kwargs):
24 for db in opened:
25 try:
26 db.total_changes
27 except sqlite3.ProgrammingError:
28 continue
29 sys.exit('rmtree ran while a SQLite connection was open')
30 return rmtree(*args, **kwargs)
31sqlite3.connect, shutil.rmtree = tracked, checked
32sys.argv = sys.argv[1:]
33runpy.run_path(sys.argv[0], run_name='__main__')
34'''
35
36
37def snapshot(directory):
38 return {
39 str(p.relative_to(directory)): hashlib.sha256(p.read_bytes()).hexdigest() if p.is_file() else 'dir'
40 for p in sorted(directory.rglob('*'))
41 }
42
43
44class SeederTest(unittest.TestCase):
45 def setUp(self):
46 temp = tempfile.TemporaryDirectory()
47 self.addCleanup(temp.cleanup)
48 self.tmp = pathlib.Path(temp.name)
49 self.fake_home = self.tmp / 'user'
50 (self.fake_home / '.t3/userdata').mkdir(parents=True)
51 self.source_home = self.tmp / 'worktree/.t3/userdata'
52 self.source_home.mkdir(parents=True)
53 self.source = self.source_home / 'statev2.sqlite'
54 db = sqlite3.connect(self.source)
55 db.execute('PRAGMA journal_mode=WAL')
56 for table, rows in FIXTURE.items():
57 db.execute(f'CREATE TABLE "{table}" ({",".join(sorted({k for r in rows for k in r}))})')
58 db.execute(f'INSERT INTO "{table}" (thread_id) VALUES (?)' if table != 'projection_projects'
59 else f'INSERT INTO "{table}" (project_id) VALUES (?)', ['sentinel'])
60 db.execute('CREATE TABLE orchestration_v2_projection_metadata (key)')
61 db.execute("INSERT INTO orchestration_v2_projection_metadata VALUES ('kept')")
62 db.execute('CREATE TABLE projection_thread_messages (id)')
63 db.execute("INSERT INTO projection_thread_messages VALUES ('sentinel')")
64 db.execute('CREATE TABLE orchestration_v2_effect_outbox (status)')
65 db.executemany('INSERT INTO orchestration_v2_effect_outbox VALUES (?)', [('pending',), ('done',)])
66 db.commit()
67 db.close()
68 (self.source_home / 'settings.json').write_text('{"sentinel": true}\n')
69 self.before = snapshot(self.source_home)
70
71 def run_seeder(self, *args, probe=()):
72 env = {**os.environ, 'HOME': str(self.fake_home)}
73 return subprocess.run([sys.executable, '-I', *probe, str(SCRIPT), '--source', str(self.source), *args],
74 env=env, capture_output=True, text=True)
75
76 def assertSourceUnchanged(self):
77 self.assertEqual(snapshot(self.source_home), self.before)
78
79 def thread_payload(self, home):
80 with sqlite3.connect(home / 'userdata/statev2.sqlite') as db:
81 self.assertEqual(db.execute('SELECT count(*) FROM projection_thread_messages').fetchone(), (0,))
82 self.assertEqual(db.execute('SELECT key FROM orchestration_v2_projection_metadata').fetchall(), [('kept',)])
83 self.assertEqual(db.execute('SELECT status FROM orchestration_v2_effect_outbox ORDER BY 1').fetchall(),
84 [('cancelled',), ('done',)])
85 rows = db.execute('SELECT thread_id, payload_json FROM orchestration_v2_projection_threads').fetchall()
86 self.assertEqual(len(rows), 1)
87 self.assertNotEqual(rows[0][0], 'sentinel')
88 return json.loads(rows[0][1])
89
90 def test_seeds_a_fresh_copy_for_every_mode_and_clock(self):
91 snooze = json.loads(FIXTURE['orchestration_v2_projection_threads'][0]['payload_json'])['snoozedUntil']
92 clocks = [([], datetime.timedelta(), '2026-10-22T19:00:00Z'),
93 (['--now', '2026-11-01T12:30:00+02:00'], datetime.timedelta(days=9, hours=15, minutes=30),
94 '2026-11-01T10:30:00Z'),
95 (['--now'], None, None)]
96 for mode in ('matching', 'absent', 'unrelated'):
97 for clock_args, delta, logged in clocks:
98 with self.subTest(mode=mode, clock=clock_args):
99 home = self.tmp / 'fixtures' / f'{mode}-{len(clock_args)}'
100 started = datetime.datetime.now(datetime.timezone.utc)
101 result = self.run_seeder('--home', str(home), mode, *clock_args)
102 self.assertEqual(result.returncode, 0, result.stderr)
103 self.assertSourceUnchanged()
104 if delta is None:
105 delta = started.replace(microsecond=0) - datetime.datetime(
106 2026, 10, 22, 19, tzinfo=datetime.timezone.utc)
107 logged = None
108 if logged:
109 self.assertIn(f'(limit hit at {logged})', result.stdout)
110 self.assertIn(f'Run: vp run dev --home-dir {home}', result.stdout)
111 expected = {'matching': snooze, 'absent': None, 'unrelated': '2026-10-24T08:50:00.000Z'}[mode]
112 actual = self.thread_payload(home)['snoozedUntil']
113 if expected is None:
114 self.assertIsNone(actual)
115 else:
116 moved = datetime.datetime.fromisoformat(actual.replace('Z', '+00:00')) - delta
117 original = datetime.datetime.fromisoformat(expected.replace('Z', '+00:00'))
118 self.assertLess(abs(moved - original), datetime.timedelta(seconds=2))
119 self.assertEqual(json.loads((home / 'userdata/settings.json').read_text()),
120 {'snoozeLimitedThreads': False, 'autoResumeLimitedThreads': False})
121
122 def test_refuses_unsafe_destinations_without_touching_anything(self):
123 existing_dir = self.tmp / 'existing'
124 existing_dir.mkdir()
125 (existing_dir / 'keep').write_text('keep')
126 existing_file = self.tmp / 'file'
127 existing_file.write_text('keep')
128 link_to_dir = self.tmp / 'link-dir'
129 link_to_dir.symlink_to(existing_dir)
130 dangling = self.tmp / 'dangling'
131 dangling.symlink_to(self.tmp / 'missing')
132 live_link = self.tmp / 'live-link'
133 live_link.symlink_to(self.fake_home / '.t3')
134 cases = {
135 'existing directory': existing_dir,
136 'existing file': existing_file,
137 'symlink to directory': link_to_dir,
138 'dangling symlink': dangling,
139 'source home': self.source_home.parent,
140 'live home': self.fake_home / '.t3',
141 'inside live home': self.fake_home / '.t3/fixture',
142 'through symlink into live home': live_link / 'fixture',
143 }
144 for name, home in cases.items():
145 with self.subTest(name):
146 listing = sorted(str(p) for p in self.tmp.rglob('*'))
147 result = self.run_seeder('--home', str(home))
148 self.assertNotEqual(result.returncode, 0)
149 self.assertIn('Refusing', result.stderr)
150 self.assertSourceUnchanged()
151 self.assertEqual(sorted(str(p) for p in self.tmp.rglob('*')), listing)
152 self.assertEqual((existing_dir / 'keep').read_text(), 'keep')
153 self.assertFalse((self.tmp / 'missing').exists())
154
155 def test_refuses_missing_or_symlinked_source(self):
156 link = self.tmp / 'linked.sqlite'
157 link.symlink_to(self.source)
158 for source in (self.tmp / 'absent.sqlite', link):
159 with self.subTest(str(source)):
160 env = {**os.environ, 'HOME': str(self.fake_home)}
161 result = subprocess.run([sys.executable, '-I', str(SCRIPT), '--source', str(source),
162 '--home', str(self.tmp / 'new')], env=env, capture_output=True, text=True)
163 self.assertNotEqual(result.returncode, 0)
164 self.assertFalse((self.tmp / 'new').exists())
165 self.assertSourceUnchanged()
166
167 def test_failed_seed_removes_only_its_new_home(self):
168 with sqlite3.connect(self.source) as db:
169 db.execute('DROP TABLE orchestration_v2_effect_outbox')
170 db.close()
171 self.before = snapshot(self.source_home)
172 home = self.tmp / 'broken'
173 result = self.run_seeder('--home', str(home), probe=('-c', OPEN_DB_PROBE))
174 self.assertNotEqual(result.returncode, 0)
175 self.assertIn('orchestration_v2_effect_outbox', result.stderr)
176 self.assertFalse(home.exists(), result.stderr)
177 self.assertSourceUnchanged()
178
179
180if __name__ == '__main__':
181 unittest.main()