139 lines
5.1 KiB
Python
139 lines
5.1 KiB
Python
"""CLI: argument parsing, exit codes, config masking, path resolution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from _fakes import make_payload
|
|
|
|
from sebrief import cli
|
|
from sebrief.config import Settings
|
|
from sebrief.models import DigestResult
|
|
from sebrief.pipeline import DELIVERY_FAILED_PREFIX, SKIP_ALREADY_SENT
|
|
|
|
|
|
def make_settings(tmp_path) -> Settings:
|
|
return Settings(
|
|
_env_file=None,
|
|
SEBRIEF_DATA_DIR=str(tmp_path / "data"),
|
|
SEBRIEF_REPORTS_DIR=str(tmp_path / "reports"),
|
|
SEBRIEF_LOG_DIR=str(tmp_path / "logs"),
|
|
)
|
|
|
|
|
|
def make_result(error: str | None = None, degraded: bool = False) -> DigestResult:
|
|
return DigestResult(payload=make_payload(), digest=None, markdown="# md", degraded=degraded, error=error)
|
|
|
|
|
|
@pytest.fixture
|
|
def run(monkeypatch, tmp_path):
|
|
"""Invoke the CLI with settings pinned to tmp_path and run_digest captured."""
|
|
calls: dict = {}
|
|
|
|
def fake_run_digest(week, settings, *, dry_run=False, force=False, no_llm=False):
|
|
calls.update(week=week, dry_run=dry_run, force=force, no_llm=no_llm)
|
|
return calls.get("result") or make_result()
|
|
|
|
monkeypatch.setattr(cli, "_load_settings", lambda: make_settings(tmp_path))
|
|
monkeypatch.setattr(cli, "run_digest", fake_run_digest)
|
|
|
|
def invoke(*argv: str, result: DigestResult | None = None) -> int:
|
|
if result is not None:
|
|
calls["result"] = result
|
|
return cli.main(list(argv))
|
|
|
|
invoke.calls = calls # type: ignore[attr-defined]
|
|
return invoke
|
|
|
|
|
|
# ---------------------------------------------------------------------------------- parsing
|
|
def test_parser_requires_command():
|
|
with pytest.raises(SystemExit):
|
|
cli.build_parser().parse_args([])
|
|
|
|
|
|
def test_parser_rejects_bad_week():
|
|
with pytest.raises(SystemExit):
|
|
cli.build_parser().parse_args(["run-once", "--week", "2026-09-01"])
|
|
|
|
|
|
def test_run_once_flags_forwarded(run):
|
|
assert run("run-once", "--week", "2026-W30", "--dry-run", "--force", "--no-llm") == cli.EXIT_OK
|
|
assert run.calls["week"] == "2026-W30"
|
|
assert run.calls["dry_run"] and run.calls["force"] and run.calls["no_llm"]
|
|
|
|
|
|
# ------------------------------------------------------------------------------- exit codes
|
|
def test_run_once_ok(run):
|
|
assert run("run-once") == cli.EXIT_OK
|
|
|
|
|
|
def test_run_once_skip_exits_2(run, capsys):
|
|
assert run("run-once", result=make_result(error=SKIP_ALREADY_SENT)) == cli.EXIT_SKIPPED
|
|
assert "skipped" in capsys.readouterr().err
|
|
|
|
|
|
def test_run_once_delivery_failure_exits_1(run, capsys):
|
|
result = make_result(error=f"{DELIVERY_FAILED_PREFIX}: file: boom")
|
|
assert run("run-once", result=result) == cli.EXIT_ERROR
|
|
assert "delivery failed" in capsys.readouterr().err
|
|
|
|
|
|
def test_run_once_degraded_still_ok(run):
|
|
assert run("run-once", result=make_result(degraded=True, error="llm down")) == cli.EXIT_OK
|
|
|
|
|
|
def test_unexpected_exception_exits_1(run, monkeypatch, capsys):
|
|
def boom(*a, **k):
|
|
raise RuntimeError("kaboom")
|
|
|
|
monkeypatch.setattr(cli, "run_digest", boom)
|
|
assert run("run-once") == cli.EXIT_ERROR
|
|
assert "kaboom" in capsys.readouterr().err
|
|
|
|
|
|
# --------------------------------------------------------------------------------- commands
|
|
def test_next_run_prints_week_and_time(run, capsys):
|
|
assert run("next-run") == cli.EXIT_OK
|
|
out = capsys.readouterr().out
|
|
assert "week" in out and "digest at KST" in out
|
|
|
|
|
|
def test_show_config_masks_secrets(monkeypatch, tmp_path, capsys):
|
|
settings = make_settings(tmp_path).model_copy(update={"minimax_api_key": "sk-super-secret-key"})
|
|
monkeypatch.setattr(cli, "_load_settings", lambda: settings)
|
|
assert cli.main(["show-config"]) == cli.EXIT_OK
|
|
out = capsys.readouterr().out
|
|
assert "sk-super-secret-key" not in out
|
|
assert "minimax_api_key = sk-s***" in out
|
|
assert "db_path" in out
|
|
|
|
|
|
def test_test_notify_writes_errors_log(monkeypatch, tmp_path, capsys):
|
|
monkeypatch.setattr(cli, "_load_settings", lambda: make_settings(tmp_path))
|
|
assert cli.main(["test-notify", "--message", "hello"]) == cli.EXIT_OK
|
|
assert "sent test notification via file" in capsys.readouterr().out
|
|
log_text = (tmp_path / "reports" / "errors.log").read_text(encoding="utf-8")
|
|
assert "hello" in log_text
|
|
|
|
|
|
# ------------------------------------------------------------------------------------ paths
|
|
def test_resolve_paths_makes_relative_paths_absolute(tmp_path):
|
|
settings = Settings(_env_file=None) # default relative data/reports/logs
|
|
resolved = cli._resolve_paths(settings, root=tmp_path)
|
|
assert resolved.data_dir == tmp_path / "data"
|
|
assert resolved.reports_dir == tmp_path / "reports"
|
|
assert resolved.log_dir == tmp_path / "logs"
|
|
|
|
|
|
def test_resolve_paths_keeps_absolute(tmp_path):
|
|
settings = Settings(_env_file=None, SEBRIEF_DATA_DIR=str(tmp_path / "abs"))
|
|
resolved = cli._resolve_paths(settings, root=tmp_path / "elsewhere")
|
|
assert resolved.data_dir == tmp_path / "abs"
|
|
|
|
|
|
def test_mask_secret():
|
|
assert cli.mask_secret(None) == "(unset)"
|
|
assert cli.mask_secret("") == "(unset)"
|
|
assert cli.mask_secret("short") == "***"
|
|
assert cli.mask_secret("sk-1234567890") == "sk-1***"
|