tegwick
8249c34100
IR-WP-0008 imports those drop-folder exports on the due-job timer and records that website automation runs still have no public feed. Assistant: grok Assistant-Session: 01a106a5-fc07-78c2-b64f-8e04cea25daf
134 lines
7.3 KiB
Python
134 lines
7.3 KiB
Python
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()
|