tests/test_ingest_completion.py
Rama main · réplica de lectura del repositorio original.
"""Saved cards must survive index failure, and automation must report that failure."""
from __future__ import annotations
import contextlib
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
from _isolation import isolated_env, needs_posix_exec
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import local_ingest
import fetch_github_repos as github
import fetch_youtube_playlist as youtube
import xkb_frontmatter
import xkb_index
CARD = """---
id: sample
title: Sample evidence
type: knowledge-card
source_type: local-paper
category: research
tags: [original]
---
# Sample evidence
## 7. 雙語摘要
ZH: 範例知識
EN: Sample knowledge
"""
class IngestCompletion(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
self.cards = self.root / "cards"
self.bookmarks = self.root / "bookmarks"
self.source = self.root / "PMC12345.txt"
self.source.write_text("Some source evidence.", encoding="utf-8")
self.stack = contextlib.ExitStack()
self.addCleanup(self.stack.close)
self.stack.enter_context(mock.patch("gbrain_publish.check_backend"))
self.stdout, self.stderr = io.StringIO(), io.StringIO()
self.stack.enter_context(contextlib.redirect_stdout(self.stdout))
self.stack.enter_context(contextlib.redirect_stderr(self.stderr))
for module in (local_ingest, github, youtube):
self.stack.enter_context(mock.patch.object(module, "INDEX_FILE", self.bookmarks / "search_index.json"))
self.stack.enter_context(mock.patch.object(module, "_gbrain_put"))
self.stack.enter_context(mock.patch.object(module, "find_related_context", return_value="related evidence"))
self.stack.enter_context(mock.patch.object(module, "classify_content", return_value={"category": "research"}))
self.stack.enter_context(mock.patch.object(local_ingest, "CARDS_DIR", self.cards))
self.stack.enter_context(mock.patch.object(github, "CARDS_DIR", self.cards))
self.stack.enter_context(mock.patch.object(github, "GITHUB_RAW_DIR", self.bookmarks / "github"))
self.stack.enter_context(mock.patch.object(youtube, "YOUTUBE_DIR", self.bookmarks / "youtube"))
self.local_llm = self.stack.enter_context(mock.patch.object(local_ingest, "llm_call", return_value=CARD))
self.stack.enter_context(mock.patch.object(github, "_llm_call", return_value=CARD))
self.stack.enter_context(mock.patch.object(github, "load_env_key", return_value="test-only"))
self.stack.enter_context(mock.patch.object(github, "get_readme_preview", return_value="Readme evidence"))
self.stack.enter_context(mock.patch.object(github, "gh_api", return_value=[{
"full_name": "sample/repo", "html_url": "https://github.com/sample/repo",
"description": "Example repository", "language": "Python",
}]))
self.stack.enter_context(mock.patch.object(youtube, "get_playlist_videos", return_value=[
{"id": "sample-video", "title": "Sample video", "duration": 120},
]))
self.stack.enter_context(mock.patch.object(youtube, "download_subtitles", return_value=("Evidence", "en")))
self.stack.enter_context(mock.patch.object(youtube, "generate_card", return_value=CARD))
def test_all_ingesters_preserve_cards_and_fail_when_index_fails(self):
cases = [
(local_ingest, [str(self.source)], self.cards / f"{local_ingest.card_id_for_file(self.source)}.md", 2),
(github, ["--forks"], self.cards / "github_fork-sample-repo.md", 2),
(youtube, ["--playlist", "https://example.test/playlist"], self.bookmarks / "youtube/sample-video.md", 0),
]
for module, args, card, success_code in cases:
for rebuilt in (False, True):
with self.subTest(module=module.__name__, rebuilt=rebuilt):
if card.exists():
card.unlink()
self.stdout.seek(0)
self.stdout.truncate()
self.stderr.seek(0)
self.stderr.truncate()
with mock.patch.object(sys, "argv", [module.__name__, *args]), \
mock.patch.object(xkb_index, "rebuild", return_value=rebuilt) as rebuild:
code = module.main()
self.assertEqual(code, success_code if rebuilt else 1)
rebuild.assert_called_once_with()
self.assertIn("Sample evidence", card.read_text(encoding="utf-8"))
if not rebuilt:
self.assertIn("card(s) saved", self.stderr.getvalue())
self.assertIn("xkb_index.py --rebuild", self.stderr.getvalue())
self.assertNotIn("✅ 完成", self.stdout.getvalue())
def test_long_local_dry_run_does_not_call_models_or_write_cards(self):
self.source.write_text("Long evidence. " * 2000, encoding="utf-8")
with mock.patch.object(sys, "argv", ["local_ingest", str(self.source), "--dry-run"]), \
mock.patch.object(local_ingest, "condense_long_content") as condense, \
mock.patch.object(xkb_index, "rebuild") as rebuild:
self.assertEqual(local_ingest.main(), 0)
condense.assert_not_called()
self.local_llm.assert_not_called()
rebuild.assert_not_called()
self.assertFalse(self.cards.exists())
def test_local_shared_prompt_retains_source_and_persists_overrides(self):
result = local_ingest.process_file(self.source, "", set(), [], "learning", ["personal"], False)
prompt = self.local_llm.call_args.args[0]
self.assertIn(self.source.name, prompt)
self.assertIn("https://www.ncbi.nlm.nih.gov/pmc/articles/PMC12345/", prompt)
self.assertIn("source_type: local-paper", prompt)
self.assertIn("Media Evidence", prompt)
self.assertIn("related evidence", prompt)
content = next(self.cards.iterdir()).read_text(encoding="utf-8")
self.assertEqual(xkb_frontmatter.get(content, "category"), "learning")
self.assertEqual(xkb_frontmatter.get(content, "tags"), '[original, "personal"]')
self.assertEqual(result["summary"], "範例知識 | Sample knowledge")
def test_appending_tags_preserves_empty_fields_and_quoted_commas(self):
for tags, expected in (
("", '["personal"]'),
('["research, methods", original]', '["research, methods", original, "personal"]'),
("['research, methods', original]", "['research, methods', original, \"personal\"]"),
):
with self.subTest(tags=tags):
# Each subcase is a fresh ingest; saved cards now retry publication
# instead of being silently regenerated.
for old_card in self.cards.glob("*.md"):
old_card.unlink()
self.local_llm.return_value = CARD.replace("tags: [original]", f"tags: {tags}\nsensitivity: public")
local_ingest.process_file(self.source, "", set(), [], "learning", ["personal"], False)
content = next(self.cards.iterdir()).read_text(encoding="utf-8")
self.assertEqual(xkb_frontmatter.get(content, "tags"), expected)
self.assertEqual(xkb_frontmatter.get(content, "sensitivity"), "public")
@needs_posix_exec
class IngestShellContracts(unittest.TestCase):
def test_sync_wrappers_propagate_failure_and_accept_new_cards(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
bin_dir = root / "bin"
bin_dir.mkdir()
trace = root / "calls"
fake_python = bin_dir / "python3"
fake_python.write_text(
'#!/bin/bash\necho "$1" >> "$CALL_TRACE"\n'
'case "$1" in *fetch_*) exit "$FETCH_STATUS" ;; esac\nexit 0\n',
encoding="utf-8",
)
fake_python.chmod(0o755)
# A vector file makes the YouTube wrapper attempt the next stage on success.
vector = root / "memory/bookmarks/vector_index.json"
vector.parent.mkdir(parents=True)
vector.write_text("{}", encoding="utf-8")
for script, status, expected, vector_expected in (
("run_github_sync.sh", 2, 0, True),
("run_github_sync.sh", 0, 0, False),
("run_github_sync.sh", 1, 1, False),
("run_youtube_sync.sh", 0, 0, True),
("run_youtube_sync.sh", 1, 1, False),
):
with self.subTest(script=script, status=status):
trace.write_text("", encoding="utf-8")
env = isolated_env(root, PATH=f"{bin_dir}:{os.environ['PATH']}",
OPENCLAW_WORKSPACE=str(root), FETCH_STATUS=str(status), CALL_TRACE=str(trace),
XKB_GITHUB_LOCK=str(root / "github.lock"), XKB_GITHUB_LOG=str(root / "github.log"),
XKB_YOUTUBE_LOG=str(root / "youtube.log"))
run = subprocess.run(["bash", str(ROOT / "scripts" / script)], env=env,
capture_output=True, text=True, encoding="utf-8", timeout=30)
self.assertEqual(run.returncode, expected, run.stderr)
self.assertEqual("build_vector_index.py" in trace.read_text(encoding="utf-8"), vector_expected)
def test_repair_and_search_use_configured_data_without_regenerating_cards(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
data = root / "custom data"
card = data / "cards/sample.md"
card.parent.mkdir(parents=True)
card.write_text(CARD, encoding="utf-8")
before = card.stat().st_mtime_ns
config = root / "config.json"
config.write_text(json.dumps({"data_dir": str(data)}), encoding="utf-8")
env = isolated_env(root, PATH=os.environ["PATH"], XKB_CONFIG=str(config), PYTHONIOENCODING="utf-8")
repair = subprocess.run([sys.executable, str(ROOT / "scripts/xkb_index.py"), "--rebuild"],
env=env, capture_output=True, text=True, encoding="utf-8", timeout=30)
self.assertEqual(repair.returncode, 0, repair.stderr)
index = data / "bookmarks/search_index.json"
self.assertEqual(json.loads(index.read_text(encoding="utf-8"))["count"], 1)
self.assertEqual(card.stat().st_mtime_ns, before)
# Also prove the search wrapper can build its missing index at this path.
index.unlink()
search = subprocess.run(["bash", str(ROOT / "scripts/search_bookmarks.sh"), "Sample", "evidence"],
env=env, capture_output=True, text=True, encoding="utf-8", timeout=30)
self.assertEqual(search.returncode, 0, search.stderr)
self.assertIn("Sample evidence", search.stdout)
self.assertTrue(index.exists())
if __name__ == "__main__":
unittest.main()