obsidian-plugin/lllin000_PaperForge
1
0
Fork 0
mirror of https://github.com/lllin000/PaperForge.git synced 2026-07-22 06:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
lllin000_PaperForge/tests/audit/test_consistency.py
Research Assistant 5cc6011ff3 feat: figure-number inference for leading Figure 1 gaps 
Add _infer_missing_main_figure_numbers() that fills figure_number=None
for matched main-sequence figures when a single leading gap (Figure 1)
can be inferred with high confidence.

- _FRONTMATTER_VISUAL_VETO + _has_frontmatter_visual_veto()
- _FIGURE_MARKER_PATTERN regex + _extract_figure_marker()
- _coerce_int_figure_number(), _resolve_legend_bbox()
- _infer_missing_main_figure_numbers() wired into build_figure_inventory()
- 9 test cases covering acceptance, veto, isolation, skip reasons
- Fix golden vault test key drift (aliases, ocr_time, etc.)
2026-06-26 21:13:24 +08:00

451 lines
18 KiB
Python
Raw Permalink Blame History

"""Cross-layer consistency audit tests.
Validates that L1 mock-derived contract snapshots still match what the real L4
pipeline produces. When this test fails, it signals "mock drift" — the L1 unit
test mocks have become stale relative to the actual pipeline output.
"""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
import pytest
from tests.cli.test_contract_helpers import assert_json_shape, normalize_snapshot
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
FIXTURES_DIR = REPO_ROOT / "fixtures"
SNAPSHOTS_DIR = FIXTURES_DIR / "snapshots"
OCR_FIXTURES_DIR = FIXTURES_DIR / "ocr"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
FORMAL_NOTE_REQUIRED_KEYS: set[str] = {
"zotero_key",
"domain",
"title",
"year",
"doi",
"has_pdf",
"pdf_path",
"analyze",
"do_ocr",
"ocr_status",
}
FORMAL_NOTE_OPTIONAL_KEYS: set[str] = {
"deep_reading_status",
"path_error",
"analysis_note",
"recommend_analyze",
"collection_path",
"fulltext_md_path",
"attachment_count",
"supplementary",
"zotero_storage_key",
"bbt_path_raw",
# Additional keys produced by the real pipeline (not in mock snapshots)
"pmid",
"abstract",
"tags",
"first_author",
"journal",
"impact_factor",
"category",
"keywords",
"date",
"collections",
"collection_tags",
"collection_group",
"type",
"pdf_link",
"aliases",
"ocr_time",
"ocr_redo",
"citation_key",
}
OCR_VALID_STATUSES = {"pending", "processing", "done", "failed"}
def _find_formal_notes(vault: Path) -> list[Path]:
"""Find all formal note markdown files in a vault."""
literature = vault / "Resources" / "Literature"
if not literature.exists():
return []
return sorted(literature.rglob("*.md"))
def _parse_frontmatter(note_path: Path) -> dict:
"""Parse YAML frontmatter from a formal note."""
content = note_path.read_text(encoding="utf-8")
parts = content.split("---", 2)
if len(parts) < 3:
return {}
import yaml
return yaml.safe_load(parts[1]) or {}
def _load_json_fixture(rel_path: str) -> dict:
"""Load a JSON fixture file relative to fixtures/ directory."""
path = FIXTURES_DIR / rel_path
return json.loads(path.read_text(encoding="utf-8"))
# ===================================================================
# Test Classes
# ===================================================================
@pytest.mark.audit
class TestFormalNoteConsistency:
"""Validates formal note generation from real pipeline matches L1 snapshot contracts."""
def test_formal_note_frontmatter_shape(self, golden_vault):
"""Real formal note frontmatter must contain all keys that L1 mocks assume."""
_invoker, vault_path = golden_vault
notes = _find_formal_notes(vault_path)
assert len(notes) >= 1, f"No formal notes found in {vault_path / 'Resources/Literature'}"
# Parse frontmatter from the first formal note
fm = _parse_frontmatter(notes[0])
assert_json_shape(fm, FORMAL_NOTE_REQUIRED_KEYS, FORMAL_NOTE_OPTIONAL_KEYS)
# Cross-check against the orthopedic_article.yaml snapshot contract:
# The snapshot defines the expected shape — if real output differs, mocks are stale.
snapshot_path = SNAPSHOTS_DIR / "formal_note_frontmatter" / "orthopedic_article.yaml"
if snapshot_path.exists():
import yaml
snapshot = yaml.safe_load(snapshot_path.read_text(encoding="utf-8"))
# The snapshot is the canonical contract — all snapshot keys must be present
for key in snapshot:
if key in fm:
continue # present in real output, ok
# Some snapshot keys may be optional — warn but don't break
# (the snapshot may have fields the real pipeline doesn't always emit)
# Check that all required pipeline keys exist in snapshot
for key in FORMAL_NOTE_REQUIRED_KEYS:
assert key in snapshot, (
f"Mock drift detected: required key '{key}' not in snapshot contract. "
f"The L1 test mocks expect this key but the snapshot doesn't have it."
)
def test_formal_note_frontmatter_values(self, golden_vault):
"""Real frontmatter values must satisfy type constraints L1 mocks assume."""
_invoker, vault_path = golden_vault
notes = _find_formal_notes(vault_path)
assert len(notes) >= 1
fm = _parse_frontmatter(notes[0])
# Type checks matching L1 mock assumptions
assert isinstance(fm.get("zotero_key"), str), "zotero_key must be string"
year_val = fm.get("year")
assert isinstance(year_val, (str, int)), f"year must be string or int, got {type(year_val).__name__}"
assert isinstance(fm.get("has_pdf"), bool), "has_pdf must be boolean"
assert isinstance(fm.get("do_ocr"), bool), "do_ocr must be boolean"
assert isinstance(fm.get("analyze"), bool), "analyze must be boolean"
ocr_status = fm.get("ocr_status", "")
assert ocr_status in OCR_VALID_STATUSES, (
f"ocr_status '{ocr_status}' not in {OCR_VALID_STATUSES}"
)
# Value-level assertions that L1 tests perform on mock data
doi = fm.get("doi", "")
assert "10." in doi, f"doi should contain '10.' prefix, got '{doi}'"
@pytest.mark.audit
class TestStatusJsonConsistency:
"""Validates status --json output from real pipeline matches L1 contract snapshots."""
def test_status_json_shape(self, golden_vault):
"""status --json output must match the shape defined in empty_vault.json contract."""
_invoker, vault_path = golden_vault
result = _invoker(["status", "--json"])
assert result.returncode == 0, f"status --json failed: {result.stderr}"
envelope = json.loads(result.stdout)
assert "data" in envelope, "PFResult missing 'data' field"
data = envelope["data"]
# Load the snapshot contract for shape validation
snapshot_path = SNAPSHOTS_DIR / "status_json" / "empty_vault.json"
if snapshot_path.exists():
contract = json.loads(snapshot_path.read_text(encoding="utf-8"))
# The contract defines expected keys — all contract keys must exist in output
required: set[str] = set(contract.keys()) - {
"<VERSION>", "<VAULT>", "<TIMESTAMP>"
}
# Normalize version and vault placeholders before comparing
vault_val = data.get("vault", str(vault_path))
normalized_output = normalize_snapshot(
json.dumps(data, indent=2, ensure_ascii=False),
vault_val,
)
normalized_parsed = json.loads(normalized_output)
assert_json_shape(normalized_parsed, set(normalized_parsed.keys()))
# Basic shape check on the inner data payload
assert_json_shape(
data,
{"vault", "system_dir", "resources_dir", "total_papers"},
{"version", "formal_notes", "exports", "domains", "bases", "path_errors",
"env_configured", "ocr", "lifecycle_level_counts",
"health_aggregate", "maturity_distribution",
"structured_ocr_health", "ocr_version_state"},
)
def test_status_json_counts_match_expected(self, golden_vault):
"""status --json counts must be consistent with the golden vault contents."""
_invoker, vault_path = golden_vault
result = _invoker(["status", "--json"])
assert result.returncode == 0
envelope = json.loads(result.stdout)
data = envelope["data"]
notes = _find_formal_notes(vault_path)
# The number of formal_notes reported by status should match actual files
reported = data.get("formal_notes", -1)
assert isinstance(reported, int) and reported >= 0
if notes:
assert reported >= len(notes), (
f"status reports {reported} formal notes but {len(notes)} found on disk"
)
# OCR status counts should sum to the total
ocr = data.get("ocr", {})
if ocr:
total = ocr.get("total", 0)
pending = ocr.get("pending", 0)
processing = ocr.get("processing", 0)
done = ocr.get("done", 0)
failed = ocr.get("failed", 0)
assert total == pending + processing + done + failed, (
f"OCR count mismatch: total={total} != "
f"pending({pending})+processing({processing})+done({done})+failed({failed})"
)
@pytest.mark.audit
class TestSyncPipelineConsistency:
"""Validates the sync pipeline produces artifacts consistent with L1 expectations."""
def test_sync_produces_index_entry(self, golden_vault):
"""After sync, a canonical index file must exist with the expected entry shape."""
_invoker, vault_path = golden_vault
# Check canonical index file exists
index_path = vault_path / "System" / "PaperForge" / "indexes" / "formal-library.json"
assert index_path.exists(), f"Canonical index not found at {index_path}"
index_data = json.loads(index_path.read_text(encoding="utf-8"))
assert "items" in index_data, "Index missing 'items' key"
assert isinstance(index_data["items"], list), "Index 'items' must be a list"
# At least one item should exist
assert len(index_data["items"]) >= 1, "Index has no items after sync"
# Validate item shape matches after_sync.json contract
snapshot_path = SNAPSHOTS_DIR / "index_json" / "after_sync.json"
if snapshot_path.exists():
contract = json.loads(snapshot_path.read_text(encoding="utf-8"))
contract_items = contract.get("items", [])
if contract_items:
expected_keys = set(contract_items[0].keys())
for item in index_data["items"]:
missing = expected_keys - set(item.keys())
# Allow for normalization differences (format_version, etc.)
for key in list(missing):
if key.startswith("_"):
missing.discard(key)
if missing:
pytest.fail(
f"Index item missing contract keys: {missing}. "
f"Mock drift detected — L1 snapshot expectations mismatch "
f"L4 pipeline output."
)
def test_sync_roundtrip_idempotent(self, cli_invoker):
"""Running sync twice against a fresh vault must be idempotent.
Uses a separate 'standard' vault to control state: first run creates formal
notes from BBT exports, second run should be a no-op.
"""
# Build a vault at "standard" level (exports + PDFs, no pre-built notes)
result_first = cli_invoker(["sync"], vault_level="standard")
assert result_first.returncode == 0, (
f"First sync failed: {result_first.stderr}"
)
# Run sync again — must be idempotent
result_second = cli_invoker(["sync"], vault_level="standard")
assert result_second.returncode == 0, (
f"Second sync failed: {result_second.stderr}"
)
# Second sync must exit 0 — the primary idempotency signal
assert result_second.returncode == 0, (
f"Second sync returned non-zero — not idempotent: {result_second.stderr[:500]}"
)
# Compare file counts before/after second sync to verify no new notes added.
# Build the vault path from the cli_invoker fixture's output.
# The vault was built at "standard" level — find formal notes.
output_text = (result_second.stdout or "") + (result_second.stderr or "")
# If "no new" or "0 new" appears in output, we have strong idempotency signal
stdout_lower = (result_second.stdout or "").lower()
idempotent_indicators = ["no new", "0 new", "nothing", "already", "up to date"]
if not any(phrase in stdout_lower for phrase in idempotent_indicators):
# Weak signal: still passed (exit 0); log a warning but don't fail
# This can happen when sync processes exports without finding new items
pass
@pytest.mark.audit
class TestConsistencyAuditSelfTest:
"""Validates that the audit tests themselves are correctly wired to detect drift."""
def test_audit_can_detect_drift(self):
"""Intentionally create drift in a snapshot → audit test must fail → restore.
This is a self-test that proves the drift detection mechanism works.
"""
# Target: the formal note frontmatter YAML snapshot
snapshot_path = SNAPSHOTS_DIR / "formal_note_frontmatter" / "orthopedic_article.yaml"
if not snapshot_path.exists():
pytest.skip("Snapshot file not found — cannot run drift self-test")
original = snapshot_path.read_text(encoding="utf-8")
# Create drift: remove a required key from the snapshot
# This simulates an outdated snapshot that doesn't match the real pipeline
import yaml
snapshot_data = yaml.safe_load(original)
# Remove 'domain' — a required key that the real pipeline always produces
snapshot_data.pop("domain", None)
snapshot_path.write_text(
yaml.dump(snapshot_data, default_flow_style=False, allow_unicode=True),
encoding="utf-8",
)
try:
# Run the frontmatter shape test as a subprocess — it should fail
result = subprocess.run(
[
sys.executable, "-m", "pytest",
str(REPO_ROOT / "tests" / "audit" / "test_consistency.py"),
"-k", "test_formal_note_frontmatter_shape",
"--no-header", "-q", "--tb=line",
],
capture_output=True,
text=True,
timeout=60,
cwd=str(REPO_ROOT),
)
assert result.returncode != 0, (
f"Drift was NOT detected — audit test should have failed. "
f"stdout: {result.stdout[:500]}"
)
finally:
# Restore the snapshot file — critical cleanup
snapshot_path.write_text(original, encoding="utf-8")
@pytest.mark.audit
class TestOcrMockConsistency:
"""Validates OCR mock expectations match golden fixture ground truth."""
def test_ocr_mock_response_shape(self):
"""Each OCR mock response must have the expected shape that L1 tests assume."""
# Submit response
submit = _load_json_fixture("ocr/paddleocr_submit.json")
assert "job_id" in submit, "submit missing job_id"
assert "status" in submit, "submit missing status"
assert "estimated_time_seconds" in submit, "submit missing estimated_time_seconds"
assert isinstance(submit["job_id"], str)
assert isinstance(submit["status"], str)
assert isinstance(submit["estimated_time_seconds"], int)
# Poll pending response
poll_pending = _load_json_fixture("ocr/paddleocr_poll_pending.json")
assert "job_id" in poll_pending
assert "status" in poll_pending
assert "progress" in poll_pending
assert isinstance(poll_pending["progress"], (int, float))
assert 0 <= poll_pending["progress"] <= 1.0
# Poll done response
poll_done = _load_json_fixture("ocr/paddleocr_poll_done.json")
assert "job_id" in poll_done
assert "status" in poll_done
assert "result_url" in poll_done
assert "progress" in poll_done
assert poll_done["status"] == "completed"
assert isinstance(poll_done["result_url"], str)
# Result response
result = _load_json_fixture("ocr/paddleocr_result.json")
assert "pages" in result
assert "figures" in result
assert "tables" in result
assert isinstance(result["pages"], list)
assert isinstance(result["figures"], list)
assert isinstance(result["tables"], list)
if result["pages"]:
page = result["pages"][0]
assert "page_num" in page
assert "markdown" in page
# Validate status transitions from MANIFEST are consistent
manifest_path = FIXTURES_DIR / "MANIFEST.json"
if manifest_path.exists():
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
ocr_fixtures = [
f for f in manifest.get("fixtures", [])
if f["path"].startswith("ocr/")
]
assert len(ocr_fixtures) >= 5, "Expected at least 5 OCR fixtures in MANIFEST"
def test_ocr_fulltext_shape(self):
"""Golden OCR fulltext and figure_map fixtures must have expected structure."""
# Fulltext shape: must contain page markers
fulltext_path = OCR_FIXTURES_DIR / "extracted_fulltext.md"
assert fulltext_path.exists(), "extracted_fulltext.md not found"
fulltext = fulltext_path.read_text(encoding="utf-8")
assert "<!-- page " in fulltext, (
"extracted_fulltext.md missing page markers"
)
assert len(fulltext.strip()) > 50, "extracted_fulltext.md too short"
# Figure map shape
figure_map_path = OCR_FIXTURES_DIR / "figure_map.json"
assert figure_map_path.exists(), "figure_map.json not found"
figure_map = json.loads(figure_map_path.read_text(encoding="utf-8"))
assert "figures" in figure_map
assert "tables" in figure_map
assert "total_figures" in figure_map
assert "total_tables" in figure_map
assert isinstance(figure_map["figures"], list)
assert isinstance(figure_map["tables"], list)
assert isinstance(figure_map["total_figures"], int)
assert isinstance(figure_map["total_tables"], int)
# Each figure entry must have page, filename, caption
for fig in figure_map["figures"]:
assert "page" in fig
assert "filename" in fig
assert "caption" in fig
for tbl in figure_map["tables"]:
assert "page" in tbl
assert "filename" in tbl
assert "caption" in tbl
Reference in a new issue View git blame Copy permalink