aaronrene / overseer-kit public
test_integrity.py python
111 lines 4.4 KB
Raw
sha256:106ef9e12cca7f3be3ffb2a6bd3abfa76f032dce453c9e5010d732a8d86f2e14 Reconcile reviewed credential compatibility 32254cd939885d9… Human patch 8 days ago
1 import hashlib
2 import os
3 from pathlib import Path
4 import subprocess
5 import time
6 from unittest.mock import patch
7
8 import pytest
9
10 from cli.v1_io import Refusal, atomic, digest, read
11 from tests.v1.conftest import command, status, write_action
12
13
14 @pytest.mark.parametrize('component',['.overseer/config.yaml','docs/NEXT.md','docs','.overseer'])
15 def test_symlinks_cannot_cross_repository(pair,tmp_path,component):
16 a,b=pair
17 target=b/component
18 saved=target.with_name(target.name+'.saved')
19 target.rename(saved)
20 target.symlink_to(a/component,target_is_directory=saved.is_dir())
21 for cmd in ('next','status','sync'):
22 r=command('-C',b,cmd)
23 assert r.returncode != 0 and 'Unique content for '+a.name not in r.stdout
24
25
26 def test_hardlinked_next_refused(pair):
27 a,b=pair
28 (b/'docs/NEXT.md').unlink();os.link(a/'docs/NEXT.md',b/'docs/NEXT.md')
29 r=command('-C',b,'next')
30 assert r.returncode != 0 and not r.stdout
31
32
33 @pytest.mark.parametrize('failure',['replace','fsync'])
34 def test_atomic_failure_preserves_old_bytes_and_removes_temp(tmp_path,failure):
35 (tmp_path/'docs').mkdir();path=tmp_path/'docs/NEXT.md';path.write_bytes(b'old\n')
36 with patch('cli.v1_io.os.'+failure,side_effect=OSError('injected before publication')):
37 with pytest.raises(OSError):
38 atomic(tmp_path,'docs/NEXT.md',b'new\n',expected=digest(b'old\n'))
39 assert path.read_bytes()==b'old\n'
40 assert list(path.parent.iterdir())==[path]
41
42
43 def test_atomic_unique_temp_and_permissions(tmp_path):
44 atomic(tmp_path,'docs/NEXT.md',b'first\n',expected='absent')
45 with pytest.raises(Refusal,match='stale_write'):
46 atomic(tmp_path,'docs/NEXT.md',b'second\n',expected='absent')
47 assert (tmp_path/'docs/NEXT.md').stat().st_mode & 0o777 == 0o644
48
49
50 def test_writer_rejects_external_or_symlink_prompt(repo,tmp_path):
51 outside=tmp_path/'outside';outside.write_text('external')
52 for candidate in ('../outside',str(outside),'prompt-link'):
53 if candidate=='prompt-link': (repo/candidate).symlink_to(outside)
54 state=status(repo)
55 result=write_action(repo,'valid',prompt_file=candidate)
56 assert result.returncode != 0
57 assert status(repo)['next_sha256']==state['next_sha256']
58
59
60 def test_oversized_next_refused(repo):
61 (repo/'docs/NEXT.md').write_bytes(b'x'*140000)
62 assert command('-C',repo,'next').returncode != 0
63
64
65 def test_concurrent_readers_and_cas_writers(repo):
66 state=status(repo)
67 (repo/'prompt-a').write_text('Concurrent A\n');(repo/'prompt-b').write_text('Concurrent B\n')
68 from tests.v1.conftest import KIT
69 args=[str(KIT/'cli/ok'),'-C',str(repo),'next-write','--repo-id',state['repository']['id'],
70 '--branch','main','--lane','product','--model','GPT-6 Astra','--action-id','CONCURRENT',
71 '--action-kind','implement','--expect-next',state['next_sha256']]
72 workers=[subprocess.Popen(args+['--prompt-file',p],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
73 for p in ('prompt-a','prompt-b')]
74 for _ in range(5):
75 r=command('-C',repo,'next');assert r.returncode==0,r.stderr
76 outcomes=[(p.communicate(timeout=10),p.returncode) for p in workers]
77 assert sorted(code for _,code in outcomes)==[0,2]
78 assert not list((repo/'docs').glob('.NEXT.md.*'))
79
80
81 def test_next_read_bounded(repo):
82 start=time.monotonic()
83 for _ in range(10):assert command('-C',repo,'next').returncode==0
84 assert time.monotonic()-start < 10
85
86
87 def test_post_replace_directory_fsync_failure_reports_failure_with_whole_file(tmp_path):
88 atomic(tmp_path,'docs/NEXT.md',b'old',expected='absent')
89 original=os.fsync;calls=0
90 def fail_directory(fd):
91 nonlocal calls
92 calls+=1
93 if calls==2:raise OSError('directory persistence failed after publication')
94 return original(fd)
95 with patch('cli.v1_io.os.fsync',side_effect=fail_directory):
96 with pytest.raises(OSError):atomic(tmp_path,'docs/NEXT.md',b'new',expected=digest(b'old'))
97 assert (tmp_path/'docs/NEXT.md').read_bytes()==b'new'
98 assert not list((tmp_path/'docs').glob('.NEXT.md.*'))
99
100
101 def test_raw_digest_preserves_exact_next_bytes():
102 assert digest(b'a\r\n') != digest(b'a\n')
103
104
105 def test_old_test_sources_preserved():
106 import json
107 from tests.v1.conftest import KIT
108 entries=json.loads((KIT/'tests/historical/baseline-test-inventory.json').read_text())
109 assert entries
110 for item in entries:
111 assert digest((KIT/item['path']).read_bytes())==item['sha256'],item['path']
File History 2 commits
sha256:106ef9e12cca7f3be3ffb2a6bd3abfa76f032dce453c9e5010d732a8d86f2e14 Reconcile reviewed credential compatibility 32254cd939885d9… Human patch 8 days ago
sha256:c87cf2c2f03da507f66f93922cb339a2cbebb33ad6372a6039032fb09a20da12 Reconcile reviewed Overseer recovery d47291d5d9030de5ebef71… Human minor ⚠ 9 days ago