from datetime import datetime, timezone import hashlib from pathlib import Path import tempfile import unittest from intelligence_radar.__main__ import check, draft from intelligence_radar.external_briefs import import_inbox, native_status, parse_name, verify from intelligence_radar.scheduler import run_due class ExternalBriefTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) / 'repo' self.inbox = Path(self.temp.name) / 'inbox' self.root.mkdir() self.inbox.mkdir() (self.inbox / 'archive').mkdir() self.now = datetime(2026, 10, 4, 12, tzinfo=timezone.utc) def test_name_parses_daily_and_weekly_dates(self): daily = parse_name('261003-intelligenceRadarDaily_grok.txt') self.assertEqual(daily['period'], '2026-10-03') self.assertEqual(daily['cadence'], 'daily') weekly = parse_name('260920-intelligenceRadarWeekly_chatgpt.txt') self.assertEqual(weekly['cadence'], 'weekly') self.assertEqual(weekly['filename_date'], '2026-09-20') self.assertTrue(weekly['period'].startswith('2026-W')) self.assertIsNone(parse_name('notes.txt')) self.assertIsNone(parse_name('261399-intelligenceRadarDaily_grok.txt')) def test_inbox_retains_new_files_and_skips_archive_and_empty(self): (self.inbox / '261003-intelligenceRadarDaily_grok.txt').write_text('Frontier note\n', encoding='utf-8') (self.inbox / '260920-intelligenceRadarWeekly_chatgpt.txt').write_text('Week note\n', encoding='utf-8') (self.inbox / '261003-intelligenceRadarDaily_chatgpt.txt').write_bytes(b' \n') (self.inbox / 'readme.txt').write_text('ignore', encoding='utf-8') (self.inbox / 'archive' / '260919-intelligenceRadarDaily_grok.txt').write_text( 'already captured\n', encoding='utf-8') first = import_inbox(self.root, self.inbox, now=self.now) self.assertTrue(first['ok']) retained = [item for item in first['files'] if item['status'] == 'retained'] self.assertEqual(len(retained), 2) self.assertEqual({item['status'] for item in first['files']}, {'retained', 'empty', 'ignored'}) self.assertFalse((self.inbox / '261003-intelligenceRadarDaily_grok.txt').exists()) self.assertTrue((self.inbox / 'archive' / '261003-intelligenceRadarDaily_grok.txt').is_file()) self.assertEqual(verify(self.root), 2) text = (self.root / retained[0]['digest_path']).read_text(encoding='utf-8') self.assertIn('Unverified', text) self.assertIn(retained[0]['content_sha256'], text) second = import_inbox(self.root, self.inbox, now=self.now) self.assertTrue(all(item['status'] != 'retained' for item in second['files'])) self.assertEqual(verify(self.root), 2) def test_same_period_keeps_both_providers_and_draft_lists_them(self): (self.inbox / '261003-intelligenceRadarDaily_grok.txt').write_text('grok\n', encoding='utf-8') (self.inbox / '261003-intelligenceRadarDaily_chatgpt.txt').write_text('chatgpt\n', encoding='utf-8') import_inbox(self.root, self.inbox, now=self.now) text = draft(self.root, '2026-10-03').read_text(encoding='utf-8') self.assertIn('External automation briefs', text) self.assertIn('chatgpt', text) self.assertIn('grok', text) self.assertIn('does not establish that no events occurred', text) def test_archive_name_collision_keeps_both_bytes(self): name = '261003-intelligenceRadarDaily_grok.txt' (self.inbox / 'archive' / name).write_text('old\n', encoding='utf-8') (self.inbox / name).write_text('new\n', encoding='utf-8') result = import_inbox(self.root, self.inbox, now=self.now) retained = result['files'][0] self.assertEqual(retained['status'], 'retained') self.assertEqual(retained['archive'], 'left_in_inbox') self.assertEqual((self.inbox / name).read_text(encoding='utf-8'), 'new\n') self.assertEqual((self.inbox / 'archive' / name).read_text(encoding='utf-8'), 'old\n') def test_unreadable_file_fails_the_import_and_stays_put(self): (self.inbox / '261003-intelligenceRadarDaily_grok.txt').write_bytes(b'\xff') result = import_inbox(self.root, self.inbox, now=self.now) self.assertFalse(result['ok']) self.assertEqual(result['files'][0]['status'], 'unreadable') self.assertTrue((self.inbox / '261003-intelligenceRadarDaily_grok.txt').is_file()) def test_tampered_artifact_fails_verification(self): (self.inbox / '261003-intelligenceRadarDaily_grok.txt').write_text('grok\n', encoding='utf-8') result = import_inbox(self.root, self.inbox, now=self.now) digest = result['files'][0]['content_sha256'] (self.root / 'data' / 'artifacts' / digest).write_bytes(b'tampered') with self.assertRaisesRegex(ValueError, 'artifact mismatch'): verify(self.root) def test_native_pull_does_not_claim_a_feed(self): report = native_status() self.assertTrue(report['ok']) self.assertEqual({item['provider'] for item in report['providers']}, {'chatgpt', 'grok'}) self.assertTrue(all(item['status'] == 'needs_session' for item in report['providers'])) def test_due_job_imports_inbox_without_blocking_source_capture(self): (self.inbox / '261003-intelligenceRadarDaily_grok.txt').write_text('grok\n', encoding='utf-8') (self.root / 'sources.json').write_text('{"sources": []}') config = {'timezone': 'Europe/Berlin', 'external_briefs': {'inbox': str(self.inbox)}, 'jobs': [{'id': 'daily', 'cadence': 'daily', 'at': '21:30', 'sources': 'sources.json'}]} result = run_due(self.root, config, datetime(2026, 10, 3, 20, tzinfo=timezone.utc), runner=lambda *args: {'ok': True}) self.assertTrue(result['ok']) self.assertEqual(result['external_briefs']['files'][0]['status'], 'retained') self.assertEqual(result['external_briefs']['native']['status'], 'needs_session') self.assertEqual(verify(self.root), 1) def test_repository_check_accepts_retained_brief(self): for name in ('INTENT.md', 'GOAL.md', 'SCOPE.md', 'AGENTS.md', 'README.md', 'TaskModelEvaluationMethod.md', 'ops/RUNBOOK.md'): path = self.root / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text('present\n') (self.root / 'task-router').mkdir() (self.root / 'task-router/METHOD.md').write_text('present\n') (self.root / 'policies').mkdir() (self.root / 'policies/model-ranking-v0.1.yaml').write_text('present\n') (self.root / 'policies/benchmark-selection-v0.1.yaml').write_text('present\n') (self.root / 'history').mkdir() (self.root / 'history/MANIFEST.json').write_text('{"archives": []}\n') (self.inbox / '261003-intelligenceRadarDaily_grok.txt').write_text('grok\n', encoding='utf-8') import_inbox(self.root, self.inbox, now=self.now) self.assertEqual(check(self.root), 0) digest = hashlib.sha256(b'grok\n').hexdigest() self.assertTrue((self.root / 'data' / 'artifacts' / digest).is_file()) if __name__ == '__main__': unittest.main()