Verify brief publication against live origin and record historical gaps

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09cbd-43c1-79f3-809e-1ee97b40b64d

scripts/test_brief_durability.py

@@ -0,0 +1,136 @@
+import io
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from verify_brief_durability import Origin, assess, brief_path, git, load_events
+
+
+class DurabilityTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.remote = self.root / 'remote.git'
+        self.repo = self.root / 'repo'
+        self.repo.mkdir()
+        git(self.root, 'init', '--bare', str(self.remote))
+        git(self.repo, 'init', '-b', 'main')
+        git(self.repo, 'config', 'user.name', 'Test')
+        git(self.repo, 'config', 'user.email', 'test@example.invalid')
+        git(self.repo, 'remote', 'add', 'origin', str(self.remote))
+        (self.repo / 'README.md').write_text('fixture\n')
+        self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+
+    def commit(self):
+        git(self.repo, 'add', '.')
+        git(self.repo, 'commit', '-m', 'fixture')
+        return git(self.repo, 'rev-parse', 'HEAD').stdout.strip()
+
+    def write_brief(self, day='2026-09-14', text='A brief\n'):
+        p = self.repo / brief_path(day)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(text)
+        return p
+
+    def event(self, day='2026-09-14', **kwargs):
+        detail = dict(repo='freedom-intelligence', date=day, path=brief_path(day), **kwargs)
+        return dict(id='event-1', detail=detail)
+
+    def test_untracked_file_does_not_count(self):
+        self.write_brief()
+        self.assertFalse(Origin(self.repo).blob(brief_path('2026-09-14')))
+
+    def test_unpushed_commit_does_not_count(self):
+        self.write_brief()
+        self.commit()
+        self.assertFalse(Origin(self.repo).blob(brief_path('2026-09-14')))
+
+    def test_fresh_remote_beats_stale_tracking_ref(self):
+        old = git(self.repo, 'rev-parse', 'HEAD').stdout.strip()
+        self.write_brief()
+        sha = self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        git(self.repo, 'update-ref', 'refs/remotes/origin/main', old)
+        origin = Origin(self.repo)
+        self.assertEqual(origin.sha, sha)
+        self.assertTrue(origin.blob(brief_path('2026-09-14')))
+
+    def test_fetches_missing_remote_objects(self):
+        clone = self.root / 'other'
+        git(self.root, 'clone', '-b', 'main', str(self.remote), str(clone))
+        self.write_brief()
+        self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        self.assertTrue(Origin(clone).blob(brief_path('2026-09-14')))
+
+    def test_valid_published_completion(self):
+        self.write_brief()
+        sha = self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        result = assess([self.event(pushed=True, origin_sha=sha)], [], Origin(self.repo), '2026-09-14')
+        self.assertEqual(len(result['valid']), 1)
+        self.assertEqual(result['failures'], [])
+
+    def test_unpublished_attestation_rejected_even_when_file_is_remote(self):
+        self.write_brief()
+        self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        (self.repo / 'extra').write_text('unpublished')
+        unpublished = self.commit()
+        result = assess([self.event(pushed=True, origin_sha=unpublished)], [], Origin(self.repo))
+        self.assertEqual(len(result['failures']), 1)
+
+    def test_missing_push_evidence_rejected_for_new_events(self):
+        self.write_brief()
+        self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        self.assertEqual(len(assess([self.event()], [], Origin(self.repo))['failures']), 1)
+
+    def test_empty_file_rejected(self):
+        self.write_brief(text='')
+        self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        self.assertFalse(Origin(self.repo).blob(brief_path('2026-09-14')))
+
+    def test_acknowledged_gap_is_not_a_success(self):
+        event = self.event('2026-09-07')
+        record = dict(original_event_id=event['id'], date='2026-09-07',
+                      path=brief_path('2026-09-07'), disposition='false_completion', reason='unpublished')
+        (self.repo / 'evidence.json').write_text(json.dumps({'corrections': [record]}))
+        sha = self.commit()
+        git(self.repo, 'push', 'origin', 'main')
+        correction = dict(detail=dict(record, repo='freedom-intelligence',
+                                      evidence_path='evidence.json', evidence_sha=sha))
+        result = assess([event], [correction], Origin(self.repo), '2026-09-07')
+        self.assertEqual(len(result['reconciled']), 1)
+        self.assertEqual(result['valid'], [])
+        self.assertEqual(len(result['failures']), 1)  # required day still fails
+        correction['detail']['original_event_id'] = 'different-event'
+        result = assess([event], [correction], Origin(self.repo))
+        self.assertEqual(result['reconciled'], [])
+        self.assertEqual(len(result['failures']), 1)
+
+    def test_malformed_dates_fail_without_path_traversal(self):
+        result = assess([dict(id='bad', detail={'date': '../../etc/passwd'})], [], Origin(self.repo))
+        self.assertEqual(len(result['failures']), 1)
+
+
+class PaginationTests(unittest.TestCase):
+    def test_reads_all_pages(self):
+        pages = [[{'id': '1'}, {'id': '2'}], [{'id': '3'}]]
+        with patch('urllib.request.urlopen', side_effect=[io.BytesIO(json.dumps(p).encode()) for p in pages]) as call:
+            self.assertEqual(len(load_events('http://hub', 1, page_size=2)), 3)
+            self.assertIn('offset=2', call.call_args.args[0].full_url)
+
+    def test_repeating_page_fails(self):
+        with patch('urllib.request.urlopen', side_effect=[io.BytesIO(b'[{"id":"1"}]'), io.BytesIO(b'[{"id":"1"}]')]):
+            with self.assertRaises(RuntimeError):
+                load_events('http://hub', 1, page_size=1)
+
+
+if __name__ == '__main__':
+    unittest.main()