"""Weekly recap tests: collaborators are monkeypatched on us2kr.weekly; no network, no real DB.""" from __future__ import annotations import json from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo import pytest from _orchestration_stubs import FakeClock, FakeStore from us2kr import weekly from us2kr.config import Settings from us2kr.llm.base import LLMError from us2kr.llm.prompts import WEEKLY_SYSTEM_PROMPT, build_weekly_user_message from us2kr.models import Report, Verification from us2kr.pipeline import SKIP_ALREADY_SENT from us2kr.render import render_weekly_markdown from us2kr.weekly import build_weekly_payload, most_recent_monday, run_weekly, week_sessions_for KST = ZoneInfo("Asia/Seoul") MONDAY = date(2026, 9, 14) WEEK = [date(2026, 9, 7) + timedelta(days=i) for i in range(5)] # Mon 9/7 .. Fri 9/11 NOW = datetime(2026, 9, 14, 7, 5, tzinfo=KST) def make_settings(tmp_path, **overrides) -> Settings: values = { "US2KR_DATA_DIR": str(tmp_path / "data"), "US2KR_REPORTS_DIR": str(tmp_path / "reports"), "US2KR_LOG_DIR": str(tmp_path / "logs"), "MINIMAX_API_KEY": "sk-test-not-a-real-key", } values.update(overrides) return Settings(_env_file=None, **values) def quote_row(session: date, symbol: str, close: float, prev_close: float, **extra) -> dict: row = { "session_date": session.isoformat(), "symbol": symbol, "close": close, "prev_close": prev_close, "change_pct": None, "change_abs": None, "day_high": extra.get("day_high"), "day_low": extra.get("day_low"), "volume": None, } return row def report_row(krx_date: date, session: date, headline: str | None) -> dict: result = {"payload": {"session_date": session.isoformat()}} result["report"] = {"headline": headline} if headline is not None else None return { "krx_date": krx_date.isoformat(), "session_date": session.isoformat(), "result_json": json.dumps(result, ensure_ascii=False), } def sample_report() -> Report: return Report( headline="반도체 랠리로 주간 상승", market_overview="한 주간 지수가 올랐다.", key_drivers=["엔비디아 실적"], rates_fx_commodities="금리는 주간 상승.", korea_implications=["국내 반도체 우호적"], watchlist_today=["삼성전자"], risk_flags=["금리 재상승"], ) def populated_store() -> FakeStore: store = FakeStore() # ^GSPC: Mon prev_close 100 -> Fri close 110 = +10% weekly for i, session in enumerate(WEEK): store.quote_rows.append( quote_row(session, "^GSPC", 102.0 + 2 * i, 100.0 + 2 * i, day_high=103.0 + 2 * i, day_low=99.0) ) # daily headlines Tue..Sat (krx dates for Mon..Fri sessions) for session in WEEK: store.report_rows.append(report_row(session + timedelta(days=1), session, f"{session.day}일 헤드라인")) return store # ------------------------------------------------------------------- date helpers def test_most_recent_monday(): assert most_recent_monday(MONDAY) == MONDAY assert most_recent_monday(date(2026, 9, 18)) == MONDAY # Friday -> that week's Monday assert most_recent_monday(date(2026, 9, 13)) == date(2026, 9, 7) # Sunday -> previous Monday def test_week_sessions_excludes_holidays(): clock = FakeClock() assert week_sessions_for(MONDAY, clock) == WEEK class HolidayClock(FakeClock): def is_us_session(self, d: date) -> bool: return super().is_us_session(d) and d != date(2026, 9, 7) # Labor-day-like Monday assert week_sessions_for(MONDAY, HolidayClock()) == WEEK[1:] # ------------------------------------------------------------------- payload builder def test_build_weekly_payload_aggregates_week(tmp_path): store = populated_store() payload, sessions = build_weekly_payload(MONDAY, make_settings(tmp_path), FakeClock(), store) assert sessions == WEEK assert payload.krx_date == MONDAY and payload.session_date == WEEK[-1] assert payload.krx_open_today is True (q,) = payload.quotes assert q.symbol == "^GSPC" and q.name == "S&P 500" and q.group == "indices" assert q.prev_close == 100.0 and q.close == 110.0 assert q.change_pct == pytest.approx(10.0) assert q.change_abs == pytest.approx(10.0) assert q.day_high == 111.0 and q.day_low == 99.0 assert payload.history_notes == [f"미 {s.isoformat()}({w}): {s.day}일 헤드라인" for s, w in zip(WEEK, "월화수목금")] assert payload.data_gaps == [] def test_build_weekly_payload_flags_missing_data(tmp_path): store = FakeStore() store.quote_rows.append(quote_row(WEEK[0], "^GSPC", 102.0, 100.0)) # only Monday stored payload, _ = build_weekly_payload(MONDAY, make_settings(tmp_path), FakeClock(), store) assert any(g.startswith("weekly quotes: missing sessions") for g in payload.data_gaps) assert "daily headlines: no stored briefings for last week" in payload.data_gaps (q,) = payload.quotes assert q.close == 102.0 and q.prev_close == 100.0 # single-session week still aggregates def test_build_weekly_payload_degraded_daily_report(tmp_path): store = populated_store() store.report_rows[0] = report_row(WEEK[0] + timedelta(days=1), WEEK[0], None) # degraded day payload, _ = build_weekly_payload(MONDAY, make_settings(tmp_path), FakeClock(), store) assert payload.history_notes[0].endswith("(요약 없음)") # ------------------------------------------------------------------- orchestration class Env: def __init__(self, monkeypatch, tmp_path, **settings_overrides): self.settings = make_settings(tmp_path, **settings_overrides) self.clock = FakeClock() self.store = populated_store() self.calls: dict[str, int] = {} self.delivered: list = [] def count(name): self.calls[name] = self.calls.get(name, 0) + 1 def summarize_weekly_payload(payload, sessions, provider): count("llm") assert sessions == WEEK return sample_report(), "MiniMax-M3" def verify_and_revise(payload, report, provider, *, data_message=None, system_prompt=None): count("verify") assert system_prompt == WEEKLY_SYSTEM_PROMPT assert data_message and "## 주간 시세" in data_message return report, Verification(ok=True) def deliver(result, settings, store, report_id): count("deliver") self.delivered.append(result) return [] monkeypatch.setattr(weekly, "MarketClock", lambda offset, **_kw: self.clock) monkeypatch.setattr(weekly, "Store", lambda path: self.store) monkeypatch.setattr(weekly, "build_provider", lambda settings: object()) monkeypatch.setattr(weekly, "summarize_weekly_payload", summarize_weekly_payload) monkeypatch.setattr(weekly, "verify_and_revise", verify_and_revise) monkeypatch.setattr(weekly, "_deliver", deliver) monkeypatch.setattr(weekly, "_now", lambda: NOW) self.monkeypatch = monkeypatch @pytest.fixture def env(monkeypatch, tmp_path): return Env(monkeypatch, tmp_path) def test_run_weekly_happy_path(env): result = run_weekly(MONDAY, env.settings) assert result.kind == "weekly" and result.degraded is False and result.error is None assert result.payload.krx_date == MONDAY assert result.verification is not None and result.verification.ok is True assert env.calls == {"llm": 1, "verify": 1, "deliver": 1} assert env.store.reports == {(MONDAY, "weekly"): 1} assert "미국 주간 리뷰" in result.markdown and "반도체 랠리로 주간 상승" in result.markdown assert env.store.closed is True def test_run_weekly_defaults_to_latest_monday(env): result = run_weekly(None, env.settings) assert result.payload.krx_date == MONDAY # NOW is that Monday morning def test_run_weekly_skips_when_already_sent(env): env.store.reports[(MONDAY, "weekly")] = 99 result = run_weekly(MONDAY, env.settings) assert result.error == SKIP_ALREADY_SENT assert "deliver" not in env.calls def test_run_weekly_force_resends(env): env.store.reports[(MONDAY, "weekly")] = 99 result = run_weekly(MONDAY, env.settings, force=True) assert result.error is None and env.calls["deliver"] == 1 def test_run_weekly_rejects_non_monday(env): with pytest.raises(ValueError, match="not a Monday"): run_weekly(date(2026, 9, 15), env.settings) def test_run_weekly_rejects_future_monday(env): with pytest.raises(ValueError, match="future"): run_weekly(date(2026, 9, 21), env.settings) def test_run_weekly_dry_run_prints_and_saves_nothing(env, capsys): result = run_weekly(MONDAY, env.settings, dry_run=True) assert "미국 주간 리뷰" in capsys.readouterr().out assert env.store.reports == {} and "deliver" not in env.calls assert result.kind == "weekly" def test_run_weekly_no_llm_degrades(env): result = run_weekly(MONDAY, env.settings, no_llm=True) assert result.degraded is True and result.report is None assert "llm" not in env.calls and "verify" not in env.calls assert env.calls["deliver"] == 1 assert "축소 모드" in result.markdown def test_run_weekly_llm_failure_degrades(env): def boom(payload, sessions, provider): raise LLMError("all models down") env.monkeypatch.setattr(weekly, "summarize_weekly_payload", boom) result = run_weekly(MONDAY, env.settings) assert result.degraded is True and result.error == "all models down" assert env.calls["deliver"] == 1 def test_run_weekly_verify_failure_is_non_fatal(env): def boom(payload, report, provider, **kwargs): raise RuntimeError("verifier down") env.monkeypatch.setattr(weekly, "verify_and_revise", boom) result = run_weekly(MONDAY, env.settings) assert result.report is not None and result.verification is None and result.error is None def test_run_weekly_verification_disabled(monkeypatch, tmp_path): env = Env(monkeypatch, tmp_path, US2KR_VERIFY_ENABLED="false") result = run_weekly(MONDAY, env.settings) assert "verify" not in env.calls and result.verification is None # ------------------------------------------------------------------- prompt & render def test_build_weekly_user_message_contents(env, tmp_path): payload, sessions = build_weekly_payload(MONDAY, env.settings, FakeClock(), populated_store()) message = build_weekly_user_message(payload, sessions) assert "지난주 미국 거래일: 2026-09-07 ~ 2026-09-11 (5거래일)" in message assert "## 주간 시세 (변동 = 주간 누적)" in message assert "^GSPC S&P 500: 110.00 (+10.00%)" in message assert "## 일별 요약" in message and "미 2026-09-08(화): 8일 헤드라인" in message def test_weekly_system_prompt_is_static_and_weekly_framed(): import re assert "주간" in WEEKLY_SYSTEM_PROMPT and "미국 거래일" in WEEKLY_SYSTEM_PROMPT assert not re.search(r"\d{4}-\d{2}-\d{2}", WEEKLY_SYSTEM_PROMPT) assert not re.search(r"20\d{2}", WEEKLY_SYSTEM_PROMPT) def test_render_weekly_markdown(env): payload, sessions = build_weekly_payload(MONDAY, env.settings, FakeClock(), populated_store()) md = render_weekly_markdown(payload, sample_report(), sessions) assert md.startswith("# 🇺🇸→🇰🇷 미국 주간 리뷰 — 2026-09-14 (미 2026-09-07 ~ 2026-09-11)") assert "## 주간 총평\n**반도체 랠리로 주간 상승**" in md assert "## 이번 주 한국 장 시사점" in md and "## 이번 주 볼 것" in md assert "📊 주요 지수 — 주간 변동" in md and "| S&P 500 | 110.00 | +10.00% |" in md assert "## 일별 헤드라인" in md def test_render_weekly_markdown_degraded(env): payload, sessions = build_weekly_payload(MONDAY, env.settings, FakeClock(), populated_store()) md = render_weekly_markdown(payload, None, sessions) assert "🧩 축소 모드(LLM 요약 없음)" in md and "## 주간 총평" not in md # ------------------------------------------------------------------- daemon wiring from us2kr.daemon import WEEKLY_JOB_ID, Daemon class MiniScheduler: def __init__(self): self.jobs: dict = {} def add_job(self, func, trigger=None, **kwargs): self.jobs[kwargs.get("id")] = {"func": func, "trigger": trigger, **kwargs} def add_listener(self, callback, mask=0): pass def shutdown(self, wait=True): pass def make_daemon(tmp_path, *, now=NOW, weekly_fn=..., store=None, **settings_overrides): calls: list[date] = [] if weekly_fn is ...: weekly_fn = calls.append daemon = Daemon( make_settings(tmp_path, **settings_overrides), scheduler=MiniScheduler(), clock=FakeClock(), store=store or FakeStore(), run_fn=lambda session: None, now_fn=lambda: now, deliverers_fn=lambda s: [], weekly_fn=weekly_fn, ) return daemon, calls def test_daemon_schedules_weekly_cron(tmp_path): daemon, _ = make_daemon(tmp_path) daemon.schedule_weekly() job = daemon.scheduler.jobs[WEEKLY_JOB_ID] fields = {f.name: str(f) for f in job["trigger"].fields} assert fields["day_of_week"] == "mon" and fields["hour"] == "7" and fields["minute"] == "0" def test_daemon_weekly_schedule_respects_disable_and_missing_fn(tmp_path): daemon, _ = make_daemon(tmp_path, US2KR_WEEKLY_ENABLED="false") daemon.schedule_weekly() assert WEEKLY_JOB_ID not in daemon.scheduler.jobs daemon, _ = make_daemon(tmp_path, weekly_fn=None) daemon.schedule_weekly() assert WEEKLY_JOB_ID not in daemon.scheduler.jobs def test_daemon_weekly_catchup_runs_missed_recap(tmp_path): daemon, calls = make_daemon(tmp_path, now=datetime(2026, 9, 14, 8, 40, tzinfo=KST)) assert daemon.run_weekly_catchup() is True assert calls == [MONDAY] def test_daemon_weekly_catchup_skips_when_sent_or_out_of_window(tmp_path): store = FakeStore() store.reports[(MONDAY, "weekly")] = 1 daemon, calls = make_daemon(tmp_path, now=datetime(2026, 9, 14, 8, 40, tzinfo=KST), store=store) assert daemon.run_weekly_catchup() is False and calls == [] daemon, calls = make_daemon(tmp_path, now=datetime(2026, 9, 15, 8, 40, tzinfo=KST)) # Tuesday assert daemon.run_weekly_catchup() is False and calls == [] daemon, calls = make_daemon(tmp_path, now=datetime(2026, 9, 14, 6, 30, tzinfo=KST)) # before 07:00 assert daemon.run_weekly_catchup() is False and calls == [] daemon, calls = make_daemon(tmp_path, now=datetime(2026, 9, 14, 10, 30, tzinfo=KST)) # window passed assert daemon.run_weekly_catchup() is False and calls == [] def test_daemon_weekly_job_failure_notifies_but_survives(tmp_path): daemon, _ = make_daemon(tmp_path, now=datetime(2026, 9, 14, 7, 0, tzinfo=KST)) errors: list[tuple[str, str]] = [] daemon._notify_error = lambda subject, detail: errors.append((subject, detail)) daemon.weekly_fn = lambda monday: (_ for _ in ()).throw(RuntimeError("boom")) daemon._weekly_job() # must not raise assert errors and "주간 리뷰" in errors[0][0]