118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tenbackward import main as main_module
|
|
from tenbackward.logging_setup import JsonFormatter
|
|
from tenbackward.main import main
|
|
|
|
|
|
@pytest.fixture()
|
|
def capture_logger() -> io.StringIO:
|
|
buf = io.StringIO()
|
|
handler = logging.StreamHandler(buf)
|
|
handler.setFormatter(JsonFormatter())
|
|
target = logging.getLogger("tenbackward")
|
|
target.handlers = [handler]
|
|
target.setLevel(logging.INFO)
|
|
target.propagate = False
|
|
return buf
|
|
|
|
|
|
def _run_lines(buf: io.StringIO) -> list[dict]:
|
|
return [json.loads(line) for line in buf.getvalue().splitlines() if line.strip()]
|
|
|
|
|
|
def _seed_state(data_dir: Path, ids: list[str]) -> None:
|
|
from tenbackward.state import save_posted
|
|
|
|
state = {pid: {"posted_at": "2024-01-01T00:00:00+00:00"} for pid in ids}
|
|
save_posted(data_dir, state)
|
|
|
|
|
|
def test_run_emits_one_info_summary_on_success(env_setup, data_dir, capture_logger, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATA_DIR", str(data_dir))
|
|
|
|
monkeypatch.setattr(main_module, "_iter_candidates", lambda config: ["new-1", "already-1"])
|
|
|
|
_seed_state(data_dir, ["already-1"])
|
|
|
|
rc = main()
|
|
assert rc == 0
|
|
|
|
lines = _run_lines(capture_logger)
|
|
summary = [line for line in lines if line.get("event") == "run_complete"]
|
|
assert len(summary) == 1
|
|
payload = summary[0]
|
|
assert payload["scanned"] == 2
|
|
assert payload["matched"] == 2
|
|
assert payload["posted"] == 1
|
|
assert payload["skipped"] == 1
|
|
assert payload["posted_ids"] == ["new-1"]
|
|
|
|
from tenbackward.state import load_posted
|
|
|
|
state = load_posted(data_dir)
|
|
assert "new-1" in state
|
|
assert "already-1" in state
|
|
|
|
|
|
def test_run_silent_when_no_matches(env_setup, data_dir, capture_logger, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATA_DIR", str(data_dir))
|
|
monkeypatch.setattr(main_module, "_iter_candidates", lambda config: [])
|
|
|
|
rc = main()
|
|
assert rc == 0
|
|
|
|
lines = _run_lines(capture_logger)
|
|
summary = [line for line in lines if line.get("event") == "run_complete"]
|
|
assert summary == []
|
|
|
|
|
|
def test_run_returns_nonzero_on_pipeline_error(env_setup, data_dir, capture_logger, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATA_DIR", str(data_dir))
|
|
monkeypatch.setenv("MAX_RETRIES", "1")
|
|
|
|
def _boom(config):
|
|
raise RuntimeError("boom-token-should-not-appear")
|
|
|
|
monkeypatch.setattr(main_module, "_iter_candidates", _boom)
|
|
|
|
rc = main()
|
|
assert rc == 1
|
|
|
|
lines = _run_lines(capture_logger)
|
|
summary = [line for line in lines if line.get("event") == "run_complete"]
|
|
assert summary == []
|
|
|
|
error_lines = [line for line in lines if line.get("level") == "ERROR"]
|
|
assert any("exc_type" in line for line in error_lines)
|
|
assert "boom-token-should-not-appear" not in capture_logger.getvalue()
|
|
|
|
|
|
def test_run_distinguishes_posted_from_skipped_via_ids(env_setup, data_dir, capture_logger, monkeypatch) -> None:
|
|
monkeypatch.setenv("DATA_DIR", str(data_dir))
|
|
monkeypatch.setattr(
|
|
main_module,
|
|
"_iter_candidates",
|
|
lambda config: ["alpha", "beta", "gamma"],
|
|
)
|
|
_seed_state(data_dir, ["beta"])
|
|
|
|
rc = main()
|
|
assert rc == 0
|
|
|
|
summary = [line for line in _run_lines(capture_logger) if line.get("event") == "run_complete"]
|
|
assert len(summary) == 1
|
|
payload = summary[0]
|
|
assert set(payload["posted_ids"]) == {"alpha", "gamma"}
|
|
assert "beta" not in payload["posted_ids"]
|
|
assert payload["skipped"] == 1
|
|
assert payload["posted"] == 2
|