coulomb/inbound-library
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
inbound-library/tests/test_library.py
tegwick c810fe6dfe Add AI topics, a sources discovery command, and intelligence-radar's pointer sources. 
Topics ai.benchmarks, ai.pricing and ai.models; 'sources --repo' lists
visible sources matching a repo's topics. Registers the five ir-* URL
sources contributed by intelligence-radar (IR-WP-0010-T04).

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
2026-10-06 00:31:51 +02:00

236 lines
13 KiB
Python
Raw Permalink Blame History

import json
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
from pathlib import Path
from inbound_library import intake, notify, store
from inbound_library import registry as reg
from inbound_library.__main__ import main, run_due
def mail(sender, subject, body, attachment=None, message_id=None):
message = EmailMessage()
message['From'] = sender
message['To'] = 'library@coulomb.social'
message['Subject'] = subject
message['Message-ID'] = message_id or f'<{abs(hash((sender, subject, body)))}@test>'
message['Authentication-Results'] = 'mx.test; dkim=pass'
message.set_content(body)
if attachment:
message.add_attachment(attachment, maintype='text', subtype='plain', filename='brief.txt')
return message.as_bytes()
def write(path, value):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value))
class Library(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name)
self.mailbox_dir = self.root / 'fixture-mail'
self.mailbox_dir.mkdir()
write(self.root / 'registry' / 'topics.json',
{'version': 1, 'topics': {'intelligence.daily-brief': 'Daily intelligence brief',
'research.papers': 'Research papers'}})
write(self.root / 'registry' / 'sources' / 'grok-daily.json', {
'id': 'grok-daily', 'owner_repo': 'intelligence-radar', 'channel': 'email',
'match': {'from': 'noreply@x.ai'}, 'topics': ['intelligence.daily-brief'],
'cadence': 'daily', 'visibility': 'shared'})
write(self.root / 'registry' / 'subscriptions' / 'intelligence-radar.json', {
'repo': 'intelligence-radar', 'agent': 'intelligence-radar',
'topics': ['intelligence.daily-brief'], 'channels': ['hub', 'pull', 'digest'],
'cadence': 'daily'})
self.posts = []
def tearDown(self):
self.tmp.cleanup()
def drop(self, name, raw):
(self.mailbox_dir / name).write_bytes(raw)
def mailbox(self, uidvalidity=1):
return intake.DirectoryMailbox(self.mailbox_dir, uidvalidity)
def post(self, api_base, payload):
self.posts.append(payload)
# Store
def test_item_is_never_overwritten_and_duplicates_collapse(self):
first, new = store.put_item(self.root, b'brief')
again, new_again = store.put_item(self.root, b'brief')
self.assertEqual(first, again)
self.assertTrue(new)
self.assertFalse(new_again)
with self.assertRaises(ValueError):
store.atomic(self.root / 'data' / 'items' / first, b'other', immutable=True)
def test_tampering_is_detected(self):
ident, _ = store.put_item(self.root, b'brief')
(self.root / 'data' / 'items' / ident).write_bytes(b'changed')
self.assertEqual(store.verify(self.root), [f'item integrity mismatch: {ident}'])
# Registry
def test_registry_rejects_unknown_topic_and_restricted_leak(self):
write(self.root / 'registry' / 'sources' / 'private.json', {
'id': 'private', 'owner_repo': 'markitect', 'channel': 'url', 'url': 'https://x',
'topics': ['nope'], 'cadence': 'irregular', 'visibility': 'restricted'})
write(self.root / 'registry' / 'subscriptions' / 'other.json', {
'repo': 'other', 'agent': 'other', 'topics': [], 'sources': ['private'],
'channels': ['pull']})
problems = reg.validate(reg.load(self.root))
self.assertIn('sources/private.json: unknown topic nope', problems)
self.assertTrue(any('restricted source private belongs to markitect' in p for p in problems))
def test_source_from_one_repo_reaches_another_by_topic(self):
write(self.root / 'registry' / 'sources' / 'arxiv-agents.json', {
'id': 'arxiv-agents', 'owner_repo': 'intelligence-radar', 'channel': 'url',
'url': 'https://arxiv.org/list/cs.MA/recent', 'topics': ['research.papers'],
'cadence': 'irregular', 'visibility': 'shared'})
write(self.root / 'registry' / 'subscriptions' / 'markitect.json', {
'repo': 'markitect', 'agent': 'markitect', 'topics': ['research.papers'],
'channels': ['pull']})
registry = reg.load(self.root)
self.assertEqual(reg.validate(registry), [])
self.assertEqual(reg.subscribers_for(registry, 'arxiv-agents'), ['markitect'])
def test_sources_lists_visible_matches_only(self):
write(self.root / 'registry' / 'sources' / 'arxiv-agents.json', {
'id': 'arxiv-agents', 'owner_repo': 'intelligence-radar', 'channel': 'url',
'url': 'https://arxiv.org/list/cs.MA/recent', 'topics': ['research.papers'],
'cadence': 'irregular', 'visibility': 'shared'})
write(self.root / 'registry' / 'sources' / 'private.json', {
'id': 'private', 'owner_repo': 'intelligence-radar', 'channel': 'url',
'url': 'https://example.org', 'topics': ['research.papers'],
'cadence': 'irregular', 'visibility': 'restricted'})
write(self.root / 'registry' / 'subscriptions' / 'markitect.json', {
'repo': 'markitect', 'agent': 'markitect', 'topics': ['research.papers'],
'channels': ['pull']})
registry = reg.load(self.root)
self.assertEqual([s['id'] for s in reg.sources_for(registry, 'markitect')], ['arxiv-agents'])
self.assertEqual([s['id'] for s in reg.sources_for(registry, 'intelligence-radar',
'research.papers')],
['arxiv-agents', 'private'])
# Intake
def test_scan_retains_message_attachment_and_provenance(self):
self.drop('1.eml', mail('Grok <noreply@x.ai>', 'Daily brief', 'Body text', b'attached'))
[delivery_id] = intake.scan(self.root, self.mailbox())
delivery = store.get_record(self.root, 'deliveries', delivery_id)
self.assertEqual(delivery['source'], 'grok-daily')
self.assertEqual(delivery['status'], 'routed')
self.assertEqual(delivery['authentication_results'], ['mx.test; dkim=pass'])
raw = store.get_item(self.root, delivery['item'])
self.assertEqual(raw, (self.mailbox_dir / '1.eml').read_bytes())
derived = store.get_derived(self.root, delivery['item'])
self.assertIn('Body text', derived['text'])
[attachment] = derived['attachments']
self.assertEqual(store.get_item(self.root, attachment['item']), b'attached')
def test_cursor_skips_seen_mail_and_uidvalidity_change_does_not_duplicate(self):
self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one'))
self.assertEqual(len(intake.scan(self.root, self.mailbox())), 1)
self.assertEqual(intake.scan(self.root, self.mailbox()), [])
rescanned = intake.scan(self.root, self.mailbox(uidvalidity=2))
self.assertEqual(len(rescanned), 1)
self.assertEqual(len(store.records(self.root, 'deliveries')), 1)
def test_unknown_sender_is_quarantined_then_requeued(self):
self.drop('1.eml', mail('someone@example.org', 'Hello', 'spam?'))
[delivery_id] = intake.scan(self.root, self.mailbox())
self.assertEqual(store.get_record(self.root, 'deliveries', delivery_id)['status'], 'quarantined')
self.assertEqual(notify.match(self.root), [])
self.assertEqual(notify.health(self.root)['quarantined_items'], 1)
write(self.root / 'registry' / 'sources' / 'someone.json', {
'id': 'someone', 'owner_repo': 'intelligence-radar', 'channel': 'email',
'match': {'from': 'someone@example.org'}, 'topics': ['intelligence.daily-brief'],
'cadence': 'irregular', 'visibility': 'shared'})
self.assertEqual(len(intake.requeue_quarantine(self.root)), 1)
self.assertEqual(notify.health(self.root)['quarantined_items'], 0)
self.assertEqual(len(notify.match(self.root)), 1)
def test_forwarded_mail_routes_only_with_matching_original_sender(self):
write(self.root / 'registry' / 'sources' / 'grok-via-founder.json', {
'id': 'grok-via-founder', 'owner_repo': 'intelligence-radar', 'channel': 'email',
'match': {'from': 'founder@example.org', 'forwarded_from': 'noreply@x.ai'},
'topics': ['intelligence.daily-brief'], 'cadence': 'daily', 'visibility': 'shared'})
grok = ('Hi\n\n---------- Forwarded message ---------\nVon: Grok <noreply@x.ai>\n'
'Subject: Brief\n\nbody')
other = ('---------- Forwarded message ---------\nFrom: Someone <a@b.c>\n\nbody')
self.drop('1.eml', mail('Founder <founder@example.org>', 'Fwd: Brief', grok))
self.drop('2.eml', mail('Founder <founder@example.org>', 'Fwd: Other', other))
first, second = intake.scan(self.root, self.mailbox())
one = store.get_record(self.root, 'deliveries', first)
two = store.get_record(self.root, 'deliveries', second)
self.assertEqual((one['source'], one['forwarded_from']), ('grok-via-founder', 'noreply@x.ai'))
self.assertEqual((two['status'], two['forwarded_from']), ('quarantined', 'a@b.c'))
def test_live_mailbox_needs_credentials_from_environment(self):
self.assertIsNone(intake.mailbox_from_config({}, environ={}))
box = intake.mailbox_from_config({}, environ={'INBOUND_LIBRARY_IMAP_USER': 'u',
'INBOUND_LIBRARY_IMAP_PASSWORD': 'p'})
self.assertEqual((box.host, box.port, box.folder), ('imap.ionos.de', 993, 'INBOX'))
# Notices and channels
def test_notices_are_idempotent_and_hub_outage_stays_pending(self):
self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one'))
intake.scan(self.root, self.mailbox())
self.assertEqual(len(notify.match(self.root)), 1)
self.assertEqual(notify.match(self.root), [])
def down(api_base, payload):
raise OSError('connection refused')
report = notify.deliver_hub(self.root, 'http://hub', post=down)
self.assertEqual(report['intelligence-radar']['pending'], 1)
report = notify.deliver_hub(self.root, 'http://hub', post=self.post)
self.assertEqual(report['intelligence-radar'], {'ok': True, 'sent': 1})
self.assertEqual(notify.deliver_hub(self.root, 'http://hub', post=self.post), {})
[message] = self.posts
self.assertEqual(message['to_agent'], 'intelligence-radar')
self.assertIn('Daily brief', message['body'])
def test_pull_and_digest(self):
self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one'))
intake.scan(self.root, self.mailbox())
notify.match(self.root)
[notice] = notify.notices_for(self.root, 'intelligence-radar',
topic='intelligence.daily-brief')
self.assertEqual(notice['subject'], 'Daily brief')
self.assertEqual(notify.notices_for(self.root, 'intelligence-radar', topic='other'), [])
now = datetime.fromisoformat(notice['received_at'])
[path] = notify.write_digests(self.root, now=now)
self.assertIn('Daily brief', (self.root / path).read_text())
def test_silent_source_is_reported(self):
later = datetime.now(timezone.utc) + timedelta(days=3)
self.assertEqual(notify.health(self.root, now=later)['silent_sources'],
[{'source': 'grok-daily', 'last_seen': None}])
self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one'))
intake.scan(self.root, self.mailbox())
self.assertEqual(notify.health(self.root)['silent_sources'], [])
self.assertEqual(len(notify.health(self.root, now=later)['silent_sources']), 1)
def test_run_due_end_to_end_and_repeatable(self):
self.drop('1.eml', mail('noreply@x.ai', 'Daily brief', 'one'))
report = run_due(self.root, {}, mailbox=self.mailbox(), post=self.post)
self.assertTrue(report['ok'], report)
self.assertEqual((report['intake']['retained'], report['notices']), (1, 1))
report = run_due(self.root, {}, mailbox=self.mailbox(), post=self.post)
self.assertEqual((report['intake']['retained'], report['notices'], report['hub']), (0, 0, {}))
self.assertEqual(len(self.posts), 1)
def test_run_due_without_credentials_still_notifies(self):
report = run_due(self.root, {}, post=self.post)
self.assertFalse(report['ok'])
self.assertEqual(report['intake']['status'], 'needs_credentials')
def test_cli_check(self):
self.assertEqual(main(['--root', str(self.root), 'check']), 0)
if __name__ == '__main__':
unittest.main()
Reference in a new issue View git blame Copy permalink