mirror of
https://github.com/lllin000/PaperForge.git
synced 2026-07-22 06:50:53 +00:00
284f074389
- Replace CST (UTC+8) with UTC timezone using timezone.utc - Add _escape_md() helper to escape markdown special chars in QA fields - Wrap JSON+MD read-modify-write in filelock.FileLock with 10s timeout - Add 5 new tests: UTC timestamp, MD escaping, CJK escaping, lock release, lock timeout
381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""Unit tests for paperforge/worker/discussion.py -- AI Discussion Recorder.
|
|
|
|
Covers: record_session() creation, append, atomic writes, error handling,
|
|
CJK encoding, and CLI invocation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import filelock
|
|
|
|
import pytest
|
|
|
|
from paperforge.worker.discussion import record_session
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _create_minimal_vault(tmp_path: Path, zotero_key: str = "TSTONE001",
|
|
domain: str = "骨科",
|
|
title: str = "Biomechanical Comparison") -> Path:
|
|
"""Create a minimal vault structure for testing record_session()."""
|
|
vault = tmp_path / "vault"
|
|
resources = vault / "03_Resources"
|
|
literature = resources / "Literature"
|
|
control = resources / "LiteratureControl"
|
|
records_dir = control / "library-records"
|
|
domain_dir = records_dir / domain
|
|
domain_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Library record with frontmatter
|
|
record_path = domain_dir / f"{zotero_key}.md"
|
|
record_path.write_text(
|
|
"---\n"
|
|
f'zotero_key: "{zotero_key}"\n'
|
|
f'domain: "{domain}"\n'
|
|
f'title: "{title}"\n'
|
|
"year: \"2024\"\n"
|
|
"analyze: true\n"
|
|
"---\n\n"
|
|
f"# {title}\n\n"
|
|
"正式库控制记录。\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Create paperforge.json
|
|
pf_json = vault / "paperforge.json"
|
|
pf_json.write_text(
|
|
json.dumps({
|
|
"version": "1.4.15",
|
|
"system_dir": "99_System",
|
|
"resources_dir": "03_Resources",
|
|
"literature_dir": "Literature",
|
|
"control_dir": "LiteratureControl",
|
|
"base_dir": "05_Bases",
|
|
"skill_dir": ".opencode/skills",
|
|
}, indent=2, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Create canonical index (Phase 38: _find_paper_metadata reads from index)
|
|
indexes_dir = vault / "99_System" / "PaperForge" / "indexes"
|
|
indexes_dir.mkdir(parents=True, exist_ok=True)
|
|
canonical_index = {
|
|
"schema_version": "2",
|
|
"generated_at": "2026-05-07T00:00:00+08:00",
|
|
"paperforge_version": "1.4.15",
|
|
"paper_count": 1,
|
|
"items": [
|
|
{
|
|
"zotero_key": zotero_key,
|
|
"domain": domain,
|
|
"title": title,
|
|
"ai_path": f"Literature/{domain}/{zotero_key} - {title}/ai/",
|
|
}
|
|
],
|
|
}
|
|
(indexes_dir / "formal-library.json").write_text(
|
|
json.dumps(canonical_index, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
return vault
|
|
|
|
|
|
def _sample_qa_pairs() -> list[dict]:
|
|
return [
|
|
{
|
|
"question": "What is the primary outcome?",
|
|
"answer": "The primary outcome is biomechanical strength.",
|
|
"source": "user_question",
|
|
"timestamp": "2026-05-06T12:00:00+08:00",
|
|
},
|
|
{
|
|
"question": "Any limitations?",
|
|
"answer": "Small sample size and in vitro design.",
|
|
"source": "agent_analysis",
|
|
"timestamp": "2026-05-06T12:01:00+08:00",
|
|
},
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRecordSession:
|
|
"""record_session() behavior tests."""
|
|
|
|
def test_create_both_files(self, tmp_path: Path) -> None:
|
|
"""Test 1: Creates ai/discussion.json and ai/discussion.md with valid data."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
result = record_session(
|
|
vault_path=vault,
|
|
zotero_key="TSTONE001",
|
|
agent="pf-paper",
|
|
model="gpt-4",
|
|
qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
json_path = Path(result["json_path"])
|
|
md_path = Path(result["md_path"])
|
|
|
|
# JSON file exists and has correct structure
|
|
assert json_path.exists()
|
|
data = json.loads(json_path.read_text(encoding="utf-8"))
|
|
assert data["schema_version"] == "1"
|
|
assert data["paper_key"] == "TSTONE001"
|
|
assert len(data["sessions"]) == 1
|
|
session = data["sessions"][0]
|
|
assert len(session["session_id"]) == 36 # UUID length
|
|
assert session["agent"] == "pf-paper"
|
|
assert session["model"] == "gpt-4"
|
|
assert "started" in session
|
|
assert session["paper_key"] == "TSTONE001"
|
|
assert session["paper_title"] == "Biomechanical Comparison"
|
|
assert session["domain"] == "骨科"
|
|
assert len(session["qa_pairs"]) == 2
|
|
assert session["qa_pairs"][0]["source"] == "user_question"
|
|
assert session["qa_pairs"][1]["source"] == "agent_analysis"
|
|
|
|
# MD file exists and has correct format
|
|
assert md_path.exists()
|
|
md_content = md_path.read_text(encoding="utf-8")
|
|
assert "# AI Discussion Record: Biomechanical Comparison" in md_content
|
|
assert "## " in md_content # session heading with date
|
|
assert "**问题:**" in md_content
|
|
assert "**解答:**" in md_content
|
|
assert "---" in md_content # separator
|
|
|
|
def test_append_second_session(self, tmp_path: Path) -> None:
|
|
"""Test 2: Second call appends, does not overwrite."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
qa = _sample_qa_pairs()
|
|
|
|
# First call
|
|
r1 = record_session(vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=qa)
|
|
assert r1["status"] == "ok"
|
|
|
|
# Second call with different Q&A
|
|
qa2 = [{"question": "Q2?", "answer": "A2.", "source": "user_question",
|
|
"timestamp": "2026-05-06T13:00:00+08:00"}]
|
|
r2 = record_session(vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=qa2)
|
|
assert r2["status"] == "ok"
|
|
|
|
# JSON: sessions length = 2, first session preserved
|
|
json_path = Path(r2["json_path"])
|
|
data = json.loads(json_path.read_text(encoding="utf-8"))
|
|
assert len(data["sessions"]) == 2
|
|
assert len(data["sessions"][0]["qa_pairs"]) == 2 # first session preserved
|
|
assert data["sessions"][1]["qa_pairs"][0]["question"] == "Q2?"
|
|
|
|
# MD: two ## session headings
|
|
md_content = Path(r2["md_path"]).read_text(encoding="utf-8")
|
|
assert md_content.count("## ") >= 2
|
|
|
|
def test_utc_timestamp(self, tmp_path: Path) -> None:
|
|
"""HARDEN-03: started timestamp uses UTC (not CST/UTC+8)."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
result = record_session(
|
|
vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
assert result["status"] == "ok"
|
|
data = json.loads(Path(result["json_path"]).read_text(encoding="utf-8"))
|
|
started = data["sessions"][0]["started"]
|
|
# UTC offset is either +00:00 or Z
|
|
assert started.endswith("+00:00") or started.endswith("Z"), f"Expected UTC, got: {started}"
|
|
|
|
def test_markdown_escaping(self, tmp_path: Path) -> None:
|
|
"""HARDEN-02: Markdown special chars in QA are escaped in discussion.md."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
qa = [
|
|
{
|
|
"question": "What does *italic* and **bold** mean?",
|
|
"answer": "Use # for headings and [link](url) syntax _underscore_ `code`",
|
|
"source": "user_question",
|
|
"timestamp": "2026-05-06T12:00:00+00:00",
|
|
},
|
|
]
|
|
result = record_session(
|
|
vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=qa,
|
|
)
|
|
assert result["status"] == "ok"
|
|
md = Path(result["md_path"]).read_text(encoding="utf-8")
|
|
# Verify escaped forms exist (not bare markdown)
|
|
assert "\\*italic\\*" in md, f"Expected escaped asterisks in: {md}"
|
|
assert "\\*\\*bold\\*\\*" in md
|
|
assert "\\_" in md
|
|
assert "\\`" in md
|
|
|
|
def test_markdown_escaping_cjk(self, tmp_path: Path) -> None:
|
|
"""HARDEN-02: CJK characters preserved alongside escaped markdown chars."""
|
|
vault = _create_minimal_vault(tmp_path, title="中文测试")
|
|
qa = [
|
|
{
|
|
"question": "什么是*p值*和#显著性？",
|
|
"answer": "_效应量_为0.5`[95% CI]",
|
|
"source": "user_question",
|
|
"timestamp": "2026-05-06T12:00:00+00:00",
|
|
},
|
|
]
|
|
result = record_session(
|
|
vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=qa,
|
|
)
|
|
assert result["status"] == "ok"
|
|
md = Path(result["md_path"]).read_text(encoding="utf-8")
|
|
# CJK preserved
|
|
assert "什么是" in md
|
|
assert "效应量" in md
|
|
# Special chars escaped
|
|
assert "\\*p值\\*" in md
|
|
assert "\\_" in md
|
|
assert "\\`" in md
|
|
|
|
def test_file_lock_prevents_concurrent_write(self, tmp_path: Path) -> None:
|
|
"""HARDEN-01: File lock (.json.lock) is created during record_session()."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
result = record_session(
|
|
vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
assert result["status"] == "ok"
|
|
json_path = Path(result["json_path"])
|
|
# Lock file should NOT remain after successful write (released + cleaned up)
|
|
lock_path = json_path.with_suffix(".json.lock")
|
|
assert not lock_path.exists(), "Lock file should be released after write"
|
|
|
|
def test_lock_timeout_returns_error(self, tmp_path: Path) -> None:
|
|
"""HARDEN-01: When lock cannot be acquired, returns error status."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
# First call to create the json file
|
|
r1 = record_session(
|
|
vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
assert r1["status"] == "ok"
|
|
json_path = Path(r1["json_path"])
|
|
|
|
# Hold an exclusive lock to simulate concurrent access
|
|
lock_path = json_path.with_suffix(".json.lock")
|
|
external_lock = filelock.FileLock(lock_path, timeout=1)
|
|
with external_lock:
|
|
# This should fail because lock is held
|
|
r2 = record_session(
|
|
vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
assert r2["status"] == "error"
|
|
assert "Concurrent access" in r2.get("message", "")
|
|
|
|
def test_missing_vault(self, tmp_path: Path) -> None:
|
|
"""Test 3: Non-existent vault returns error status."""
|
|
result = record_session(
|
|
vault_path=tmp_path / "nonexistent",
|
|
zotero_key="TSTONE001",
|
|
agent="pf-paper",
|
|
model="gpt-4",
|
|
qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
assert result["status"] == "error"
|
|
assert "message" in result
|
|
|
|
def test_unknown_key(self, tmp_path: Path) -> None:
|
|
"""Test 4: Unknown zotero_key returns error without creating files."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
result = record_session(
|
|
vault_path=vault,
|
|
zotero_key="UNKNOW01",
|
|
agent="pf-paper",
|
|
model="gpt-4",
|
|
qa_pairs=_sample_qa_pairs(),
|
|
)
|
|
assert result["status"] == "error"
|
|
|
|
# No files created
|
|
ai_dir = vault / "03_Resources" / "Literature" / "UNKNOW01 - untitled" / "ai"
|
|
assert not ai_dir.exists() or not list(ai_dir.iterdir()) == []
|
|
|
|
def test_cjk_encoding(self, tmp_path: Path) -> None:
|
|
"""Test 5: CJK content round-trips correctly via ensure_ascii=False."""
|
|
vault = _create_minimal_vault(tmp_path, title="中文测试论文")
|
|
cjk_qa = [
|
|
{
|
|
"question": "问题一",
|
|
"answer": "答案一",
|
|
"source": "user_question",
|
|
"timestamp": "2026-05-06T12:00:00+08:00",
|
|
},
|
|
]
|
|
result = record_session(vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=cjk_qa)
|
|
assert result["status"] == "ok"
|
|
|
|
# JSON read-back
|
|
json_path = Path(result["json_path"])
|
|
raw = json_path.read_text(encoding="utf-8")
|
|
assert "中文测试论文" in raw
|
|
assert "问题一" in raw
|
|
assert "答案一" in raw
|
|
|
|
# Parse and verify
|
|
data = json.loads(raw)
|
|
assert data["sessions"][0]["paper_title"] == "中文测试论文"
|
|
assert data["sessions"][0]["qa_pairs"][0]["question"] == "问题一"
|
|
|
|
def test_atomic_write_no_partial(self, tmp_path: Path) -> None:
|
|
"""Test 6: Atomic write via tempfile + os.replace prevents partial writes."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
result = record_session(vault_path=vault, zotero_key="TSTONE001",
|
|
agent="pf-paper", model="gpt-4", qa_pairs=_sample_qa_pairs())
|
|
assert result["status"] == "ok"
|
|
|
|
json_path = Path(result["json_path"])
|
|
md_path = Path(result["md_path"])
|
|
|
|
# Verify files are valid (not partial)
|
|
data = json.loads(json_path.read_text(encoding="utf-8"))
|
|
assert data["schema_version"] == "1"
|
|
assert md_path.read_text(encoding="utf-8").startswith("# AI Discussion Record")
|
|
|
|
def test_cli_invocation(self, tmp_path: Path) -> None:
|
|
"""Test 7: CLI `python -m paperforge.worker.discussion record` works."""
|
|
vault = _create_minimal_vault(tmp_path)
|
|
import subprocess
|
|
import sys
|
|
|
|
qa_json = json.dumps(_sample_qa_pairs(), ensure_ascii=False)
|
|
cmd = [
|
|
sys.executable, "-m", "paperforge.worker.discussion", "record",
|
|
"TSTONE001",
|
|
"--vault", str(vault),
|
|
"--agent", "pf-paper",
|
|
"--model", "gpt-4",
|
|
"--qa-pairs", qa_json,
|
|
]
|
|
# Use binary pipes to avoid UTF-8 decode issues on Windows (Python 3.14+)
|
|
env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
|
|
proc = subprocess.run(cmd, capture_output=True, env=env)
|
|
proc_stdout = proc.stdout.decode("utf-8", errors="replace")
|
|
proc_stderr = proc.stderr.decode("utf-8", errors="replace")
|
|
assert proc.returncode == 0, f"CLI failed: {proc_stderr}"
|
|
|
|
output = json.loads(proc_stdout)
|
|
assert output["status"] == "ok"
|
|
json_path = Path(output["json_path"])
|
|
assert json_path.exists()
|
|
data = json.loads(json_path.read_text(encoding="utf-8"))
|
|
assert len(data["sessions"]) == 1
|
|
assert data["sessions"][0]["agent"] == "pf-paper"
|