"""The Radar (projects/radar.py, projects/papers.py) with a pretend Claude.""" import json from datetime import datetime from types import SimpleNamespace as NS from unittest.mock import patch from zoneinfo import ZoneInfo from django.contrib.auth.models import User from django.test import TestCase from . import papers, radar from .models import Project, Radar from .radar_presets import NEUROPOLITIK BERLIN = ZoneInfo('Europe/Berlin') REC = [ {'id': 'pmid:1', 'doi': '10.1/a', 'source': 'PubMed', 'topic': 'technik', 'title': 'Speech BCI', 'journal': 'Nature', 'first_author': 'Willett', 'n_authors': 9, 'pub_types': [], 'date': '2026', 'url': 'https://pubmed.ncbi.nlm.nih.gov/1/', 'abstract': 'A speech neuroprosthesis.'}, {'id': 'arxiv:2610.1', 'doi': None, 'source': 'arXiv', 'topic': None, 'title': 'EEG noise', 'journal': 'arXiv eess.SP', 'first_author': 'A', 'n_authors': 2, 'pub_types': ['Preprint'], 'date': '2026-10-18', 'url': 'https://arxiv.org/abs/2610.1', 'abstract': 'EEG.'}, ] def message(text, stop='end_turn', searches=0): return NS(content=[NS(type='text', text=text)], stop_reason=stop, usage=NS(input_tokens=1000, output_tokens=200, cache_read_input_tokens=0, cache_creation_input_tokens=0, server_tool_use=NS(web_search_requests=searches) if searches else None)) class FakeClient: def __init__(self, status='ended', papers_text=None): self.created = [] self.status = status self.papers_text = papers_text or json.dumps({ 'auf_einen_blick': 'Eine ruhige Woche.', 'grundlagenbegriff': {'begriff': 'Decoder', 'erklaerung': 'Ein Modell, das …'}, 'items': [{'id': 'pmid:1', 'gruppe': 'lesenswert', 'worum_es_geht': 'Sprache.', 'einordnung': 'Peer Review.', 'politischer_haken': 'Stimme.', 'kurz': 'Sprache aus dem Kortex.'}, {'id': 'erfunden', 'gruppe': 'technik', 'worum_es_geht': '', 'einordnung': '', 'politischer_haken': '', 'kurz': 'x'}]}) outer = self class Batches: def create(self, requests): outer.created.append(requests) return NS(id='msgbatch_1', processing_status='in_progress') def retrieve(self, bid): return NS(id=bid, processing_status=outer.status) def results(self, bid): return [NS(custom_id='papers', result=NS(type='succeeded', message=message(outer.papers_text))), NS(custom_id='news', result=NS(type='succeeded', message=message('- **X:** y. [Q](https://q)', searches=2)))] self.messages = NS(batches=Batches()) class RadarTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='radar', password='pw12345678') self.client.force_login(self.user) self.project = Project.objects.create(user=self.user, data_ct='x', data_iv='a1' * 12) self.radar = Radar.objects.create(project=self.project, config=NEUROPOLITIK) def test_due_on_its_weekday_once_a_week(self): sat = datetime(2026, 10, 24, 20, 0, tzinfo=BERLIN) sun_early = datetime(2026, 10, 25, 17, 0, tzinfo=BERLIN) sun = datetime(2026, 10, 25, 18, 5, tzinfo=BERLIN) self.radar.last_run_at = datetime(2026, 10, 18, 18, 1, tzinfo=BERLIN) self.assertFalse(radar.due(self.radar, sat)) self.assertFalse(radar.due(self.radar, sun_early)) self.assertTrue(radar.due(self.radar, sun)) self.radar.last_run_at = sun self.assertFalse(radar.due(self.radar, datetime(2026, 10, 25, 23, 0, tzinfo=BERLIN))) self.radar.last_run_at = None self.assertTrue(radar.due(self.radar, sat), 'never run: at once') def test_submit_and_collect(self): fake = FakeClient(status='in_progress') with patch.object(radar, '_client', return_value=fake), \ patch.object(papers, 'collect', return_value=(list(REC), ['medrxiv: down'])): self.assertIn('2 Papers', radar.submit(self.radar)) reqs = fake.created[0] self.assertEqual([r['custom_id'] for r in reqs], ['papers', 'news']) self.assertIn('Speech BCI', reqs[0]['params']['messages'][0]['content']) self.assertEqual(radar.collect(self.radar), 'waiting') fake.status = 'ended' self.assertEqual(radar.collect(self.radar), 'done') self.radar.refresh_from_db() self.assertIsNone(self.radar.pending) self.assertEqual(set(self.radar.seen), {'pmid:1', 'arxiv:2610.1'}) b = self.radar.digests.get().bundle self.assertEqual({i['id']: i['gruppe'] for i in b['items']}, {'pmid:1': 'lesenswert', 'arxiv:2610.1': 'technik'}) self.assertEqual(b['items'][0]['title'], 'Speech BCI', 'titles from the sources, not the model') self.assertEqual(b['stats']['usage'], {'input_tokens': 2000, 'output_tokens': 400, 'web_search_requests': 2}) self.assertEqual(b['stats']['errors'], ['medrxiv: down']) # The next week the basic term isn't asked for again. self.assertIn('Decoder', radar.build_requests(NEUROPOLITIK, REC, radar._terms(self.radar), '2026-10-26')[0] ['params']['messages'][0]['content']) def test_a_failed_sorting_forgets_nothing(self): fake = FakeClient(papers_text='kein json') self.radar.pending = {'batch_id': 'b', 'generated': '2026-10-19', 'days': 7, 'records': REC, 'errors': []} self.radar.save() with patch.object(radar, '_client', return_value=fake): self.assertIn('fehlgeschlagen', radar.collect(self.radar)) self.radar.refresh_from_db() self.assertEqual((self.radar.seen, self.radar.pending, self.radar.digests.count()), ([], None, 0)) def test_nothing_new_no_batch(self): fake = FakeClient() with patch.object(radar, '_client', return_value=fake), patch.object(papers, 'collect', return_value=([], [])): self.assertEqual(radar.submit(self.radar), 'Keine neuen Papers.') self.assertEqual(fake.created, []) def test_api_view_run_import(self): r = self.client.get(f'/projects/{self.project.id}/radar/').json() self.assertEqual((r['radar']['title'], r['digests']), ('Neuropolitik-Digest', [])) bundle = {'version': 1, 'generated': '2026-10-19', 'items': [{'id': 'pmid:9', 'gruppe': 'technik'}], 'batch_id': 'msgbatch_x'} self.client.post(f'/projects/{self.project.id}/radar/import/', json.dumps(bundle), content_type='application/json') self.client.post(f'/projects/{self.project.id}/radar/import/', json.dumps(bundle), content_type='application/json') r = self.client.get(f'/projects/{self.project.id}/radar/').json() self.assertEqual(len(r['digests']), 1, 'the same batch once') self.assertIn('pmid:9', Radar.objects.get().seen) self.assertEqual(self.client.post(f'/projects/{self.project.id}/radar/import/', '{}', content_type='application/json').status_code, 400) r = self.client.post(f'/projects/{self.project.id}/radar/run/').json() self.assertTrue(radar.due(Radar.objects.get())) other = User.objects.create_user(username='x', password='pw12345678') self.client.force_login(other) self.assertEqual(self.client.get(f'/projects/{self.project.id}/radar/').status_code, 404) class PapersTests(TestCase): def test_collect_leaves_out_doubles_and_seen(self): dup = dict(REC[0], id='doi:10.1/a', source='biorxiv') with patch.object(papers, 'fetch_pubmed', return_value=[REC[0]]), \ patch.object(papers, 'fetch_arxiv', return_value=[REC[1]]), \ patch.object(papers, 'fetch_rxiv', side_effect=[[dup], OSError('down')]): got, errors = papers.collect(NEUROPOLITIK, 7, seen={'arxiv:2610.1'}) self.assertEqual([r['id'] for r in got], ['pmid:1']) self.assertEqual(errors, ['medrxiv: down'])