144 lines
7.9 KiB
Python
144 lines
7.9 KiB
Python
"""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'])
|