thysrael--horizon
wehub-resource-sync
106 行
2.9 KiB
Python
106 行
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.mcp.run_store import RunStore
|
|
|
|
|
|
def test_create_run_writes_meta(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
|
|
run_id = store.create_run()
|
|
meta = store.load_meta(run_id)
|
|
|
|
assert run_id.startswith("run-")
|
|
assert meta["run_id"] == run_id
|
|
assert "created_at" in meta
|
|
|
|
|
|
def test_save_and_load_stage_items(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run_id = store.create_run("run-fixed")
|
|
items = [{"title": "foo"}, {"title": "bar"}]
|
|
|
|
path = store.save_items(run_id, "raw", items)
|
|
loaded = store.load_items(run_id, "raw")
|
|
|
|
assert path.name == "raw_items.json"
|
|
assert loaded == items
|
|
assert store.has_stage(run_id, "raw") is True
|
|
|
|
|
|
def test_update_meta_sets_updated_at(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run_id = store.create_run("run-meta")
|
|
|
|
meta = store.update_meta(run_id, {"status": "done"})
|
|
|
|
assert meta["status"] == "done"
|
|
assert "updated_at" in meta
|
|
|
|
|
|
def test_save_and_load_summary(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run_id = store.create_run("run-summary")
|
|
|
|
saved = store.save_summary(run_id, "zh", "# 摘要")
|
|
content = store.load_summary(run_id, "zh")
|
|
|
|
assert saved.name == "summary-zh.md"
|
|
assert content == "# 摘要"
|
|
|
|
|
|
def test_unsupported_stage_raises(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run_id = store.create_run("run-invalid-stage")
|
|
|
|
with pytest.raises(ValueError, match="Unsupported stage"):
|
|
store.save_items(run_id, "unknown", [])
|
|
|
|
|
|
def test_missing_run_raises(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
|
|
with pytest.raises(FileNotFoundError, match="Run not found"):
|
|
store.run_dir("missing-run")
|
|
|
|
|
|
def test_rejects_path_traversal_run_id(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path / "runs")
|
|
|
|
with pytest.raises(ValueError, match="Invalid run_id"):
|
|
store.create_run("../outside")
|
|
|
|
assert not (tmp_path / "outside").exists()
|
|
|
|
|
|
def test_rejects_unsafe_summary_language(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run_id = store.create_run("run-summary-safe")
|
|
|
|
with pytest.raises(ValueError, match="Invalid summary language"):
|
|
store.save_summary(run_id, "../zh", "# unsafe")
|
|
|
|
|
|
def test_missing_artifact_raises(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run_id = store.create_run("run-missing-file")
|
|
|
|
with pytest.raises(FileNotFoundError, match="Artifact not found"):
|
|
store.read_json(run_id, "does-not-exist.json")
|
|
|
|
|
|
def test_list_runs_returns_desc_order(tmp_path: Path) -> None:
|
|
store = RunStore(tmp_path)
|
|
run1 = store.create_run("run-1")
|
|
store.update_meta(run1, {"seq": 1})
|
|
run2 = store.create_run("run-2")
|
|
store.update_meta(run2, {"seq": 2})
|
|
|
|
runs = store.list_runs(limit=10)
|
|
|
|
assert runs[0]["run_id"] == "run-2"
|
|
assert runs[1]["run_id"] == "run-1"
|