tegwick
b1a65e7214
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 7780@bnt-lap001 Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
156 lines
7.5 KiB
Python
156 lines
7.5 KiB
Python
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 <noreply@x.ai>
|
|
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
|
|
<https://grok.com/chat/2eae1664-9d14-4fd2-87d2-d78a5cca73a2>
|
|
'''
|
|
|
|
|
|
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()
|