mirror of
https://github.com/lllin000/PaperForge.git
synced 2026-07-22 06:50:53 +00:00
test(26-02): write 14 tests for context command modes and provenance output
- TestContextSingleKey: single key JSON object with _provenance and _ai_readiness - TestContextSingleKeyNotFound: missing key returns 1 with error - TestContextDomainFilter: --domain filters correctly (AIC-03) - TestContextCollectionFilter: --collection prefix match (AIC-03) - TestContextAll: --all returns all entries - TestContextProvenanceTraceability: all 9 provenance path keys present (AIC-04) - TestContextAiReadinessBlocking: blocking explanation for pdf_ready/fulltext_ready - TestContextNoEntriesMatch: empty results and no-mode error - TestContextIndexMissing: missing index guidance
This commit is contained in:
parent
26a5e001ef
commit
2e02729f83
1 changed files with 555 additions and 0 deletions
555
tests/test_context.py
Normal file
555
tests/test_context.py
Normal file
|
|
@ -0,0 +1,555 @@
|
|||
"""Tests for context.py — paperforge context CLI command.
|
||||
|
||||
Phase 26-02 Task 3: Tests for single-key, --domain, --collection, --all
|
||||
modes, provenance traceability, and AI readiness blocking explanations.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _minimal_vault(tmp_path: Path) -> Path:
|
||||
"""Create a minimal vault with paperforge.json for path resolution."""
|
||||
vault = tmp_path / "test_vault"
|
||||
vault.mkdir(parents=True, exist_ok=True)
|
||||
pf_json = vault / "paperforge.json"
|
||||
pf_json.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"version": "1.2.0",
|
||||
"vault_config": {
|
||||
"system_dir": "99_System",
|
||||
"resources_dir": "03_Resources",
|
||||
"literature_dir": "Literature",
|
||||
"control_dir": "LiteratureControl",
|
||||
"base_dir": "05_Bases",
|
||||
"skill_dir": ".opencode/skills",
|
||||
},
|
||||
},
|
||||
indent=2,
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
return vault
|
||||
|
||||
|
||||
def _write_index(vault: Path, items: list[dict]) -> None:
|
||||
"""Write a canonical index envelope to the vault."""
|
||||
from paperforge.worker.asset_index import atomic_write_index, build_envelope, get_index_path
|
||||
|
||||
idx_path = get_index_path(vault)
|
||||
idx_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
envelope = build_envelope(items)
|
||||
atomic_write_index(idx_path, envelope)
|
||||
|
||||
|
||||
def _make_entry(key: str, domain: str = "default", **overrides: dict) -> dict:
|
||||
"""Create a canonical index entry dict with default fields."""
|
||||
entry: dict = {
|
||||
"zotero_key": key,
|
||||
"domain": domain,
|
||||
"title": f"Paper {key}",
|
||||
"authors": [f"Author {key}"],
|
||||
"abstract": f"Abstract for {key}",
|
||||
"journal": "Test Journal",
|
||||
"year": "2024",
|
||||
"doi": f"10.1234/{key}",
|
||||
"pmid": "12345",
|
||||
"collections": [domain],
|
||||
"collection_path": domain,
|
||||
"collection_tags": [],
|
||||
"has_pdf": True,
|
||||
"pdf_path": f"[[storage/{key}/paper.pdf]]",
|
||||
"ocr_status": "done",
|
||||
"ocr_md_path": f"[[ocr/{key}/fulltext.md]]",
|
||||
"ocr_json_path": f"ocr/{key}/meta.json",
|
||||
"deep_reading_status": "done",
|
||||
"note_path": f"Literature/{domain}/{key} - Paper_{key}.md",
|
||||
"deep_reading_md_path": f"Literature/{domain}/{key} - Paper_{key}.md",
|
||||
"paper_root": f"Literature/{domain}/{key} - Paper_{key}/",
|
||||
"main_note_path": f"Literature/{domain}/{key} - Paper_{key}/{key} - Paper_{key}.md",
|
||||
"fulltext_path": f"Literature/{domain}/{key} - Paper_{key}/fulltext.md",
|
||||
"deep_reading_path": f"Literature/{domain}/{key} - Paper_{key}/deep-reading.md",
|
||||
"ai_path": f"Literature/{domain}/{key} - Paper_{key}/ai/",
|
||||
"lifecycle": "ai_context_ready",
|
||||
"health": {
|
||||
"pdf_health": "healthy",
|
||||
"ocr_health": "healthy",
|
||||
"note_health": "healthy",
|
||||
"asset_health": "healthy",
|
||||
},
|
||||
"maturity": {"level": 6, "blocking": []},
|
||||
"next_step": "ready",
|
||||
}
|
||||
entry.update(overrides)
|
||||
return entry
|
||||
|
||||
|
||||
def _make_args(
|
||||
key: str | None = None,
|
||||
domain: str | None = None,
|
||||
collection: str | None = None,
|
||||
all_mode: bool = False,
|
||||
vault_path: Path | None = None,
|
||||
) -> argparse.Namespace:
|
||||
"""Build an argparse.Namespace for context.run()."""
|
||||
return argparse.Namespace(
|
||||
key=key,
|
||||
domain=domain,
|
||||
collection=collection,
|
||||
all=all_mode,
|
||||
vault_path=vault_path,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestContextSingleKey:
|
||||
"""AIC-02: single paper context output."""
|
||||
|
||||
def test_context_single_key_output(self, tmp_path: Path) -> None:
|
||||
"""Single key returns JSON object with matching zotero_key."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [
|
||||
_make_entry("KEY001"),
|
||||
_make_entry("KEY002"),
|
||||
]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
output = " ".join(captured_out)
|
||||
data = json.loads(output)
|
||||
assert data["zotero_key"] == "KEY001"
|
||||
# Must be a dict (single JSON object), not a list
|
||||
assert isinstance(data, dict)
|
||||
|
||||
def test_context_single_key_includes_provenance(self, tmp_path: Path) -> None:
|
||||
"""Single key output includes _provenance block."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [_make_entry("KEY001")]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
prov = data.get("_provenance", {})
|
||||
assert isinstance(prov, dict)
|
||||
assert "paper_root" in prov
|
||||
assert "main_note_path" in prov
|
||||
assert "fulltext_path" in prov
|
||||
assert "pdf_path" in prov
|
||||
|
||||
def test_context_single_key_includes_ai_readiness(self, tmp_path: Path) -> None:
|
||||
"""Single key output includes _ai_readiness block."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [_make_entry("KEY001")]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
readiness = data.get("_ai_readiness", {})
|
||||
assert isinstance(readiness, dict)
|
||||
assert "ai_context_ready" in readiness
|
||||
assert "lifecycle" in readiness
|
||||
|
||||
|
||||
class TestContextSingleKeyNotFound:
|
||||
"""Missing key returns exit code 1."""
|
||||
|
||||
def test_context_key_not_found(self, tmp_path: Path, capsys) -> None:
|
||||
"""When key is not in index, return 1 and print error."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [_make_entry("KEY001")]
|
||||
_write_index(vault, items)
|
||||
|
||||
code = run(_make_args(key="KEY999", vault_path=vault))
|
||||
assert code == 1
|
||||
captured = capsys.readouterr()
|
||||
assert "not found" in captured.err.lower()
|
||||
|
||||
|
||||
class TestContextDomainFilter:
|
||||
"""AIC-03: domain-level context output."""
|
||||
|
||||
def test_context_domain_filter(self, tmp_path: Path) -> None:
|
||||
"""--domain filters to entries in that domain."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [
|
||||
_make_entry("KEY001", domain="骨科"),
|
||||
_make_entry("KEY002", domain="运动医学"),
|
||||
_make_entry("KEY003", domain="骨科"),
|
||||
]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(domain="骨科", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 2
|
||||
for entry in data:
|
||||
assert entry["domain"] == "骨科"
|
||||
|
||||
|
||||
class TestContextCollectionFilter:
|
||||
"""AIC-03: collection-path prefix match."""
|
||||
|
||||
def test_context_collection_filter(self, tmp_path: Path) -> None:
|
||||
"""--collection filters by prefix match on collections list."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [
|
||||
_make_entry("KEY001", domain="骨科", collections=["骨科/脊柱"]),
|
||||
_make_entry("KEY002", domain="运动医学", collections=["运动医学"]),
|
||||
_make_entry("KEY003", domain="骨科", collections=["骨科/创伤"]),
|
||||
_make_entry("KEY004", domain="神经科", collections=["神经科"]),
|
||||
]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(collection="骨科", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 2
|
||||
keys = [e["zotero_key"] for e in data]
|
||||
assert "KEY001" in keys
|
||||
assert "KEY003" in keys
|
||||
assert "KEY002" not in keys
|
||||
|
||||
|
||||
class TestContextAll:
|
||||
"""--all returns all index entries."""
|
||||
|
||||
def test_context_all(self, tmp_path: Path) -> None:
|
||||
"""--all outputs all entries as JSON array."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [
|
||||
_make_entry("KEY001"),
|
||||
_make_entry("KEY002"),
|
||||
_make_entry("KEY003"),
|
||||
]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(all_mode=True, vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 3
|
||||
|
||||
|
||||
class TestContextProvenanceTraceability:
|
||||
"""AIC-04: provenance paths are present and non-empty."""
|
||||
|
||||
def test_provenance_all_paths_present(self, tmp_path: Path) -> None:
|
||||
"""Output _provenance block has all path keys with non-empty values."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [_make_entry("KEY001")]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
prov = data.get("_provenance", {})
|
||||
expected_keys = [
|
||||
"paper_root",
|
||||
"main_note_path",
|
||||
"fulltext_path",
|
||||
"ocr_md_path",
|
||||
"pdf_path",
|
||||
"deep_reading_path",
|
||||
"ai_path",
|
||||
"note_path",
|
||||
"deep_reading_md_path",
|
||||
]
|
||||
for key in expected_keys:
|
||||
assert key in prov, f"Missing provenance key: {key}"
|
||||
assert isinstance(prov[key], str), f"Provenance key {key} should be a string"
|
||||
assert prov[key] != "", f"Provenance key {key} should be non-empty"
|
||||
|
||||
|
||||
class TestContextAiReadinessBlocking:
|
||||
"""AIC-04: blocking explanation when AI context not ready."""
|
||||
|
||||
def test_ai_context_ready_has_no_blocking(self, tmp_path: Path) -> None:
|
||||
"""ai_context_ready entry has no blocking factors."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [_make_entry("KEY001", lifecycle="ai_context_ready")]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
readiness = data["_ai_readiness"]
|
||||
assert readiness["ai_context_ready"] is True
|
||||
assert readiness["blocking_factors"] == []
|
||||
|
||||
def test_pdf_ready_has_blocking_explanation(self, tmp_path: Path) -> None:
|
||||
"""pdf_ready entry has blocking explanation about missing OCR."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [
|
||||
_make_entry(
|
||||
"KEY001",
|
||||
lifecycle="pdf_ready",
|
||||
has_pdf=True,
|
||||
ocr_status="pending",
|
||||
deep_reading_status="pending",
|
||||
health={"pdf_health": "healthy"},
|
||||
maturity={"level": 2, "blocking": ["ocr not done"]},
|
||||
next_step="ocr",
|
||||
)
|
||||
]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
readiness = data["_ai_readiness"]
|
||||
assert readiness["ai_context_ready"] is False
|
||||
assert "blocking_explanation" in readiness
|
||||
assert "ocr" in readiness["blocking_explanation"].lower()
|
||||
|
||||
def test_fulltext_ready_has_blocking_explanation(self, tmp_path: Path) -> None:
|
||||
"""fulltext_ready entry has blocking explanation about missing deep reading."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [
|
||||
_make_entry(
|
||||
"KEY001",
|
||||
lifecycle="fulltext_ready",
|
||||
has_pdf=True,
|
||||
ocr_status="done",
|
||||
deep_reading_status="pending",
|
||||
health={
|
||||
"pdf_health": "healthy",
|
||||
"ocr_health": "healthy",
|
||||
},
|
||||
maturity={"level": 3, "blocking": ["deep reading not done"]},
|
||||
next_step="/pf-deep",
|
||||
)
|
||||
]
|
||||
_write_index(vault, items)
|
||||
|
||||
captured_out: list[str] = []
|
||||
|
||||
def _capture_print(*args, **kwargs) -> None:
|
||||
captured_out.extend(str(a) for a in args)
|
||||
|
||||
import builtins
|
||||
original_print = builtins.print
|
||||
builtins.print = _capture_print
|
||||
|
||||
try:
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
finally:
|
||||
builtins.print = original_print
|
||||
|
||||
assert code == 0
|
||||
data = json.loads(" ".join(captured_out))
|
||||
readiness = data["_ai_readiness"]
|
||||
assert readiness["ai_context_ready"] is False
|
||||
assert "blocking_explanation" in readiness
|
||||
assert "deep reading" in readiness["blocking_explanation"].lower()
|
||||
|
||||
|
||||
class TestContextNoEntriesMatch:
|
||||
"""Empty filter results."""
|
||||
|
||||
def test_domain_no_match_returns_empty_array(self, tmp_path: Path, capsys) -> None:
|
||||
"""When --domain matches nothing, print [] and return 0."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
items = [_make_entry("KEY001", domain="骨科")]
|
||||
_write_index(vault, items)
|
||||
|
||||
code = run(_make_args(domain="Nonexistent", vault_path=vault))
|
||||
assert code == 0
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out.strip() == "[]"
|
||||
|
||||
def test_no_mode_specified_returns_error(self, tmp_path: Path, capsys) -> None:
|
||||
"""No key, no flags -> error message and exit code 1."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
_write_index(vault, []) # empty index
|
||||
|
||||
code = run(
|
||||
argparse.Namespace(
|
||||
key=None,
|
||||
domain=None,
|
||||
collection=None,
|
||||
all=False,
|
||||
vault_path=vault,
|
||||
)
|
||||
)
|
||||
assert code == 1
|
||||
captured = capsys.readouterr()
|
||||
assert "Specify" in captured.err
|
||||
|
||||
|
||||
class TestContextIndexMissing:
|
||||
"""Missing index file."""
|
||||
|
||||
def test_index_missing_returns_error(self, tmp_path: Path, capsys) -> None:
|
||||
"""When no index exists, return 1 with guidance message."""
|
||||
from paperforge.commands.context import run
|
||||
|
||||
vault = _minimal_vault(tmp_path)
|
||||
# Do NOT write an index
|
||||
|
||||
code = run(_make_args(key="KEY001", vault_path=vault))
|
||||
assert code == 1
|
||||
captured = capsys.readouterr()
|
||||
assert "not found" in captured.err.lower()
|
||||
Loading…
Reference in a new issue