243 lines
9.2 KiB
Python
243 lines
9.2 KiB
Python
"""CLI tests: settings/clock/pipeline monkeypatched on us2kr.cli."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
from _orchestration_stubs import FakeClock, FakeDeliverer
|
|
|
|
from us2kr import cli
|
|
from us2kr.config import Settings
|
|
from us2kr.models import BriefingResult, MarketPayload
|
|
|
|
KST = ZoneInfo("Asia/Seoul")
|
|
ET = ZoneInfo("America/New_York")
|
|
SECRET = "sk-cp-verysecretkeyvalue1234567890"
|
|
BOT_TOKEN = "123456789:AAH-abcdefghijklmnopqrstuvwxyz0123"
|
|
|
|
|
|
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": SECRET,
|
|
"TELEGRAM_BOT_TOKEN": BOT_TOKEN,
|
|
"TELEGRAM_CHAT_ID": "-1001",
|
|
}
|
|
values.update(overrides)
|
|
return Settings(_env_file=None, **values)
|
|
|
|
|
|
def make_result(session=date(2026, 8, 25), error=None, degraded=False) -> BriefingResult:
|
|
payload = MarketPayload(
|
|
session_date=session,
|
|
us_close_time=datetime(2026, 8, 25, 16, 0, tzinfo=ET),
|
|
generated_at=datetime(2026, 8, 26, 6, 15, tzinfo=KST),
|
|
krx_date=date(2026, 8, 26),
|
|
krx_open_today=True,
|
|
)
|
|
return BriefingResult(payload=payload, report=None, markdown="# md", degraded=degraded, error=error)
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(monkeypatch, tmp_path):
|
|
s = make_settings(tmp_path)
|
|
monkeypatch.setattr(cli, "_load_settings", lambda: s)
|
|
monkeypatch.setattr(cli, "setup_logging", lambda settings: None)
|
|
return s
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_clock(monkeypatch):
|
|
clock = FakeClock()
|
|
monkeypatch.setattr(cli, "MarketClock", lambda offset, **_kw: clock)
|
|
return clock
|
|
|
|
|
|
# ---------------------------------------------------------------------------------- next-run
|
|
def test_next_run_prints_session_kst_time_et_close_and_krx_status(settings, fake_clock, capsys):
|
|
fake_clock.next_run = (date(2026, 8, 25), datetime(2026, 8, 26, 6, 15, tzinfo=KST))
|
|
assert cli.main(["next-run"]) == 0
|
|
out = capsys.readouterr().out
|
|
assert "2026-08-25" in out
|
|
assert "2026-08-26T06:15+09:00" in out
|
|
assert "2026-08-25T16:00-04:00" in out and "America/New_York" in out
|
|
assert "2026-08-26 (open)" in out
|
|
|
|
|
|
def test_next_run_shows_krx_closed(settings, fake_clock, capsys):
|
|
fake_clock.krx_open = False
|
|
assert cli.main(["next-run"]) == 0
|
|
assert "CLOSED" in capsys.readouterr().out
|
|
|
|
|
|
# ------------------------------------------------------------------------------- show-config
|
|
def test_show_config_masks_secrets(settings, capsys):
|
|
assert cli.main(["show-config"]) == 0
|
|
out = capsys.readouterr().out
|
|
assert SECRET not in out and BOT_TOKEN not in out
|
|
assert "minimax_api_key = sk-c***" in out
|
|
assert "telegram_bot_token = 1234***" in out
|
|
assert "telegram_chat_id = -1001" in out # not a secret
|
|
assert "minimax_max_completion_tokens = 16384" in out # 'tokens' is not a secret field
|
|
assert "fred_api_key = (unset)" in out
|
|
assert "run_offset_minutes = 75" in out
|
|
assert "telegram_enabled = True" in out
|
|
assert "db_path = " in out and "us2kr.sqlite3" in out
|
|
|
|
|
|
def test_show_config_unset_secret(monkeypatch, tmp_path, capsys):
|
|
s = make_settings(tmp_path, MINIMAX_API_KEY="", TELEGRAM_BOT_TOKEN=None)
|
|
monkeypatch.setattr(cli, "_load_settings", lambda: s)
|
|
monkeypatch.setattr(cli, "setup_logging", lambda settings: None)
|
|
cli.main(["show-config"])
|
|
out = capsys.readouterr().out
|
|
assert "minimax_api_key = (unset)" in out
|
|
|
|
|
|
# ---------------------------------------------------------------------------------- run-once
|
|
class RunBriefingRecorder:
|
|
def __init__(self, result=None, exc=None):
|
|
self.result = result or make_result()
|
|
self.exc = exc
|
|
self.calls: list[tuple] = []
|
|
|
|
def __call__(self, session, settings, *, dry_run=False, force=False, no_llm=False):
|
|
self.calls.append((session, settings, dry_run, force, no_llm))
|
|
if self.exc:
|
|
raise self.exc
|
|
return self.result
|
|
|
|
|
|
def test_run_once_dry_run_calls_pipeline_with_dry_run(settings, monkeypatch):
|
|
rec = RunBriefingRecorder()
|
|
monkeypatch.setattr(cli, "run_briefing", rec)
|
|
assert cli.main(["run-once", "--dry-run"]) == 0
|
|
assert rec.calls == [(None, settings, True, False, False)]
|
|
|
|
|
|
def test_run_once_passes_date_force_no_llm(settings, monkeypatch):
|
|
rec = RunBriefingRecorder()
|
|
monkeypatch.setattr(cli, "run_briefing", rec)
|
|
assert cli.main(["run-once", "--date", "2026-08-25", "--force", "--no-llm"]) == 0
|
|
assert rec.calls == [(date(2026, 8, 25), settings, False, True, True)]
|
|
|
|
|
|
def test_run_once_invalid_date_is_usage_error(settings):
|
|
with pytest.raises(SystemExit) as exc:
|
|
cli.main(["run-once", "--date", "2026/08/25"])
|
|
assert exc.value.code == 2
|
|
|
|
|
|
@pytest.mark.parametrize("reason", ["already sent", "krx closed"])
|
|
def test_run_once_skip_reasons_exit_2(settings, monkeypatch, capsys, reason):
|
|
monkeypatch.setattr(cli, "run_briefing", RunBriefingRecorder(result=make_result(error=reason)))
|
|
assert cli.main(["run-once"]) == 2
|
|
assert reason in capsys.readouterr().err
|
|
|
|
|
|
def test_run_once_degraded_still_exit_0(settings, monkeypatch):
|
|
rec = RunBriefingRecorder(result=make_result(error="LLM failed", degraded=True))
|
|
monkeypatch.setattr(cli, "run_briefing", rec)
|
|
assert cli.main(["run-once"]) == 0
|
|
|
|
|
|
def test_run_once_delivery_failure_exit_1(settings, monkeypatch, capsys):
|
|
err = "delivery failed: telegram: DeliveryError: 400 chat not found"
|
|
monkeypatch.setattr(cli, "run_briefing", RunBriefingRecorder(result=make_result(error=err)))
|
|
assert cli.main(["run-once"]) == 1
|
|
assert "telegram" in capsys.readouterr().err
|
|
|
|
|
|
def test_run_once_exception_exit_1(settings, monkeypatch, capsys):
|
|
monkeypatch.setattr(cli, "run_briefing", RunBriefingRecorder(exc=ValueError("not a US trading session")))
|
|
assert cli.main(["run-once", "--date", "2026-08-23"]) == 1
|
|
assert "not a US trading session" in capsys.readouterr().err
|
|
|
|
|
|
# ------------------------------------------------------------------------------------ daemon
|
|
def test_daemon_subcommand_calls_run_daemon(settings, monkeypatch):
|
|
called = []
|
|
monkeypatch.setattr(cli, "run_daemon", lambda s: called.append(s))
|
|
assert cli.main(["daemon"]) == 0
|
|
assert called == [settings]
|
|
|
|
|
|
# ------------------------------------------------------------------------------- test-notify
|
|
def test_test_notify_sends_through_every_deliverer(settings, monkeypatch, capsys):
|
|
a, b = FakeDeliverer("file"), FakeDeliverer("telegram")
|
|
monkeypatch.setattr(cli, "build_deliverers", lambda s: [a, b])
|
|
assert cli.main(["test-notify", "--message", "hello from test"]) == 0
|
|
assert a.errors == [("US2KR 테스트 알림", "hello from test")]
|
|
assert b.errors == a.errors
|
|
out = capsys.readouterr().out
|
|
assert "file" in out and "telegram" in out
|
|
|
|
|
|
def test_test_notify_without_channels_is_error(settings, monkeypatch):
|
|
monkeypatch.setattr(cli, "build_deliverers", lambda s: [])
|
|
assert cli.main(["test-notify"]) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------- settings/paths
|
|
def test_resolve_paths_makes_relative_paths_absolute(tmp_path):
|
|
s = Settings(
|
|
_env_file=None, US2KR_DATA_DIR="data", US2KR_REPORTS_DIR="/abs/reports", US2KR_LOG_DIR="logs"
|
|
)
|
|
resolved = cli._resolve_paths(s, root=tmp_path)
|
|
assert resolved.data_dir == tmp_path / "data"
|
|
assert resolved.log_dir == tmp_path / "logs"
|
|
assert resolved.reports_dir == Path("/abs/reports")
|
|
assert resolved.db_path == tmp_path / "data" / "us2kr.sqlite3"
|
|
assert s.data_dir == Path("data") # original untouched
|
|
|
|
|
|
def test_load_settings_falls_back_to_project_root_env(monkeypatch, tmp_path):
|
|
root = tmp_path / "proj"
|
|
root.mkdir()
|
|
(root / ".env").write_text("US2KR_RUN_OFFSET_MINUTES=90\nUS2KR_DATA_DIR=var/data\n", encoding="utf-8")
|
|
cwd = tmp_path / "elsewhere"
|
|
cwd.mkdir()
|
|
monkeypatch.chdir(cwd) # like launchd starting with cwd=/
|
|
monkeypatch.delenv("US2KR_RUN_OFFSET_MINUTES", raising=False)
|
|
monkeypatch.delenv("US2KR_DATA_DIR", raising=False)
|
|
monkeypatch.setattr(cli, "PROJECT_ROOT", root)
|
|
|
|
s = cli._load_settings()
|
|
|
|
assert s.run_offset_minutes == 90
|
|
assert s.data_dir == root / "var" / "data"
|
|
assert s.log_dir == root / "logs"
|
|
|
|
|
|
def test_load_settings_prefers_cwd_env(monkeypatch, tmp_path):
|
|
root = tmp_path / "proj"
|
|
root.mkdir()
|
|
(root / ".env").write_text("US2KR_RUN_OFFSET_MINUTES=90\n", encoding="utf-8")
|
|
cwd = tmp_path / "cwd"
|
|
cwd.mkdir()
|
|
(cwd / ".env").write_text("US2KR_RUN_OFFSET_MINUTES=45\n", encoding="utf-8")
|
|
monkeypatch.chdir(cwd)
|
|
monkeypatch.delenv("US2KR_RUN_OFFSET_MINUTES", raising=False)
|
|
monkeypatch.setattr(cli, "PROJECT_ROOT", root)
|
|
assert cli._load_settings().run_offset_minutes == 45
|
|
|
|
|
|
def test_main_reports_bad_settings(monkeypatch, capsys):
|
|
def boom():
|
|
raise ValueError("bad env")
|
|
|
|
monkeypatch.setattr(cli, "_load_settings", boom)
|
|
assert cli.main(["show-config"]) == 1
|
|
assert "bad env" in capsys.readouterr().err
|
|
|
|
|
|
def test_mask_secret():
|
|
assert cli.mask_secret("") == "(unset)"
|
|
assert cli.mask_secret("short") == "***"
|
|
assert cli.mask_secret("sk-cp-abcdefghijklmnop") == "sk-c***"
|