intelligence-radar/tests/test_library_briefs.py

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()