import hashlib import json import tempfile import unittest from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo from intelligence_radar import external_briefs, library_briefs NOW = datetime(2026, 10, 6, 9, 0, tzinfo=ZoneInfo('Europe/Berlin')) SPEC = {'root': '/nowhere', 'repo': 'intelligence-radar', 'sources': {'grok-briefs-via-founder': 'grok'}} TEASER = '''---------- Forwarded message --------- Von: Grok Date: So., 4. Okt. 2026 um 20:07 Uhr Subject: Claude Sonnet 5.5 & GPT-6.1 Sol launch Automation · intelligence-radar-weekly I'll compile the ISO week ending today ... Fairwind Progra ... Continue reading ''' class FakeLibrary: def __init__(self, items, down=False): self.items, self.down, self.calls = items, down, [] def __call__(self, spec, command, *args): self.calls.append((command, *args)) if self.down: raise library_briefs.LibraryUnavailable('inbound-library not found at /nowhere') if command == 'new': return json.dumps([{'item': i, 'source': v['source'], 'received_at': v['received_at']} for i, v in self.items.items()]).encode() if command == 'show': v = self.items[args[0]] return json.dumps({'derived': {'subject': v['subject'], 'text': v['text']}}).encode() if command == 'get': return self.items[args[0]]['raw'] raise AssertionError(command) def item(text, subject='Fwd: Brief', source='grok-briefs-via-founder', received_at='2026-10-05T13:05:06+00:00'): raw = ('Subject: ' + subject + '\n\n' + text).encode() return hashlib.sha256(raw).hexdigest(), {'raw': raw, 'text': text, 'subject': subject, 'source': source, 'received_at': received_at} class LibraryBriefs(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) def tearDown(self): self.tmp.cleanup() def test_original_date_german_and_english(self): self.assertEqual(str(library_briefs.original_date(TEASER)), '2026-10-04') english = TEASER.replace('Date: So., 4. Okt. 2026 um 20:07 Uhr', 'Date: Sun, Oct 4, 2026 at 8:07 PM') self.assertEqual(str(library_briefs.original_date(english)), '2026-10-04') def test_no_notices(self): result = library_briefs.pull(self.root, SPEC, cli=FakeLibrary({}), now=NOW) self.assertEqual(result, {'ok': True, 'status': 'pulled', 'items': []}) def test_teaser_retained_as_partial_weekly_brief(self): ident, value = item(TEASER) result = library_briefs.pull(self.root, SPEC, cli=FakeLibrary({ident: value}), now=NOW) [retained] = result['items'] self.assertEqual((retained['status'], retained['cadence'], retained['period'], retained['completeness']), ('retained', 'weekly', '2026-W40', 'partial')) self.assertEqual((self.root / 'data' / 'artifacts' / ident).read_bytes(), value['raw']) markdown = (self.root / retained['digest_path']).read_text() self.assertIn('Partial: mailed teaser', markdown) self.assertIn('https://grok.com/chat/2eae1664', markdown) self.assertNotIn('2026-W40-grok.md', retained['digest_path']) [brief] = external_briefs.briefs_for_period(self.root, '2026-W40') self.assertEqual((brief['schema_version'], brief['period_basis']), (2, 'forwarded_date')) self.assertEqual(external_briefs.verify(self.root), 1) def test_item_seen_twice_is_a_duplicate(self): ident, value = item(TEASER) library = FakeLibrary({ident: value}) library_briefs.pull(self.root, SPEC, cli=library, now=NOW) result = library_briefs.pull(self.root, SPEC, cli=library, now=NOW) self.assertEqual(result['items'], [{'item': ident, 'status': 'duplicate'}]) self.assertEqual(len(list((self.root / 'data' / 'external-briefs').iterdir())), 1) def test_unreachable_library_fails_only_this_part(self): result = library_briefs.pull(self.root, SPEC, cli=FakeLibrary({}, down=True), now=NOW) self.assertEqual((result['ok'], result['status']), (False, 'library_unavailable')) def test_unconfigured_source_and_non_brief_are_skipped(self): other, other_value = item(TEASER, source='someone-else') plain, plain_value = item('Just a forwarded note.', subject='Fwd: hello') result = library_briefs.pull(self.root, SPEC, now=NOW, cli=FakeLibrary({other: other_value, plain: plain_value})) self.assertEqual(result['items'], [{'item': plain, 'status': 'not_a_brief'}]) def test_tampered_bytes_are_refused(self): ident, value = item(TEASER) value['raw'] = b'changed' result = library_briefs.pull(self.root, SPEC, cli=FakeLibrary({ident: value}), now=NOW) self.assertFalse(result['ok']) self.assertEqual(result['items'][0]['status'], 'error') def test_daily_period_falls_back_to_received_date(self): text = 'Automation · intelligence-radar-daily\nTonight\'s brief is complete.\n' ident, value = item(text, received_at='2026-10-05T22:30:00+00:00') [retained] = library_briefs.pull(self.root, SPEC, cli=FakeLibrary({ident: value}), now=NOW)['items'] self.assertEqual((retained['cadence'], retained['period'], retained['completeness']), ('daily', '2026-10-06', 'complete')) class ContributeSources(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) / 'radar' self.library = Path(self.tmp.name) / 'library' (self.root / 'catalog').mkdir(parents=True) (self.library / 'registry' / 'sources').mkdir(parents=True) (self.root / 'catalog' / 'sources.json').write_text(json.dumps({'sources': [ {'id': 'swe-bench', 'publisher': 'SWE-bench maintainers', 'url': 'https://www.swebench.com/'}]})) self.spec = {'root': str(self.library), 'repo': 'intelligence-radar', 'contribute': {'swe-bench': ['ai.benchmarks']}} def tearDown(self): self.tmp.cleanup() def test_create_then_unchanged_and_pointer_shape(self): self.assertEqual(library_briefs.contribute_sources(self.root, self.spec), [{'source': 'ir-swe-bench', 'status': 'created'}]) self.assertEqual(library_briefs.contribute_sources(self.root, self.spec), [{'source': 'ir-swe-bench', 'status': 'unchanged'}]) source = json.loads((self.library / 'registry' / 'sources' / 'ir-swe-bench.json').read_text()) self.assertEqual((source['channel'], source['url'], source['visibility'], source['topics']), ('url', 'https://www.swebench.com/', 'shared', ['ai.benchmarks'])) def test_refuses_another_repos_file_and_unknown_source(self): (self.library / 'registry' / 'sources' / 'ir-swe-bench.json').write_text( json.dumps({'id': 'ir-swe-bench', 'owner_repo': 'markitect'})) with self.assertRaises(ValueError): library_briefs.contribute_sources(self.root, self.spec) with self.assertRaises(ValueError): library_briefs.contribute_sources(self.root, {**self.spec, 'contribute': {'nope': []}}) if __name__ == '__main__': unittest.main()