"""Shared fakes and payload builders for the orchestration tests.""" from __future__ import annotations from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo from sebrief.models import ( Digest, Highlight, Paper, PaperSummary, PaperSummaryItem, Pick, TopicGroup, TopicPaper, WeekPayload, ) KST = ZoneInfo("Asia/Seoul") # Fixed test timeline: 2026-W36 = 2026-08-31 (Mon) .. 2026-09-06 (Sun); the covering run fires # Monday 2026-09-07 07:00 KST. NOW is that Monday at noon. WEEK = "2026-W36" WEEK_START = date(2026, 8, 31) WEEK_END = date(2026, 9, 6) RUN_AT = datetime(2026, 9, 7, 7, 0, tzinfo=KST) NOW = datetime(2026, 9, 7, 12, 0, tzinfo=KST) def make_paper(doi: str = "10.1145/1234567", **overrides) -> Paper: values = { "doi": doi, "title": "Test-Driven Digest Generation", "abstract": "We study how weekly digests are generated. Results show 42% faster reading.", "authors": ["Alice Kim", "Bob Lee"], "venue": "ACM Transactions on Software Engineering and Methodology", "published_date": date(2026, 9, 2), "oa_url": "https://dl.acm.org/doi/10.1145/1234567", "pdf_url": None, "topics": ["Software Engineering Research"], "source": "openalex", } values.update(overrides) return Paper(**values) def make_summary(**overrides) -> PaperSummary: values = { "tldr": "테스트 주도 다이제스트 생성 기법을 제안합니다.", "problem": "주간 다이제스트 생성이 느립니다.", "approach": "저자들은 테스트 주도 생성 기법을 제안합니다.", "findings": "저자들은 읽기 속도가 42% 빨라졌다고 보고합니다.", "implication": "다이제스트 파이프라인에 적용해볼 만합니다.", } values.update(overrides) return PaperSummary(**values) def make_payload(papers: list[Paper] | None = None, gaps: list[str] | None = None) -> WeekPayload: return WeekPayload( week=WEEK, week_start=WEEK_START, week_end=WEEK_END, generated_at=NOW, papers=papers if papers is not None else [make_paper()], data_gaps=gaps or [], ) def make_digest(doi: str = "10.1145/1234567", **overrides) -> Digest: values = { "overview": "이번 주는 다이제스트 생성 연구가 눈에 띕니다.", "highlights": [ Highlight( doi=doi, title="Test-Driven Digest Generation", problem="주간 다이제스트 생성이 느립니다.", approach="저자들은 테스트 주도 생성 기법을 제안합니다.", findings="저자들은 읽기 속도가 42% 빨라졌다고 보고합니다.", implication="다이제스트 파이프라인에 적용해볼 만합니다.", ) ], "by_topic": [ TopicGroup( topic="도구·데이터셋", papers=[TopicPaper(doi=doi, title="Test-Driven Digest Generation", note="다이제스트 생성 도구 연구입니다.")], ) ], "worth_reading": [Pick(doi=doi, title="Test-Driven Digest Generation", reason="방법론이 참신합니다.")], "data_gaps": [], } values.update(overrides) return Digest(**values) class FakeClock: """Deterministic stand-in for sebrief.week.WeeklyClock.""" def __init__( self, *, catchup: str | None = None, next_run: tuple[str, datetime] | None = None, ) -> None: self.catchup = catchup self._next_run = next_run self.calls: list[tuple[str, object]] = [] def next_run(self, now: datetime) -> tuple[str, datetime]: self.calls.append(("next_run", now)) if self._next_run is not None: return self._next_run # default: the Monday 07:00 after `now`, covering the week before it d = now.astimezone(KST).date() while d.weekday() != 0: d += timedelta(days=1) run_at = datetime(d.year, d.month, d.day, 7, 0, tzinfo=KST) if run_at <= now: run_at += timedelta(days=7) from sebrief.week import week_id return week_id(run_at.date() - timedelta(days=7)), run_at def latest_completed_week(self, now: datetime) -> str: from sebrief.week import week_id return week_id(now.astimezone(KST).date() - timedelta(days=7)) def catchup_candidate(self, now: datetime, window_hours: int) -> str | None: self.calls.append(("catchup_candidate", (now, window_hours))) return self.catchup class FakeStore: """In-memory stand-in for sebrief.store.Store.""" def __init__(self, *, pending: list[Paper] | None = None) -> None: self.upserted: list[list[Paper]] = [] self.saved_summaries: list[dict] = [] self.published_dois: set[str] = set() self.pending = pending if pending is not None else [] self.digests: dict[str, object] = {} self.digested: list[tuple[list[str], str]] = [] self.sent: list[tuple[str, str, str]] = [] self.closed = False def upsert_papers(self, papers: list[Paper]) -> int: self.upserted.append(list(papers)) return len(papers) def save_paper_summaries(self, summaries: dict) -> None: self.saved_summaries.append(dict(summaries)) def pending_papers(self, week: str) -> list[Paper]: return list(self.pending) def mark_digested(self, dois: list[str], week: str) -> None: self.digested.append((list(dois), week)) def digested_dois(self, week: str) -> set[str]: return set(self.published_dois) def has_digest(self, week: str) -> bool: return week in self.digests def save_digest(self, result) -> None: self.digests[result.payload.week] = result def mark_sent(self, week: str, channel: str, receipt: str) -> None: self.sent.append((week, channel, receipt)) def close(self) -> None: self.closed = True class FakeDeliverer: def __init__(self, name: str = "fake", *, fail: bool = False) -> None: self.name = name self.fail = fail self.delivered: list = [] self.errors: list[tuple[str, str]] = [] def deliver(self, result) -> str: if self.fail: raise RuntimeError(f"{self.name} is down") self.delivered.append(result) return f"{self.name}-receipt-{len(self.delivered)}" def notify_error(self, subject: str, detail: str) -> None: self.errors.append((subject, detail)) class FakeProvider: """LLMProvider stand-in: returns canned digest / verification objects.""" def __init__( self, digest: Digest | None = None, verifications: list | None = None, revised: Digest | None = None, fail_summarize: bool = False, fail_revise: bool = False, fail_paper_summaries: bool = False, paper_verifications: list | None = None, revised_papers: list | None = None, ) -> None: self.digest = digest or make_digest() self.verifications = list(verifications or []) self.revised = revised self.fail_summarize = fail_summarize self.fail_revise = fail_revise self.fail_paper_summaries = fail_paper_summaries self.paper_verifications = list(paper_verifications or []) self.revised_papers = revised_papers self.calls: list[str] = [] def summarize(self, payload): self.calls.append("summarize") if self.fail_summarize: from sebrief.llm.base import LLMError raise LLMError("all models failed") return self.digest, "fake-model" def summarize_papers(self, papers): self.calls.append(f"summarize_papers:{len(papers)}") if self.fail_paper_summaries: from sebrief.llm.base import LLMError raise LLMError("paper batch failed") items = [ PaperSummaryItem(doi=p.doi, **make_summary(tldr=f"{p.doi} 한 줄 요약입니다.").model_dump()) for p in papers ] return items, "fake-model" def verify_paper_summaries(self, papers, items): self.calls.append("verify_papers") if not self.paper_verifications: from sebrief.models import Verification return Verification(ok=True, issues=[]), "fake-model" return self.paper_verifications.pop(0), "fake-model" def revise_paper_summaries(self, papers, items, issues): self.calls.append("revise_papers") if self.revised_papers is not None: return list(self.revised_papers), "fake-model" return list(items), "fake-model" def verify(self, payload, digest): self.calls.append("verify") if not self.verifications: from sebrief.models import Verification return Verification(ok=True, issues=[]), "fake-model" return self.verifications.pop(0), "fake-model" def revise(self, payload, digest, issues): self.calls.append("revise") if self.fail_revise: from sebrief.llm.base import LLMError raise LLMError("revise failed") return self.revised or self.digest, "fake-model"