tegwick
29ceb77938
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09c77-b1a9-7eb3-b501-6cb24b458c97
67 lines
3.1 KiB
Python
67 lines
3.1 KiB
Python
from datetime import datetime, timezone
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from intelligence_radar.ingest import capture, check_artifacts, MAX_BYTES
|
|
from intelligence_radar.scheduler import run_due
|
|
from intelligence_radar.store import all_records
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
headers={'Content-Type':'text/plain'}
|
|
def geturl(self):
|
|
return 'https://example.invalid/source'
|
|
|
|
|
|
class IngestionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root=Path(self.temp.name)
|
|
self.source={'id':'test','publisher':'test','url':'https://example.invalid/source'}
|
|
|
|
def test_capture_deduplication_and_integrity(self):
|
|
first,r=capture(self.root,self.source,opener=lambda *a,**k:Response(b'body'))
|
|
second,s=capture(self.root,self.source,opener=lambda *a,**k:Response(b'body'))
|
|
self.assertTrue(r['changed'])
|
|
self.assertFalse(s['changed'])
|
|
self.assertEqual(len(list((self.root/'data/artifacts').iterdir())),1)
|
|
self.assertEqual(check_artifacts(self.root),2)
|
|
(self.root/'data/artifacts'/r['content_sha256']).write_bytes(b'tampered')
|
|
with self.assertRaises(ValueError):
|
|
check_artifacts(self.root)
|
|
|
|
def test_retry_and_capture_limit(self):
|
|
with patch('intelligence_radar.ingest.time.sleep'), patch('urllib.request.urlopen'):
|
|
count=[0]
|
|
def fetch(*args,**kwargs):
|
|
count[0]+=1
|
|
if count[0]<3:
|
|
raise OSError('transient')
|
|
return Response(b'ok')
|
|
_,r=capture(self.root,self.source,opener=fetch)
|
|
self.assertEqual(r['attempts'],3)
|
|
with self.assertRaises(ValueError):
|
|
capture(self.root,self.source,opener=lambda *a,**k:Response(b'x'*(MAX_BYTES+1)))
|
|
|
|
def test_scheduler_retries_failed_job_and_preserves_existing_digest(self):
|
|
(self.root/'sources.json').write_text('{"sources": []}')
|
|
config={'timezone':'Europe/Berlin','jobs':[{'id':'daily','cadence':'daily','at':'21:30','sources':'sources.json'}]}
|
|
now=datetime(2026,9,13,20,tzinfo=timezone.utc)
|
|
failed=run_due(self.root,config,now,runner=lambda *a:{'ok':False})
|
|
self.assertFalse(failed['ok'])
|
|
passed=run_due(self.root,config,now,runner=lambda *a:{'ok':True})
|
|
self.assertTrue(passed['ok'])
|
|
self.assertEqual(len(passed['jobs']),1)
|
|
self.assertEqual(run_due(self.root,config,now,runner=lambda *a:self.fail('already done'))['jobs'],[])
|
|
self.assertEqual(len(all_records(self.root,'scheduler-runs')),2)
|
|
|
|
def test_weekly_iso_boundary_and_dst_zone(self):
|
|
(self.root/'sources.json').write_text('{"sources": []}')
|
|
config={'timezone':'Europe/Berlin','jobs':[{'id':'weekly','cadence':'weekly','weekday':6,'at':'20:00','sources':'sources.json'}]}
|
|
result=run_due(self.root,config,datetime(2026,1,4,19,tzinfo=timezone.utc),runner=lambda *a:{'ok':True})
|
|
self.assertEqual(result['jobs'][0]['period'],'2026-W01')
|