Codex
6edf02c3eb
Default ticket, multipart, and text-note ingest entry points to staging without launching a job. Propagate WIKI_ARCHIVE_SOURCES from project configuration into the managed service environment.\n\nValidated with the full 206-test suite.
1042 lines
45 KiB
Python
1042 lines
45 KiB
Python
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from obsidian_wiki.mcp_ingest import (
|
|
IngestError,
|
|
IngestManager,
|
|
parse_bool,
|
|
)
|
|
|
|
|
|
class McpIngestTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tempdir.name)
|
|
self.vault = self.root / "vault"
|
|
self.project = self.root / "project"
|
|
self.runtime = self.project / ".runtime" / "mcp-ingest"
|
|
self.vault.mkdir()
|
|
self.project.mkdir()
|
|
self._write_opencode_skill_fixture(self.project)
|
|
self.manager = IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.runtime,
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
max_file_bytes=100,
|
|
max_chunk_bytes=8,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.manager.close()
|
|
self.tempdir.cleanup()
|
|
|
|
def test_parse_bool_accepts_enabled_values_only(self):
|
|
self.assertTrue(parse_bool("1"))
|
|
self.assertTrue(parse_bool("true"))
|
|
self.assertTrue(parse_bool("YES"))
|
|
self.assertFalse(parse_bool(""))
|
|
self.assertFalse(parse_bool("0"))
|
|
self.assertFalse(parse_bool("false"))
|
|
|
|
def test_close_still_closes_supervisor_when_queued_job_write_fails(self):
|
|
queued = {"state": "queued"}
|
|
with (
|
|
patch.object(self.manager, "_iter_jobs", return_value=[queued]),
|
|
patch.object(self.manager, "_write_job", side_effect=OSError("disk full")),
|
|
patch.object(self.manager._supervisor, "close") as supervisor_close,
|
|
):
|
|
with self.assertRaises(OSError):
|
|
self.manager.close()
|
|
|
|
supervisor_close.assert_called_once_with()
|
|
|
|
def test_uploads_are_staged_in_runtime_not_vault(self):
|
|
upload = self.manager.create_upload(file_name="drawing.pdf", size_bytes=0, sha256=self._sha256(b""))
|
|
upload_path = self._upload_payload_path(upload["upload_id"]).resolve()
|
|
|
|
self.assertNotIn("path", upload)
|
|
self.assertTrue(upload_path.is_relative_to((self.runtime / "uploads").resolve()))
|
|
self.assertFalse(upload_path.is_relative_to(self.vault.resolve()))
|
|
|
|
def test_runtime_root_must_not_be_inside_vault(self):
|
|
with self.assertRaises(IngestError):
|
|
IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.vault / ".runtime" / "mcp-ingest",
|
|
)
|
|
|
|
def test_check_opencode_readiness_static_success(self):
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run) as run_mock:
|
|
result = self.manager.check_opencode_readiness()
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertTrue(result["static_preflight"])
|
|
self.assertEqual(result["runtime_smoke_test"], "skipped")
|
|
self.assertEqual(result["opencode_version"], "1.17.13")
|
|
self.assertFalse(result["errors"])
|
|
self.assertTrue(any(path.endswith(".agents/skills/wiki-ingest/SKILL.md") for path in result["checked_paths"]))
|
|
self.assertEqual(run_mock.call_count, 2)
|
|
|
|
def test_check_opencode_readiness_reports_missing_skill_files(self):
|
|
missing_project = self.root / "missing-project"
|
|
missing_project.mkdir()
|
|
(missing_project / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
manager = IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=missing_project,
|
|
runtime_root=self.root / "runtime-missing",
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
)
|
|
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run):
|
|
result = manager.check_opencode_readiness()
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertFalse(result["static_preflight"])
|
|
self.assertIn(".agents/skills/wiki-ingest/SKILL.md", "\n".join(result["errors"]))
|
|
self.assertIn(".agents/skills/llm-wiki/SKILL.md", "\n".join(result["errors"]))
|
|
|
|
def test_check_opencode_readiness_smoke_test_success(self):
|
|
self.manager.opencode_model = "openai/gpt-5.5"
|
|
self.manager.opencode_variant = "high"
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_smoke_success_run) as run_mock:
|
|
result = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["runtime_smoke_test"], "passed")
|
|
self.assertEqual(run_mock.call_count, 3)
|
|
smoke_args = run_mock.call_args_list[-1].args[0]
|
|
self.assertEqual(smoke_args[:4], ["/usr/bin/opencode", "run", "--format", "json"])
|
|
self.assertEqual(smoke_args[smoke_args.index("--model") + 1], "openai/gpt-5.5")
|
|
self.assertEqual(smoke_args[smoke_args.index("--variant") + 1], "high")
|
|
self.assertIn("--auto", smoke_args)
|
|
self.assertIn("--dir", smoke_args)
|
|
self.assertIn("MCP ingest readiness smoke test", smoke_args)
|
|
|
|
def test_check_opencode_readiness_smoke_test_failure_and_timeout(self):
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_smoke_failure_run):
|
|
failed = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
self.assertFalse(failed["ok"])
|
|
self.assertEqual(failed["runtime_smoke_test"], "failed")
|
|
self.assertIn("SKILL_LOAD_FAILED", "\n".join(failed["errors"]))
|
|
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_smoke_timeout_run):
|
|
timed_out = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
self.assertFalse(timed_out["ok"])
|
|
self.assertEqual(timed_out["runtime_smoke_test"], "failed")
|
|
self.assertIn("timed out", "\n".join(timed_out["errors"]))
|
|
|
|
def test_start_ingest_job_fails_before_opencode_when_static_readiness_fails(self):
|
|
payload = b"source data"
|
|
upload = self._complete_upload("source.pdf", payload)
|
|
(self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink()
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen") as popen_mock,
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(job["state"], "failed")
|
|
self.assertIn("OpenCode readiness failed", job["error"])
|
|
popen_mock.assert_not_called()
|
|
|
|
def test_upload_chunk_lifecycle_verifies_offset_size_and_sha(self):
|
|
payload = b"hello ingest"
|
|
upload = self.manager.create_upload(
|
|
file_name="drawing.pdf",
|
|
size_bytes=len(payload),
|
|
sha256="sha256:" + self._sha256(payload),
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
)
|
|
|
|
first = self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=0,
|
|
data_base64=base64.b64encode(payload[:5]).decode("ascii"),
|
|
)
|
|
self.assertEqual(first["received_bytes"], 5)
|
|
|
|
with self.assertRaises(IngestError):
|
|
self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=3,
|
|
data_base64=base64.b64encode(payload[5:]).decode("ascii"),
|
|
)
|
|
|
|
second = self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=5,
|
|
data_base64=base64.b64encode(payload[5:]).decode("ascii"),
|
|
)
|
|
self.assertEqual(second["received_bytes"], len(payload))
|
|
|
|
complete = self.manager.complete_upload(upload["upload_id"])
|
|
|
|
self.assertEqual(complete["state"], "complete")
|
|
self.assertNotIn("path", complete)
|
|
self.assertEqual(self._upload_payload_path(complete["upload_id"]).read_bytes(), payload)
|
|
|
|
def test_create_completed_upload_accepts_bytes_without_agent_chunking(self):
|
|
payload = b"direct upload payload"
|
|
|
|
upload = self.manager.create_completed_upload(
|
|
file_name="drawing.pdf",
|
|
payload=payload,
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
)
|
|
|
|
self.assertEqual(upload["state"], "complete")
|
|
self.assertEqual(upload["received_bytes"], len(payload))
|
|
self.assertEqual(upload["size_bytes"], len(payload))
|
|
self.assertEqual(upload["sha256"], "sha256:" + self._sha256(payload))
|
|
self.assertEqual(upload["project"], "bjd")
|
|
self.assertEqual(upload["doc_type_hint"], "drawing")
|
|
self.assertNotIn("path", upload)
|
|
self.assertEqual(self._upload_payload_path(upload["upload_id"]).read_bytes(), payload)
|
|
|
|
def test_create_completed_upload_from_stream_reads_bounded_chunks(self):
|
|
payload = b"streamed upload payload" * 100_000
|
|
|
|
class BoundedReadStream(io.BytesIO):
|
|
def __init__(self, data):
|
|
super().__init__(data)
|
|
self.read_sizes = []
|
|
|
|
def read(self, size=-1):
|
|
if size < 0:
|
|
raise AssertionError("stream upload must not use an unbounded read")
|
|
self.read_sizes.append(size)
|
|
return super().read(size)
|
|
|
|
manager = IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.root / "stream-runtime",
|
|
max_file_bytes=len(payload) + 1,
|
|
)
|
|
stream = BoundedReadStream(payload)
|
|
|
|
upload = manager.create_completed_upload_from_stream(
|
|
file_name="stream.bin",
|
|
stream=stream,
|
|
size_bytes=len(payload),
|
|
)
|
|
|
|
self.assertEqual(upload["state"], "complete")
|
|
self.assertEqual(upload["sha256"], "sha256:" + self._sha256(payload))
|
|
self.assertGreater(len(stream.read_sizes), 2)
|
|
self.assertTrue(all(size == 1024 * 1024 for size in stream.read_sizes))
|
|
|
|
def test_create_completed_upload_from_stream_cleans_partial_file_when_oversized(self):
|
|
manager = IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.root / "small-stream-runtime",
|
|
max_file_bytes=5,
|
|
)
|
|
|
|
with self.assertRaises(IngestError):
|
|
manager.create_completed_upload_from_stream(
|
|
file_name="too-large.bin",
|
|
stream=io.BytesIO(b"123456"),
|
|
)
|
|
|
|
self.assertEqual(list(manager.upload_root.rglob("upload.json")), [])
|
|
self.assertEqual([path for path in manager.upload_root.rglob("*") if path.is_file()], [])
|
|
|
|
def test_upload_ticket_is_single_use_and_preserves_defaults(self):
|
|
ticket = self.manager.create_upload_ticket(
|
|
expires_seconds=60,
|
|
max_files=1,
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
note="customer uploaded drawing",
|
|
bundle_id="drawing-package-1",
|
|
source_type="report_package",
|
|
label="Customer drawing package",
|
|
)
|
|
|
|
self.assertIn("upload_token", ticket)
|
|
self.assertEqual(ticket["max_files"], 1)
|
|
self.assertEqual(ticket["project"], "bjd")
|
|
self.assertEqual(ticket["doc_type_hint"], "drawing")
|
|
self.assertEqual(ticket["note"], "customer uploaded drawing")
|
|
self.assertEqual(ticket["bundle_id"], "drawing-package-1")
|
|
self.assertEqual(ticket["source_type"], "report_package")
|
|
self.assertEqual(ticket["label"], "Customer drawing package")
|
|
self.assertFalse(ticket["start_job"])
|
|
self.assertEqual(ticket["used_at"], None)
|
|
|
|
used = self.manager.consume_upload_ticket(ticket["upload_token"])
|
|
|
|
self.assertEqual(used["project"], "bjd")
|
|
self.assertEqual(used["bundle_id"], "drawing-package-1")
|
|
self.assertIsNotNone(used["used_at"])
|
|
with self.assertRaises(IngestError):
|
|
self.manager.consume_upload_ticket(ticket["upload_token"])
|
|
|
|
def test_upload_rejects_unsafe_names_oversized_files_and_large_chunks(self):
|
|
with self.assertRaises(IngestError):
|
|
self.manager.create_upload(file_name="../secret.pdf", size_bytes=10, sha256="a" * 64)
|
|
|
|
with self.assertRaises(IngestError):
|
|
self.manager.create_upload(file_name="large.pdf", size_bytes=101, sha256="a" * 64)
|
|
|
|
upload = self.manager.create_upload(file_name="ok.pdf", size_bytes=20, sha256=self._sha256(b"x" * 20))
|
|
with self.assertRaises(IngestError):
|
|
self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=0,
|
|
data_base64=base64.b64encode(b"x" * 9).decode("ascii"),
|
|
)
|
|
|
|
with self.assertRaises(IngestError):
|
|
self.manager.create_completed_upload(file_name="too-large.pdf", payload=b"x" * 101)
|
|
|
|
def test_validate_upload_plan_reports_limits_and_exact_duplicates(self):
|
|
duplicate_digest = self._sha256(b"already archived")
|
|
registry = self.vault / "_meta" / "source-registry.jsonl"
|
|
registry.parent.mkdir(parents=True)
|
|
registry.write_text(
|
|
json.dumps(
|
|
{
|
|
"record_type": "source_archive",
|
|
"source_id": "sha256:" + duplicate_digest,
|
|
"archive_path": "_sources/archive/existing.pdf",
|
|
"original_name": "existing.pdf",
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.manager.validate_upload_plan(
|
|
[
|
|
{"file_name": "existing.pdf", "size_bytes": 12, "sha256": duplicate_digest},
|
|
{"file_name": "too-large.pdf", "size_bytes": 101, "sha256": "b" * 64},
|
|
{"file_name": "../secret.pdf", "size_bytes": 10, "sha256": "c" * 64},
|
|
{"file_name": "bad-sha.pdf", "size_bytes": 10, "sha256": "not-a-sha"},
|
|
],
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
)
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual(result["project"], "bjd")
|
|
self.assertEqual(result["doc_type_hint"], "drawing")
|
|
self.assertEqual(result["max_file_bytes"], 100)
|
|
self.assertEqual(result["max_files_per_job"], 10)
|
|
self.assertEqual(result["files"][0]["status"], "warning")
|
|
self.assertTrue(result["files"][0]["duplicates"])
|
|
self.assertEqual(result["files"][1]["status"], "error")
|
|
self.assertIn("exceeds max upload size", result["files"][1]["errors"][0])
|
|
self.assertIn("unsafe file name", result["files"][2]["errors"][0])
|
|
self.assertIn("sha256", result["files"][3]["errors"][0])
|
|
|
|
def test_preview_upload_metadata_reports_duplicate_and_requires_completed_uploads(self):
|
|
payload = b"duplicate payload"
|
|
digest = self._sha256(payload)
|
|
registry = self.vault / "_meta" / "source-registry.jsonl"
|
|
registry.parent.mkdir(parents=True)
|
|
registry.write_text(
|
|
json.dumps(
|
|
{
|
|
"record_type": "source_archive",
|
|
"source_id": "sha256:" + digest,
|
|
"archive_path": "_sources/archive/duplicate.pdf",
|
|
"original_name": "duplicate.pdf",
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
complete = self.manager.create_completed_upload(
|
|
"duplicate.pdf",
|
|
payload,
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
)
|
|
incomplete = self.manager.create_upload("pending.pdf", 7, self._sha256(b"pending"))
|
|
|
|
preview = self.manager.preview_upload_metadata([complete["upload_id"]])
|
|
|
|
self.assertTrue(preview["ok"])
|
|
self.assertEqual(preview["next_action"], "start_ingest_job")
|
|
self.assertEqual(preview["uploads"][0]["file_extension"], ".pdf")
|
|
self.assertTrue(preview["uploads"][0]["duplicates"])
|
|
self.assertEqual(preview["uploads"][0]["duplicate_count"], 1)
|
|
with self.assertRaises(IngestError):
|
|
self.manager.preview_upload_metadata([incomplete["upload_id"]])
|
|
|
|
def test_start_ingest_job_constructs_restricted_opencode_command(self):
|
|
payload = b"source data"
|
|
upload = self._complete_upload("source.pdf", payload)
|
|
self.manager.create_ingest_bundle(
|
|
[upload["upload_id"]],
|
|
source_type="report_package",
|
|
label="Quality report",
|
|
)
|
|
self.manager.opencode_model = "openai/gpt-5.5"
|
|
self.manager.opencode_variant = "high"
|
|
popen = FakePopen()
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen) as popen_mock,
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]], note="customer asked for drawings")
|
|
|
|
args = popen_mock.call_args.args[0]
|
|
self.assertEqual(args[:5], ["/usr/bin/opencode", "run", "--format", "json", "--auto"])
|
|
self.assertEqual(args[args.index("--model") + 1], "openai/gpt-5.5")
|
|
self.assertEqual(args[args.index("--variant") + 1], "high")
|
|
self.assertIn("--dir", args)
|
|
self.assertEqual(args[args.index("--dir") + 1], str(self.project))
|
|
self.assertIn("--title", args)
|
|
self.assertIn(f"MCP ingest {job['job_id']}", args)
|
|
prompt = args[-1]
|
|
self.assertIn("/wiki-ingest", prompt)
|
|
self.assertIn('"source_type": "report_package"', prompt)
|
|
self.assertIn('"label": "Quality report"', prompt)
|
|
self.assertIn('"bundle_role": "attachment"', prompt)
|
|
self.assertIn(str(self._upload_payload_path(upload["upload_id"])), prompt)
|
|
self.assertIn("temporary runtime staging files", prompt)
|
|
self.assertIn("Do not ingest any file outside this list.", prompt)
|
|
self.assertIn("obsidian-wiki file-index rebuild", prompt)
|
|
self.assertNotIn("../secret", prompt)
|
|
self.assertNotIn("path", job["files"][0])
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(self._json_text_event("MCP_INGEST_OK\nIngested source.pdf"), encoding="utf-8")
|
|
popen.returncode = 0
|
|
with patch(
|
|
"obsidian_wiki.mcp_ingest.verify_ingest_outputs",
|
|
return_value=self._passed_verification(1),
|
|
):
|
|
self.manager.refresh_jobs()
|
|
status = self.manager.get_job_status(job["job_id"])
|
|
self.assertEqual(status["state"], "succeeded")
|
|
self.assertEqual(status["exit_code"], 0)
|
|
self.assertNotIn("path", status["files"][0])
|
|
|
|
def test_job_reports_authoritative_preflight_and_agent_running_phases(self):
|
|
upload = self._complete_upload("source.pdf", b"source data")
|
|
popen = FakePopen(returncode=None)
|
|
observed = {}
|
|
|
|
def readiness_during_launch(*, run_smoke_test=False):
|
|
internal = next(iter(self.manager._iter_jobs()))
|
|
observed["state"] = internal["state"]
|
|
observed["phase"] = internal["phase"]
|
|
observed["verification"] = internal["verification"]
|
|
lock_probe = []
|
|
|
|
def acquire_manager_lock():
|
|
acquired = self.manager._lock.acquire(timeout=0.2)
|
|
lock_probe.append(acquired)
|
|
if acquired:
|
|
self.manager._lock.release()
|
|
|
|
thread = threading.Thread(target=acquire_manager_lock)
|
|
thread.start()
|
|
thread.join()
|
|
observed["status_lock_available"] = lock_probe == [True]
|
|
return {
|
|
"ok": True,
|
|
"errors": [],
|
|
"warnings": [],
|
|
"static_preflight": True,
|
|
"runtime_smoke_test": "skipped",
|
|
}
|
|
|
|
with (
|
|
patch.object(self.manager, "check_opencode_readiness", side_effect=readiness_during_launch),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
status = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(observed["state"], "running")
|
|
self.assertEqual(observed["phase"], "preflight")
|
|
self.assertEqual(observed["verification"]["status"], "not_run")
|
|
self.assertTrue(observed["status_lock_available"])
|
|
self.assertEqual(status["state"], "running")
|
|
self.assertEqual(status["phase"], "agent_running")
|
|
self.assertTrue(status["phase_started_at"])
|
|
self.assertTrue(status["process_alive"])
|
|
self.assertEqual(status["verification"]["status"], "not_run")
|
|
|
|
def test_queued_and_cancelled_job_phases_are_authoritative(self):
|
|
first = self._complete_upload("first.pdf", b"first")
|
|
second = self._complete_upload("second.pdf", b"second")
|
|
popen = FakePopen(returncode=None)
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
self.manager.start_ingest_job([first["upload_id"]])
|
|
queued = self.manager.start_ingest_job([second["upload_id"]])
|
|
|
|
self.assertEqual(queued["state"], "queued")
|
|
self.assertEqual(queued["phase"], "queued")
|
|
self.assertFalse(queued["process_alive"])
|
|
self.assertEqual(queued["verification"]["status"], "not_run")
|
|
|
|
cancelled = self.manager.cancel_ingest_job(queued["job_id"])
|
|
self.assertEqual(cancelled["state"], "cancelled")
|
|
self.assertEqual(cancelled["phase"], "complete")
|
|
self.assertFalse(cancelled["process_alive"])
|
|
self.assertEqual(cancelled["verification"]["status"], "not_run")
|
|
|
|
def test_preflight_failure_completes_without_verification(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
(self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink()
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen") as popen_mock,
|
|
):
|
|
status = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(status["state"], "failed")
|
|
self.assertEqual(status["phase"], "complete")
|
|
self.assertFalse(status["process_alive"])
|
|
self.assertEqual(status["verification"]["status"], "not_run")
|
|
popen_mock.assert_not_called()
|
|
|
|
def test_preflight_job_can_be_cancelled_while_readiness_runs(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
cancelled = {}
|
|
|
|
def cancel_during_readiness(*, run_smoke_test=False):
|
|
internal = next(iter(self.manager._iter_jobs()))
|
|
|
|
def cancel_job():
|
|
cancelled.update(self.manager.cancel_ingest_job(internal["job_id"]))
|
|
|
|
thread = threading.Thread(target=cancel_job)
|
|
thread.start()
|
|
thread.join()
|
|
return {
|
|
"ok": True,
|
|
"errors": [],
|
|
"warnings": [],
|
|
"static_preflight": True,
|
|
"runtime_smoke_test": "skipped",
|
|
}
|
|
|
|
with (
|
|
patch.object(self.manager, "check_opencode_readiness", side_effect=cancel_during_readiness),
|
|
patch(
|
|
"obsidian_wiki.mcp_ingest.subprocess.Popen",
|
|
return_value=FakePopen(returncode=None),
|
|
) as popen_mock,
|
|
):
|
|
status = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(cancelled["state"], "cancelled")
|
|
self.assertEqual(status["state"], "cancelled")
|
|
self.assertEqual(status["phase"], "complete")
|
|
self.assertEqual(status["verification"]["status"], "not_run")
|
|
popen_mock.assert_not_called()
|
|
|
|
def test_unexpected_preflight_error_finishes_job(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
with (
|
|
patch.object(
|
|
self.manager,
|
|
"check_opencode_readiness",
|
|
side_effect=RuntimeError("readiness crashed"),
|
|
),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen") as popen_mock,
|
|
):
|
|
status = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(status["state"], "failed")
|
|
self.assertEqual(status["phase"], "complete")
|
|
self.assertEqual(status["verification"]["status"], "not_run")
|
|
self.assertIn("readiness crashed", status["error"])
|
|
popen_mock.assert_not_called()
|
|
|
|
def test_legacy_job_records_derive_phase_and_verification_defaults(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
(self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink()
|
|
with patch(
|
|
"obsidian_wiki.mcp_ingest.subprocess.run",
|
|
side_effect=self._fake_static_readiness_run,
|
|
):
|
|
status = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
internal = self.manager._load_job(status["job_id"])
|
|
internal.pop("phase", None)
|
|
internal.pop("phase_started_at", None)
|
|
internal.pop("verification", None)
|
|
self.manager._write_job(internal)
|
|
|
|
legacy = self.manager.get_ingest_job_status(status["job_id"])
|
|
self.assertEqual(legacy["phase"], "complete")
|
|
self.assertEqual(legacy["verification"]["status"], "not_run")
|
|
self.assertFalse(legacy["process_alive"])
|
|
|
|
def test_service_restart_completes_interrupted_job_without_verification(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
(self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink()
|
|
with patch(
|
|
"obsidian_wiki.mcp_ingest.subprocess.run",
|
|
side_effect=self._fake_static_readiness_run,
|
|
):
|
|
status = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
internal = self.manager._load_job(status["job_id"])
|
|
internal["state"] = "running"
|
|
internal["phase"] = "verifying"
|
|
internal["verification"] = {"status": "running"}
|
|
internal["finished_at"] = None
|
|
self.manager._write_job(internal)
|
|
self.manager._mark_interrupted_jobs()
|
|
|
|
restarted = self.manager.get_ingest_job_status(status["job_id"])
|
|
self.assertEqual(restarted["state"], "failed")
|
|
self.assertEqual(restarted["phase"], "complete")
|
|
self.assertEqual(restarted["verification"]["status"], "not_run")
|
|
self.assertIn("service restart", restarted["error"])
|
|
|
|
def test_success_marker_runs_verification_before_succeeding(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
popen = FakePopen(returncode=None)
|
|
observed = {}
|
|
passed = {
|
|
"status": "passed",
|
|
"checked_at": "2026-07-16T10:01:00Z",
|
|
"failures": [],
|
|
"warnings": [],
|
|
"outputs": {
|
|
"source_count": 1,
|
|
"registry_count": 1,
|
|
"archive_count": 1,
|
|
"manifest_count": 1,
|
|
"indexed_count": 1,
|
|
},
|
|
}
|
|
|
|
def verify_while_observable(**kwargs):
|
|
internal = self.manager._load_job(job["job_id"])
|
|
public = self.manager._public_job(internal)
|
|
observed["phase"] = public["phase"]
|
|
observed["process_alive"] = public["process_alive"]
|
|
observed["verification"] = public["verification"]
|
|
observed["uploads"] = kwargs["uploads"]
|
|
return passed
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
self._json_text_event("MCP_INGEST_OK\nIngested source.pdf"),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
with patch(
|
|
"obsidian_wiki.mcp_ingest.verify_ingest_outputs",
|
|
side_effect=verify_while_observable,
|
|
):
|
|
self.manager.refresh_jobs()
|
|
|
|
status = self.manager.get_ingest_job_status(job["job_id"])
|
|
self.assertEqual(observed["phase"], "verifying")
|
|
self.assertFalse(observed["process_alive"])
|
|
self.assertEqual(observed["verification"]["status"], "running")
|
|
self.assertEqual(observed["uploads"][0]["path"], str(self._upload_payload_path(upload["upload_id"])))
|
|
self.assertEqual(status["state"], "succeeded")
|
|
self.assertEqual(status["phase"], "complete")
|
|
self.assertFalse(status["process_alive"])
|
|
self.assertEqual(status["verification"], passed)
|
|
|
|
def test_failed_verification_overrides_agent_success_and_preserves_output(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
popen = FakePopen(returncode=None)
|
|
failed = {
|
|
"status": "failed",
|
|
"checked_at": "2026-07-16T10:01:00Z",
|
|
"failures": [{"code": "manifest_missing", "upload_id": upload["upload_id"]}],
|
|
"warnings": [],
|
|
"outputs": {
|
|
"source_count": 1,
|
|
"registry_count": 1,
|
|
"archive_count": 1,
|
|
"manifest_count": 0,
|
|
"indexed_count": 1,
|
|
},
|
|
}
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
self._json_text_event("MCP_INGEST_OK\nCreated references/source.md"),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
with patch(
|
|
"obsidian_wiki.mcp_ingest.verify_ingest_outputs",
|
|
return_value=failed,
|
|
):
|
|
self.manager.refresh_jobs()
|
|
|
|
result = self.manager.get_ingest_job_result(job["job_id"])
|
|
self.assertEqual(result["state"], "failed")
|
|
self.assertEqual(result["phase"], "complete")
|
|
self.assertEqual(result["verification"], failed)
|
|
self.assertIn("verification failed", result["error"])
|
|
self.assertIn("Created references/source.md", result["opencode_output_tail"])
|
|
|
|
def test_protocol_failure_skips_verification(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
popen = FakePopen(returncode=None)
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
self._json_text_event("MCP_INGEST_FAILED\nagent failed"),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
with patch("obsidian_wiki.mcp_ingest.verify_ingest_outputs") as verify_mock:
|
|
self.manager.refresh_jobs()
|
|
|
|
status = self.manager.get_ingest_job_status(job["job_id"])
|
|
self.assertEqual(status["state"], "failed")
|
|
self.assertEqual(status["phase"], "complete")
|
|
self.assertEqual(status["verification"]["status"], "not_run")
|
|
verify_mock.assert_not_called()
|
|
|
|
def test_ingest_result_rejects_skill_failure_with_zero_exit_code(self):
|
|
upload = self._complete_upload("source.pdf", b"source data")
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(self._json_text_event("MCP_INGEST_FAILED\nSKILL_LOAD_FAILED: missing"), encoding="utf-8")
|
|
popen.returncode = 0
|
|
result = self.manager.get_ingest_job_result(job["job_id"])
|
|
|
|
self.assertEqual(result["state"], "failed")
|
|
self.assertIn("SKILL_LOAD_FAILED", result["error"])
|
|
|
|
def test_ingest_public_status_hides_raw_tool_logs_paths_and_pid(self):
|
|
upload = self._complete_upload("source.pdf", b"source data")
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text(
|
|
'{"type":"tool_use","part":{"state":{"output":"/home/private/secret"}}}\n'
|
|
+ self._json_text_event("MCP_INGEST_OK\nIngest complete"),
|
|
encoding="utf-8",
|
|
)
|
|
popen.returncode = 0
|
|
with patch(
|
|
"obsidian_wiki.mcp_ingest.verify_ingest_outputs",
|
|
return_value=self._passed_verification(1),
|
|
):
|
|
result = self.manager.get_ingest_job_result(job["job_id"])
|
|
status = self.manager.get_ingest_job_status(job["job_id"])
|
|
|
|
self.assertEqual(result["state"], "succeeded")
|
|
self.assertEqual(result["opencode_output_tail"], "Ingest complete")
|
|
self.assertNotIn("/home/private", result["opencode_output_tail"])
|
|
self.assertNotIn("pid", status)
|
|
self.assertNotIn("checked_paths", status["preflight"])
|
|
|
|
internal = self.manager._load_job(job["job_id"])
|
|
internal["error"] = f"failed while reading {self.project / 'secret.txt'}"
|
|
internal["preflight"]["warnings"] = [f"check {self.runtime / 'private.log'}"]
|
|
self.manager._write_job(internal)
|
|
redacted = self.manager.get_ingest_job_status(job["job_id"])
|
|
serialized = json.dumps(redacted)
|
|
self.assertNotIn(str(self.project), serialized)
|
|
self.assertNotIn(str(self.runtime), serialized)
|
|
self.assertIn("<project>", serialized)
|
|
self.assertIn("<runtime>", serialized)
|
|
|
|
def test_jobs_queue_one_at_a_time_and_cancel_queued_job(self):
|
|
first = self._complete_upload("first.pdf", b"first")
|
|
second = self._complete_upload("second.pdf", b"second")
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
first_job = self.manager.start_ingest_job([first["upload_id"]])
|
|
second_job = self.manager.start_ingest_job([second["upload_id"]])
|
|
|
|
self.assertEqual(self.manager.get_job_status(first_job["job_id"])["state"], "running")
|
|
self.assertEqual(self.manager.get_job_status(second_job["job_id"])["state"], "queued")
|
|
|
|
cancelled = self.manager.cancel_ingest_job(second_job["job_id"])
|
|
|
|
self.assertEqual(cancelled["state"], "cancelled")
|
|
|
|
def test_start_ingest_job_is_idempotent_for_claimed_uploads(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen) as popen_mock,
|
|
):
|
|
first = self.manager.start_ingest_job([upload["upload_id"]])
|
|
repeated = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(repeated["job_id"], first["job_id"])
|
|
self.assertEqual(repeated["state"], "running")
|
|
popen_mock.assert_called_once()
|
|
|
|
def test_create_bundle_persists_body_and_attachment_relationship(self):
|
|
body = self._complete_upload("thread.md", b"# Thread\n\nBody")
|
|
attachment = self._complete_upload("report.pdf", b"report")
|
|
|
|
bundle = self.manager.create_ingest_bundle(
|
|
[body["upload_id"], attachment["upload_id"]],
|
|
body_upload_id=body["upload_id"],
|
|
bundle_id="mail-thread-123",
|
|
source_type="email_thread",
|
|
label="Customer quality thread",
|
|
)
|
|
|
|
self.assertEqual(bundle["bundle_id"], "mail-thread-123")
|
|
self.assertEqual(bundle["source_type"], "email_thread")
|
|
self.assertEqual(bundle["label"], "Customer quality thread")
|
|
self.assertEqual(bundle["body_upload_id"], body["upload_id"])
|
|
self.assertEqual(bundle["attachment_upload_ids"], [attachment["upload_id"]])
|
|
self.assertEqual(
|
|
self.manager._load_upload(body["upload_id"])["bundle_role"],
|
|
"primary",
|
|
)
|
|
self.assertEqual(
|
|
self.manager._load_upload(attachment["upload_id"])["bundle_role"],
|
|
"attachment",
|
|
)
|
|
|
|
preview = self.manager.preview_upload_metadata(
|
|
[body["upload_id"], attachment["upload_id"]]
|
|
)
|
|
self.assertEqual(preview["bundle"], bundle)
|
|
|
|
def test_start_ingest_job_rejects_partial_bundle(self):
|
|
body = self._complete_upload("thread.md", b"body")
|
|
attachment = self._complete_upload("report.pdf", b"report")
|
|
self.manager.create_ingest_bundle(
|
|
[body["upload_id"], attachment["upload_id"]],
|
|
body_upload_id=body["upload_id"],
|
|
)
|
|
|
|
with self.assertRaisesRegex(IngestError, "complete bundle"):
|
|
self.manager.start_ingest_job([body["upload_id"]])
|
|
|
|
def test_start_ingest_job_rejects_duplicate_upload_ids(self):
|
|
upload = self._complete_upload("source.pdf", b"source")
|
|
|
|
with self.assertRaisesRegex(IngestError, "upload_ids must be unique"):
|
|
self.manager.start_ingest_job([upload["upload_id"], upload["upload_id"]])
|
|
|
|
def test_cleanup_runtime_prunes_expired_tickets_and_old_completed_artifacts(self):
|
|
upload = self._complete_upload("old.pdf", b"old payload")
|
|
ticket = self.manager.create_upload_ticket(expires_seconds=60, max_files=1)
|
|
ticket_path = next(self.manager.upload_ticket_root.glob("*.json"))
|
|
ticket_meta = json.loads(ticket_path.read_text(encoding="utf-8"))
|
|
ticket_meta["expires_epoch"] = time.time() - 10
|
|
ticket_path.write_text(json.dumps(ticket_meta), encoding="utf-8")
|
|
|
|
popen = FakePopen(returncode=0)
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
self.manager.refresh_jobs()
|
|
|
|
old_epoch = time.time() - 40 * 24 * 60 * 60
|
|
upload_path = self._upload_payload_path(upload["upload_id"])
|
|
upload_dir = upload_path.parent
|
|
job_dir = self.manager.jobs_root / job["job_id"]
|
|
bundle_dir = self.manager.bundles_root / job["bundle"]["bundle_id"]
|
|
self._touch_tree(upload_dir, old_epoch)
|
|
self._touch_tree(job_dir, old_epoch)
|
|
self._touch_tree(bundle_dir, old_epoch)
|
|
os.utime(ticket_path, (old_epoch, old_epoch))
|
|
|
|
result = self.manager.cleanup_runtime(max_age_days=30)
|
|
|
|
self.assertGreaterEqual(result["expired_tickets_deleted"], 1)
|
|
self.assertEqual(result["old_uploads_deleted"], 1)
|
|
self.assertEqual(result["old_bundles_deleted"], 1)
|
|
self.assertEqual(result["old_jobs_deleted"], 1)
|
|
self.assertFalse(upload_dir.exists())
|
|
self.assertFalse(bundle_dir.exists())
|
|
self.assertFalse(job_dir.exists())
|
|
self.assertFalse(ticket_path.exists())
|
|
|
|
def _complete_upload(self, name: str, payload: bytes) -> dict:
|
|
upload = self.manager.create_upload(name, len(payload), self._sha256(payload))
|
|
offset = 0
|
|
while offset < len(payload):
|
|
chunk = payload[offset : offset + self.manager.max_chunk_bytes]
|
|
self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=offset,
|
|
data_base64=base64.b64encode(chunk).decode("ascii"),
|
|
)
|
|
offset += len(chunk)
|
|
return self.manager.complete_upload(upload["upload_id"])
|
|
|
|
def _upload_payload_path(self, upload_id: str) -> Path:
|
|
return Path(self.manager._load_upload(upload_id)["path"])
|
|
|
|
def _touch_tree(self, path: Path, epoch: float) -> None:
|
|
for child in path.rglob("*"):
|
|
os.utime(child, (epoch, epoch))
|
|
os.utime(path, (epoch, epoch))
|
|
|
|
def _sha256(self, payload: bytes) -> str:
|
|
import hashlib
|
|
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
def _write_opencode_skill_fixture(self, root: Path) -> None:
|
|
(root / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
wiki_ingest = root / ".agents" / "skills" / "wiki-ingest"
|
|
llm_wiki = root / ".agents" / "skills" / "llm-wiki"
|
|
wiki_ingest_refs = wiki_ingest / "references"
|
|
wiki_ingest_refs.mkdir(parents=True)
|
|
llm_wiki.mkdir(parents=True)
|
|
(wiki_ingest / "SKILL.md").write_text(
|
|
"---\nname: wiki-ingest\ndescription: Ingest documents into the wiki.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
(llm_wiki / "SKILL.md").write_text(
|
|
"---\nname: llm-wiki\ndescription: Shared wiki framework instructions.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
(wiki_ingest_refs / "ingest-prompts.md").write_text("# Prompt templates\n", encoding="utf-8")
|
|
|
|
def _fake_static_readiness_run(self, args, **_kwargs):
|
|
if args == ["/usr/bin/opencode", "--version"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout="1.17.13\n", stderr="")
|
|
if args == ["/usr/bin/opencode", "run", "--help"]:
|
|
return subprocess.CompletedProcess(
|
|
args,
|
|
0,
|
|
stdout="--format\n--auto\n--dir\n--title\n",
|
|
stderr="",
|
|
)
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_success_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout='{"message":"WIKI_INGEST_SKILL_LOADED"}\n', stderr="")
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_failure_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout="SKILL_LOAD_FAILED: missing\n", stderr="")
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_timeout_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
raise subprocess.TimeoutExpired(args, kwargs.get("timeout"))
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _passed_verification(self, source_count):
|
|
return {
|
|
"status": "passed",
|
|
"checked_at": "2026-07-16T10:01:00Z",
|
|
"failures": [],
|
|
"warnings": [],
|
|
"outputs": {
|
|
"source_count": source_count,
|
|
"registry_count": source_count,
|
|
"archive_count": source_count,
|
|
"manifest_count": source_count,
|
|
"indexed_count": source_count,
|
|
},
|
|
}
|
|
|
|
def _json_text_event(self, text):
|
|
return json.dumps(
|
|
{
|
|
"type": "text",
|
|
"part": {
|
|
"type": "text",
|
|
"text": text,
|
|
"metadata": {"openai": {"phase": "final_answer"}},
|
|
},
|
|
}
|
|
) + "\n"
|
|
|
|
|
|
class FakePopen:
|
|
def __init__(self, returncode=None):
|
|
self.returncode = returncode
|
|
self.pid = 12345
|
|
self.terminated = False
|
|
|
|
def poll(self):
|
|
return self.returncode
|
|
|
|
def wait(self, timeout=None):
|
|
deadline = time.monotonic() + (timeout or 0)
|
|
while self.returncode is None and (timeout is None or time.monotonic() < deadline):
|
|
time.sleep(0.01)
|
|
if self.returncode is None:
|
|
raise subprocess.TimeoutExpired("fake", timeout)
|
|
return self.returncode
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
self.returncode = -15
|
|
|
|
def kill(self):
|
|
self.returncode = -9
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|