aaronrene / overseer-kit public
test_mirror.py python
545 lines 25.1 KB
Raw
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8 docs: recenter public story on governance loop Human 7 days ago
1 """Real local Muse/Git objects; all delivery writes are recording stand-ins."""
2 from copy import deepcopy
3 import json
4 import os
5 from pathlib import Path
6 import subprocess
7 from unittest.mock import patch
8
9 import pytest
10 import yaml
11
12 from cli import v1_mirror as mirror
13 from cli.v1_io import Refusal, digest
14 from cli.v1_revision import muse_read
15 from tests.muse.test_handoffs import init, inventory
16 from tests.v1.conftest import command, KIT, seed
17
18
19 @pytest.fixture
20 def case(muse_root, muse_python, muse, tmp_path):
21 root = muse_root
22 init(root, muse_python)
23 (root / 'app.py').write_text('value = 1\n')
24 (root / 'data.bin').write_bytes(b'\0\xff\r\n\x01')
25 (root / 'script.txt').write_bytes(b'#!/bin/sh\n# A shebang is not executable authority.\n')
26 (root / 'run-tool').write_bytes(b'executable without a shebang\n')
27 (root / 'run-tool').chmod(0o755)
28 (root / 'templates').mkdir()
29 (root / 'templates/NEXT.md').write_text('Portable template\n')
30 muse(root, 'code', 'add', '.')
31 muse(root, 'commit', '-m', 'approved fixture source', '--json')
32 state = muse_read(root, str(muse_python))
33 plan = {'schema': 1, 'source': {'repo_id': state['repo_id'], 'branch': 'main',
34 'revision': state['head'], 'hub_url': 'https://staging.musehub.ai/fixture/source'},
35 'destination': {'url': 'https://github.com/fixture/mirror.git', 'repository': 'fixture/mirror',
36 'branch': 'muse-mirror', 'base': 'main', 'expected_head': 'absent'},
37 'target': str((tmp_path/'bare-mirror').resolve()), 'expected_target': 'absent',
38 'executables': ['run-tool']}
39 return root, plan
40
41
42 def execute(case, operation='prepare', transport=None):
43 root, plan = case
44 original = (root/'.overseer/config.yaml').read_bytes()
45 return mirror.execute(root, yaml.safe_load(original), original, plan, digest(mirror.packed(plan)), operation, transport)
46
47
48 def prepared(case):
49 result = execute(case)
50 assert result['ok'], json.dumps(result)
51 return result
52
53
54 class Delivery:
55 def __init__(self, plan):
56 self.plan = plan
57 self.head = plan['destination']['expected_head']
58 self.calls = []
59 self.fail_push = None
60 self.fail_pr = False
61 self.authoritative = (plan['source']['repo_id'], plan['source']['revision'])
62 self.pr = None
63
64 def authority(self, source):
65 self.calls.append(('authority', deepcopy(source)))
66 return self.authoritative
67
68 def remote_head(self, dest):
69 self.calls.append(('remote', deepcopy(dest)))
70 return self.head
71
72 def push(self, target, dest, commit):
73 self.calls.append(('push', str(target), deepcopy(dest), commit))
74 if self.fail_push != 'before':
75 self.head = commit
76 if self.fail_push:
77 raise OSError('interrupted response')
78
79 def ensure_pr(self, target, dest, source, commit):
80 self.calls.append(('pr', str(target), deepcopy(dest), deepcopy(source), commit))
81 if self.fail_pr:
82 raise Refusal('fixture_pr_failure')
83 self.pr = 'https://github.com/fixture/mirror/pull/7'
84 return self.pr
85
86
87 def test_projection_modes_mapping_and_offline_retry(case):
88 root, plan = case
89 before = inventory(root)
90 result = prepared(case)
91 assert result['push'] == result['pr'] == 'not_attempted'
92 assert result['authority'] == 'approved_input_not_checked_offline'
93 target = Path(plan['target'])
94 assert mirror.git(target, 'rev-parse', '--is-bare-repository') == b'true\n'
95 paths = mirror.git(target, 'ls-tree', '-r', '--name-only', 'HEAD').decode().splitlines()
96 assert 'templates/NEXT.md' in paths and 'docs/ROADMAP.md' in paths
97 assert '.overseer/config.yaml' not in paths and 'docs/NEXT.md' not in paths
98 assert mirror.git(target, 'show', 'HEAD:data.bin') == b'\0\xff\r\n\x01'
99 assert mirror.git(target, 'ls-tree', 'HEAD', 'run-tool').startswith(b'100755 blob ')
100 assert mirror.git(target, 'ls-tree', 'HEAD', 'script.txt').startswith(b'100644 blob ')
101 record = (root/mirror.RECORD).read_bytes()
102 again = prepared(case)
103 assert again['export'] == 'no_change_verified' and again['git_head'] == result['git_head']
104 assert (root/mirror.RECORD).read_bytes() == record
105 now = inventory(root)
106 del now[mirror.RECORD]
107 assert now == before
108 snapshots = inventory(root), inventory(target)
109 assert execute(case, 'verify')['ok']
110 assert snapshots == (inventory(root), inventory(target))
111
112
113 @pytest.mark.parametrize('change', ['head', 'branch', 'source_id', 'unknown_revision', 'working', 'mode', 'stage'])
114 def test_source_drift_refused_before_target(case, change, muse):
115 root, plan = case
116 if change == 'head':
117 plan['source']['revision'] = 'sha256:' + '1'*64
118 elif change == 'unknown_revision':
119 (root/'.muse/refs/heads/main').write_text('sha256:' + '2'*64)
120 elif change == 'branch':
121 muse(root, 'checkout', '-b', 'feature', '--json')
122 elif change == 'source_id':
123 plan['source']['repo_id'] = 'sha256:' + '3'*64
124 elif change == 'working':
125 (root/'app.py').write_text('changed after approval\n')
126 elif change == 'mode':
127 (root/'run-tool').chmod(0o644)
128 elif change == 'stage':
129 (root/'app.py').write_text('staged edit\n')
130 muse(root, 'code', 'add', 'app.py')
131 try:
132 result = execute(case)
133 assert not result['ok'], result
134 except Refusal:
135 pass
136 assert not Path(plan['target']).exists()
137
138
139 @pytest.mark.parametrize('kind', ['blob', 'commit', 'snapshot'])
140 def test_missing_objects_refused(case, muse_python, kind):
141 root, plan = case
142 code = '''import sys
143 from pathlib import Path
144 from muse.core.commits import read_commit
145 from muse.core.snapshots import read_snapshot
146 from muse.core.object_store import object_path
147 root=Path(sys.argv[1]);head=sys.argv[2];kind=sys.argv[3]
148 c=read_commit(root,head);s=read_snapshot(root,c.snapshot_id)
149 oid={'blob':s.manifest['app.py'],'commit':head,'snapshot':c.snapshot_id}[kind]
150 object_path(root,oid).unlink()
151 '''
152 subprocess.run([str(muse_python), '-I', '-B', '-c', code, str(root), plan['source']['revision'], kind], check=True)
153 assert not execute(case)['ok']
154 assert not Path(plan['target']).exists()
155
156
157 @pytest.mark.parametrize('kind', ['same', 'child', 'parent', 'dot_parent', 'alias', 'alias_parent', 'git_worktree', 'foreign', 'empty'])
158 def test_targets_are_physical_isolated_and_owned(case, tmp_path, kind):
159 root, plan = case
160 if kind == 'same': target = root
161 elif kind == 'child': target = root/'.muse/mirror'
162 elif kind == 'parent': target = root.parent
163 elif kind == 'dot_parent': target = root/'.muse/..'
164 elif kind == 'alias':
165 target = tmp_path/'alias'; target.symlink_to(root, target_is_directory=True)
166 elif kind == 'alias_parent':
167 alias = tmp_path/'alias'; alias.symlink_to(root.parent, target_is_directory=True)
168 target = alias/'bare-mirror'
169 elif kind == 'git_worktree':
170 dev = seed(tmp_path/'git')
171 target = tmp_path/'linked'
172 subprocess.run(['/usr/bin/git','-C',str(dev),'worktree','add','-b','mirror',str(target)],check=True,capture_output=True)
173 else:
174 target = Path(plan['target']); target.mkdir()
175 if kind == 'foreign': (target/'foreign').write_text('preserve\n')
176 plan['target'] = str(target)
177 before = inventory(root)
178 with pytest.raises(Refusal) if kind in {'same','child','parent','dot_parent','alias','alias_parent'} else __import__('contextlib').nullcontext():
179 result = execute(case)
180 assert not result['ok']
181 assert inventory(root) == before
182
183
184 @pytest.mark.parametrize('kind', ['untracked', 'config', 'hooks', 'ref', 'alternate', 'symlink'])
185 def test_dirty_stale_or_hooked_owned_target_refused(case, kind):
186 root, plan = case
187 prepared(case)
188 target = Path(plan['target'])
189 if kind == 'untracked': (target/'foreign.txt').write_text('must survive\n')
190 elif kind == 'config':
191 with (target/'config').open('a') as f: f.write('[core]\n hooksPath=/tmp/unsafe\n')
192 elif kind == 'hooks':
193 (target/'hooks').mkdir(); (target/'hooks/pre-push').write_text('#!/bin/sh\ntouch BAD\n')
194 elif kind == 'ref':
195 mirror.git(target, 'update-ref', 'refs/heads/foreign', mirror.head(target))
196 elif kind == 'alternate':
197 (target/'objects/info').mkdir(); (target/'objects/info/alternates').write_text('/unsafe\n')
198 elif kind == 'symlink': (target/'objects/alias').symlink_to(root)
199 before = inventory(target)
200 result = execute(case)
201 assert not result['ok']
202 assert inventory(target) == before
203 assert not (root/'BAD').exists()
204
205
206 def test_inherited_muse_hook_refused_preserved_and_never_run(case):
207 root, plan = case
208 marker = root.parent/'HOOK_RAN'
209 hook = root/'.muse/bridge-hooks.toml'
210 hook.write_text('[pre_bridge]\nhooks=[{run="touch ' + str(marker) + '"}]\n')
211 before = inventory(root)
212 result = execute(case)
213 assert not result['ok'] and 'inherited_muse_hooks: .muse/bridge-hooks.toml' in result['error']
214 assert inventory(root) == before and not marker.exists()
215 assert not Path(plan['target']).exists()
216
217
218 def test_inherited_git_environment_and_global_hooks_disabled(case, tmp_path, monkeypatch):
219 hook = tmp_path/'hooks'; hook.mkdir()
220 marker = tmp_path/'HOOK_RAN'
221 script = hook/'reference-transaction'
222 script.write_text('#!/bin/sh\ntouch "' + str(marker) + '"\n'); script.chmod(0o755)
223 global_config = tmp_path/'global-config'
224 global_config.write_text('[core]\n hooksPath=' + str(hook) + '\n')
225 monkeypatch.setenv('GIT_CONFIG_GLOBAL', str(global_config))
226 monkeypatch.setenv('GIT_CONFIG_COUNT', '1')
227 monkeypatch.setenv('GIT_CONFIG_KEY_0', 'core.hooksPath')
228 monkeypatch.setenv('GIT_CONFIG_VALUE_0', str(hook))
229 monkeypatch.setenv('GIT_DIR', '/wrong')
230 prepared(case)
231 assert not marker.exists()
232
233
234 def test_existing_snapshot_local_bindings_and_secrets_are_projected_out(case, muse):
235 root, plan = case
236 ignore = (root/'.museignore').read_bytes()
237 (root/'.museignore').write_text('[global]\npatterns=[]\n[force_track]\npaths=[".env.local"]\n')
238 (root/'.env.local').write_text('fixture-local-value\n')
239 muse(root, 'code', 'add', 'docs/NEXT.md', '.overseer/config.yaml', '.env.local')
240 muse(root, 'commit', '-m', 'legacy local binding fixture', '--json')
241 (root/'.museignore').write_bytes(ignore)
242 # .museignore was not staged above; the approved tracked copy remains equal.
243 state = muse_read(root, yaml.safe_load((root/'.overseer/config.yaml').read_bytes())['muse']['runtime']['python'])
244 plan['source']['revision'] = state['head']
245 result = prepared(case)
246 assert {'docs/NEXT.md', '.overseer/config.yaml', '.env.local'} <= set(result['excluded'])
247 assert b'local-value' not in mirror.git(Path(plan['target']), 'show', 'HEAD:templates/NEXT.md')
248
249
250 @pytest.mark.parametrize('kind', ['symlink', 'hardlink', 'unlisted_executable', 'missing_executable', 'unsupported_mode'])
251 def test_supported_modes_refuse_unsupported_source(case, kind):
252 root, plan = case
253 if kind == 'symlink':
254 (root/'app.py').unlink(); (root/'app.py').symlink_to(root/'data.bin')
255 elif kind == 'hardlink':
256 os.link(root/'app.py', root/'second.py')
257 elif kind == 'unlisted_executable': plan['executables'] = []
258 elif kind == 'missing_executable': plan['executables'] += ['unknown']
259 elif kind == 'unsupported_mode':
260 # This host strips setuid/setgid from fixture files; sticky persists.
261 (root/'app.py').chmod(0o1600)
262 assert (root/'app.py').stat().st_mode & 0o1000
263 assert not execute(case)['ok']
264 assert not Path(plan['target']).exists()
265
266
267 @pytest.mark.parametrize('stage', ['before_ref', 'after_ref'])
268 def test_interrupted_prepare_can_retry(case, stage):
269 root, plan = case
270 if stage == 'before_ref':
271 original = mirror.git
272 def fail(target, *args, **kw):
273 if args[0] == 'update-ref': raise OSError('interrupted before ref update')
274 return original(target, *args, **kw)
275 with patch.object(mirror, 'git', side_effect=fail): assert not execute(case)['ok']
276 else:
277 with patch.object(mirror, 'save_record', side_effect=OSError('interrupted record write')):
278 result = execute(case)
279 assert not result['ok'] and result['export'] == 'prepared'
280 commit = result['git_head']
281 result = prepared(case)
282 if stage == 'after_ref':
283 assert result['git_head'] == commit and result['export'] == 'no_change_verified'
284 assert json.loads((root/mirror.RECORD).read_bytes())['last_export']['git_sha'] == result['git_head']
285
286
287 @pytest.mark.parametrize('interruption', ['before', 'after'])
288 def test_interrupted_push_no_change_retry(case, interruption):
289 first = prepared(case)
290 transport = Delivery(case[1]); transport.fail_push = interruption
291 result = execute(case, 'deliver', transport)
292 assert result['export'] == 'no_change_verified'
293 assert result['ok'] == (interruption == 'after')
294 assert result['push'] == ('pending_retry' if interruption == 'before' else 'verified')
295 transport.fail_push = None
296 transport.calls.clear()
297 result = execute(case, 'deliver', transport)
298 assert result['ok'] and result['git_head'] == first['git_head']
299 assert sum(c[0] == 'push' for c in transport.calls) == (interruption == 'before')
300 assert result['pr'] == 'verified'
301
302
303 def test_pr_failure_preserves_push_and_retries_pr_with_repository_context(case):
304 prepared(case)
305 transport = Delivery(case[1]); transport.fail_pr = True
306 result = execute(case, 'deliver', transport)
307 assert not result['ok'] and result['push'] == 'verified' and result['pr'] == 'pending_retry'
308 transport.fail_pr = False; transport.calls.clear()
309 result = execute(case, 'deliver', transport)
310 assert result['ok'] and result['push'] == 'already_delivered'
311 assert not any(c[0] == 'push' for c in transport.calls)
312 pr = next(c for c in transport.calls if c[0] == 'pr')
313 assert pr[1] == case[1]['target'] and pr[2]['repository'] == 'fixture/mirror'
314
315
316 @pytest.mark.parametrize('kind', ['unknown_hub', 'unpushed_source', 'remote_drift', 'source_drift', 'target_drift'])
317 def test_delivery_requires_current_source_and_expected_remote(case, kind):
318 prepared(case)
319 transport = Delivery(case[1])
320 if kind == 'unknown_hub': transport.authoritative = None
321 elif kind == 'unpushed_source': transport.authoritative = (case[1]['source']['repo_id'], 'sha256:'+'7'*64)
322 elif kind == 'remote_drift': transport.head = '6'*40
323 elif kind == 'source_drift': (case[0]/'app.py').write_text('later edit\n')
324 elif kind == 'target_drift': (Path(case[1]['target'])/'foreign').write_text('foreign')
325 result = execute(case, 'deliver', transport)
326 assert not result['ok'] and not any(c[0] in {'push','pr'} for c in transport.calls)
327
328
329 def test_source_change_during_prepare_refuses_ref_and_can_resume(case):
330 root, plan = case
331 original = mirror.check_source
332 calls = 0
333 def changing(*args):
334 nonlocal calls
335 calls += 1
336 if calls == 2: (root/'app.py').write_text('changed mid-export\n')
337 return original(*args)
338 with patch.object(mirror, 'check_source', side_effect=changing): result = execute(case)
339 assert not result['ok'] and mirror.head(Path(plan['target'])) == 'absent'
340 (root/'app.py').write_text('value = 1\n')
341 prepared(case)
342
343
344 def test_add_change_delete_and_no_content_source_commit(case, muse):
345 root, plan = case
346 first = prepared(case)
347 (root/'data.bin').unlink()
348 (root/'app.py').write_text('value = 2\n')
349 (root/'new.txt').write_text('addition\n')
350 muse(root, 'code', 'add', '.')
351 muse(root, 'commit', '-m', 'approved second snapshot', '--json')
352 config = yaml.safe_load((root/'.overseer/config.yaml').read_bytes())
353 plan['source']['revision'] = muse_read(root, config['muse']['runtime']['python'])['head']
354 plan['expected_target'] = first['git_head']
355 plan['destination']['expected_head'] = first['git_head']
356 second = prepared(case)
357 paths = mirror.git(Path(plan['target']), 'ls-tree', '-r', '--name-only', 'HEAD').decode().splitlines()
358 assert 'data.bin' not in paths and 'new.txt' in paths
359 transport = Delivery(plan); transport.fail_push = 'before'
360 assert not execute(case, 'deliver', transport)['ok']
361 transport.fail_push = None
362 assert execute(case, 'deliver', transport)['ok']
363 # New approval with the same bytes retains correspondence through an empty commit.
364 muse(root, 'commit', '--allow-empty', '-m', 'new approved source; same projection', '--json')
365 plan['source']['revision'] = muse_read(root, config['muse']['runtime']['python'])['head']
366 plan['expected_target'] = second['git_head']
367 plan['destination']['expected_head'] = second['git_head']
368 third = prepared(case)
369 assert third['export'] == 'mapped_no_content_change' and third['git_head'] != second['git_head']
370
371
372 def test_concurrent_operators_and_cli_plan_pin(case):
373 root, plan = case
374 path = '.overseer/local/approved-mirror.json'
375 (root/'.overseer/local').mkdir(exist_ok=True)
376 raw = mirror.packed(plan); (root/path).write_bytes(raw)
377 args = [str(KIT/'cli/ok'), '-C', str(root), 'mirror', 'prepare', '--plan-file', path,
378 '--expect-plan', digest(raw), '--json']
379 workers = [subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) for _ in range(2)]
380 results = [p.communicate(timeout=30) for p in workers]
381 assert sorted(p.returncode for p in workers) in ([0,2], [0,0])
382 successes = [json.loads(out) for p,(out,err) in zip(workers,results) if p.returncode == 0]
383 assert successes and len({r['git_head'] for r in successes}) == 1
384 before = inventory(root), inventory(Path(plan['target']))
385 assert command(*args[1:-1], '--expect-plan', '0'*64).returncode != 0
386 assert before == (inventory(root), inventory(Path(plan['target'])))
387
388
389 def test_explicit_pair_binding_prevents_second_target(case):
390 prepared(case)
391 case[1]['target'] += '-other'
392 result = execute(case)
393 assert not result['ok'] and 'pair_changed' in result['error']
394 assert not Path(case[1]['target']).exists()
395
396
397 def test_network_adapter_argv_and_pr_failure_are_explicit(case, tmp_path):
398 root, plan = case
399 adapter = mirror.NetworkDelivery('/usr/bin/true')
400 target = Path(plan['target'])
401 calls = []
402 def run(_target, *args):
403 calls.append((_target, args))
404 return b'[]' if args[1] == 'list' else b''
405 with patch.object(adapter, 'gh_run', side_effect=run):
406 with pytest.raises(Refusal, match='pr_not_verified'):
407 adapter.ensure_pr(target, plan['destination'], plan['source'], 'a'*40)
408 assert all(c[0] == target and '--repo' in c[1] and 'fixture/mirror' in c[1] for c in calls)
409 with patch.object(mirror, 'git') as git:
410 adapter.push(target, plan['destination'], 'a'*40)
411 args = git.call_args.args
412 assert args == (target, 'push', '--porcelain', 'https://github.com/fixture/mirror.git', 'a'*40+':refs/heads/muse-mirror')
413
414
415 def test_target_ref_change_during_export_is_not_overwritten(case):
416 root, plan = case
417 original = mirror.check_source
418 calls = 0
419 moved = None
420 def changing(*args):
421 nonlocal calls, moved
422 calls += 1
423 result = original(*args)
424 if calls == 2:
425 target = Path(plan['target'])
426 empty = mirror.git(target, 'mktree', data=b'').decode().strip()
427 moved = mirror.git(target, 'commit-tree', empty, data=b'foreign operator\n').decode().strip()
428 mirror.git(target, 'update-ref', 'refs/heads/muse-mirror', moved)
429 return result
430 with patch.object(mirror, 'check_source', side_effect=changing): result = execute(case)
431 assert not result['ok'] and 'target_changed' in result['error']
432 assert mirror.head(Path(plan['target'])) == moved
433
434
435 @pytest.mark.parametrize('kind', ['foreign', 'bytes', 'mode'])
436 def test_corrupt_export_projection_cannot_update_ref(case, kind):
437 root, plan = case
438 original = mirror.tree
439 def foreign(target, files):
440 copy = deepcopy(files)
441 if kind == 'foreign': copy['foreign.txt'] = {'data': 'Zm9yZWlnbg==', 'mode': '100644'}
442 elif kind == 'bytes': copy['app.py']['data'] = 'Zm9yZWlnbg=='
443 elif kind == 'mode': copy['app.py']['mode'] = '100755'
444 return original(target, copy)
445 with patch.object(mirror, 'tree', side_effect=foreign): result = execute(case)
446 assert not result['ok'] and 'tree_' in result['error']
447 assert mirror.head(Path(plan['target'])) == 'absent'
448
449
450 def test_missing_git_blob_invalidates_no_change(case):
451 prepared(case)
452 target = Path(case[1]['target'])
453 oid = mirror.git(target, 'rev-parse', 'HEAD:app.py').decode().strip()
454 (target/'objects'/oid[:2]/oid[2:]).unlink()
455 assert not execute(case)['ok']
456
457
458 def test_remote_change_immediately_before_push_stops_delivery(case):
459 prepared(case)
460 transport = Delivery(case[1])
461 original = transport.remote_head
462 calls = 0
463 def moving(dest):
464 nonlocal calls
465 calls += 1
466 if calls == 2: transport.head = 'b'*40
467 return original(dest)
468 transport.remote_head = moving
469 result = execute(case, 'deliver', transport)
470 assert not result['ok'] and not any(c[0] in {'push','pr'} for c in transport.calls)
471
472
473 def test_mirror_requires_muse_selection_without_importing_it_in_git_mode(tmp_path):
474 root = seed(tmp_path/'git-only')
475 assert command('-C', root, 'init').returncode == 0
476 (root/'.overseer/local').mkdir(exist_ok=True)
477 # Complete but Git-only plan: selection is checked before any Muse operation.
478 plan = {'schema':1, 'source':dict(repo_id='sha256:'+'1'*64,branch='main',revision='sha256:'+'2'*64,
479 hub_url='https://staging.musehub.ai/fixture/source'),
480 'destination':dict(url='https://github.com/fixture/mirror.git',repository='fixture/mirror',
481 branch='muse-mirror',base='main',expected_head='absent'),
482 'target':str((tmp_path/'target').resolve()),'expected_target':'absent','executables':[]}
483 raw = mirror.packed(plan); path = '.overseer/local/plan.json'; (root/path).write_bytes(raw)
484 r = command('-C', root, 'mirror', 'prepare', '--plan-file', path, '--expect-plan', digest(raw))
485 assert r.returncode == 2 and 'explicit_muse_authority' in r.stderr
486 assert not Path(plan['target']).exists()
487
488
489 def test_invalid_cli_plan_and_missing_delivery_executable_refuse_cleanly(case):
490 root, plan = case
491 path = '.overseer/local/plan.json'; (root/'.overseer/local').mkdir(exist_ok=True)
492 for value, op in ((dict(plan, source=None), 'prepare'), (plan, 'deliver')):
493 raw = mirror.packed(value); (root/path).write_bytes(raw)
494 r = command('-C', root, 'mirror', op, '--plan-file', path, '--expect-plan', digest(raw))
495 assert r.returncode == 2 and 'Traceback' not in r.stderr, r.stderr
496
497
498 def test_legacy_git_destination_is_not_silently_replaced(case):
499 case[1]['destination']['expected_head'] = 'a'*40
500 with pytest.raises(Refusal, match='legacy_destination'):
501 execute(case)
502 assert not Path(case[1]['target']).exists()
503
504
505 @pytest.mark.parametrize('value', [[], None, {'domain':'code','branch_heads':[]}, {'domain':'midi','branch_heads':{}}])
506 def test_malformed_authority_observation_is_a_refusal(value):
507 from io import BytesIO
508 adapter = mirror.NetworkDelivery('/usr/bin/true')
509 source = {'hub_url':'https://staging.musehub.ai/fixture/source','branch':'main'}
510 with patch('urllib.request.build_opener') as opener:
511 opener.return_value.open.return_value = BytesIO(json.dumps(value).encode())
512 with pytest.raises(Refusal, match='mirror_authority_'):
513 adapter.authority(source)
514
515
516 @pytest.mark.parametrize('raw', [b'malformed\n', b'bad\trefs/heads/muse-mirror\n', b'a'*40+b'\trefs/heads/other\n'])
517 def test_malformed_git_observation_is_a_refusal(raw):
518 adapter = mirror.NetworkDelivery('/usr/bin/true')
519 with patch.object(mirror, 'git', return_value=raw):
520 with pytest.raises(Refusal, match='remote_response_invalid'):
521 adapter.remote_head({'url':'https://github.com/fixture/mirror.git','branch':'muse-mirror'})
522
523
524 def test_malformed_pr_url_is_a_retryable_refusal():
525 adapter = mirror.NetworkDelivery('/usr/bin/true')
526 dest = {'repository':'fixture/mirror','branch':'muse-mirror','base':'main'}
527 with patch.object(adapter, 'gh_run', return_value=json.dumps([
528 {'url':42,'headRefOid':'a'*40,'isCrossRepository':False}]).encode()):
529 with pytest.raises(Refusal, match='pr_not_verified'):
530 adapter.ensure_pr(Path('/'), dest, {}, 'a'*40)
531
532
533 @pytest.mark.parametrize('value', [{}, '', None, False, 0, [None], [{}],
534 [{'url': None, 'headRefOid': 'a'*40, 'isCrossRepository': False}],
535 [{'url': '', 'headRefOid': 'a'*40, 'isCrossRepository': False}],
536 [{'url': 'https://github.com/fixture/mirror/pull/1', 'headRefOid': 'a'*40, 'isCrossRepository': 0}],
537 [{'url': 'https://github.com/fixture/mirror/pull/1', 'headRefOid': 'invalid', 'isCrossRepository': False}]])
538 def test_unknown_pr_state_never_creates_a_pr(value):
539 adapter = mirror.NetworkDelivery('/usr/bin/true')
540 dest = {'repository':'fixture/mirror','branch':'muse-mirror','base':'main'}
541 with patch.object(adapter, 'gh_run', return_value=json.dumps(value).encode()) as gh:
542 with pytest.raises(Refusal, match='mirror_pr_'):
543 adapter.ensure_pr(Path('/'), dest, {}, 'a'*40)
544 assert gh.call_count == 1
545 assert gh.call_args.args[1:3] == ('pr', 'list')
File History 1 commit
sha256:ff22b6a82a08c117176c3e2603264a0e49fe8204ce57c3cefe7f5b165a8634f8 docs: recenter public story on governance loop Human 7 days ago