Codex
1150107aa3
Harden the HTTP ingest upload path by validating host/origin and method before ticket consumption, enforcing request body limits, and returning 413 for oversized uploads. Hide local runtime/log paths from default remote responses while preserving internal paths for local OpenCode prompts. Add conservative runtime cleanup for tickets, uploads, jobs, and logs, plus MemoryMax on the managed systemd unit. Validation: python3 -m unittest tests/test_mcp_server_remote.py tests/test_mcp_ingest.py tests/test_mcp_query.py tests/test_mcp_service.py -v; python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests scripts; git diff --check. Restarted obsidian-wiki-mcp.service and confirmed ActiveState=active, SubState=running, MemoryMax=2147483648, Restart=on-failure.
256 lines
11 KiB
Python
256 lines
11 KiB
Python
import subprocess
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from obsidian_wiki.mcp_query import QueryError, QueryManager
|
|
|
|
|
|
class McpQueryTests(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-query"
|
|
self.vault.mkdir()
|
|
self.project.mkdir()
|
|
self._write_opencode_skill_fixture(self.project)
|
|
self.manager = QueryManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.runtime,
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
max_query_chars=80,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.tempdir.cleanup()
|
|
|
|
def test_check_opencode_readiness_static_success(self):
|
|
with patch("obsidian_wiki.mcp_query.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-query/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 = QueryManager(
|
|
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_query.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-query/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 = "minimal"
|
|
with patch("obsidian_wiki.mcp_query.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", "default"])
|
|
self.assertEqual(smoke_args[smoke_args.index("--model") + 1], "openai/gpt-5.5")
|
|
self.assertEqual(smoke_args[smoke_args.index("--variant") + 1], "minimal")
|
|
self.assertIn("--auto", smoke_args)
|
|
self.assertIn("--dir", smoke_args)
|
|
self.assertIn("MCP query readiness smoke test", smoke_args)
|
|
|
|
def test_query_rejects_empty_oversized_and_invalid_mode(self):
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("")
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("x" * 81)
|
|
with self.assertRaises(QueryError):
|
|
self.manager.start_query_job("what do I know?", mode="full")
|
|
|
|
def test_start_query_job_constructs_restricted_opencode_command(self):
|
|
self.manager.opencode_model = "openai/gpt-5.5"
|
|
self.manager.opencode_variant = "minimal"
|
|
popen = FakePopen()
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen) as popen_mock,
|
|
):
|
|
job = self.manager.start_query_job(
|
|
"what do I know about Mini FAKRA?",
|
|
mode="index_only",
|
|
public_only=True,
|
|
note="remote agent asked",
|
|
)
|
|
|
|
args = popen_mock.call_args.args[0]
|
|
self.assertEqual(args[:5], ["/usr/bin/opencode", "run", "--format", "default", "--auto"])
|
|
self.assertEqual(args[args.index("--model") + 1], "openai/gpt-5.5")
|
|
self.assertEqual(args[args.index("--variant") + 1], "minimal")
|
|
self.assertIn("--dir", args)
|
|
self.assertEqual(args[args.index("--dir") + 1], str(self.project))
|
|
self.assertIn("--title", args)
|
|
self.assertIn(f"MCP query {job['job_id']}", args)
|
|
prompt = args[-1]
|
|
self.assertIn("/wiki-query", prompt)
|
|
self.assertIn("what do I know about Mini FAKRA?", prompt)
|
|
self.assertIn("index-only", prompt)
|
|
self.assertIn("public only", prompt)
|
|
self.assertIn("normal wiki-query log entry is allowed", prompt)
|
|
self.assertIn("wiki-ingest, wiki-update, wiki-rebuild", prompt)
|
|
self.assertIn("obsidian-wiki business schema --json", prompt)
|
|
self.assertIn("obsidian-wiki business template", prompt)
|
|
self.assertIn("obsidian-wiki business sql", prompt)
|
|
self.assertIn("read-only business facts", prompt)
|
|
self.assertIn("hybrid business + wiki questions", prompt)
|
|
self.assertIn("two read-only tracks", prompt)
|
|
self.assertIn("parallel subagents when the host supports them", prompt)
|
|
self.assertIn("run those tracks sequentially", prompt)
|
|
self.assertIn("customer, product, salesperson, project, delivery, anomaly, ranking, or quality", prompt)
|
|
self.assertNotIn("spawn read-only subagents", prompt)
|
|
self.assertIn("Do not run arbitrary shell commands", prompt)
|
|
self.assertIn("If Visibility is public only, do not use business facts", prompt)
|
|
self.assertNotIn("shell commands, file writes", prompt)
|
|
self.assertNotIn("rm -rf", prompt)
|
|
|
|
def test_query_result_returns_opencode_output_tail(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("what do I know?")
|
|
|
|
status = self.manager.get_query_job_status(job["job_id"])
|
|
self.assertNotIn("log_path", status)
|
|
log_path = Path(self.manager._load_job(job["job_id"])["log_path"])
|
|
log_path.write_text("Based on the wiki:\nAnswer body\n", encoding="utf-8")
|
|
popen.returncode = 0
|
|
result = self.manager.get_query_job_result(job["job_id"])
|
|
|
|
self.assertEqual(result["state"], "succeeded")
|
|
self.assertIn("Based on the wiki", result["answer"])
|
|
self.assertIn("Answer body", result["opencode_output_tail"])
|
|
|
|
def test_jobs_queue_one_at_a_time_and_cancel_queued_job(self):
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
first = self.manager.start_query_job("first query")
|
|
second = self.manager.start_query_job("second query")
|
|
|
|
self.assertEqual(self.manager.get_query_job_status(first["job_id"])["state"], "running")
|
|
self.assertEqual(self.manager.get_query_job_status(second["job_id"])["state"], "queued")
|
|
|
|
cancelled = self.manager.cancel_query_job(second["job_id"])
|
|
|
|
self.assertEqual(cancelled["state"], "cancelled")
|
|
|
|
def test_cleanup_runtime_prunes_old_terminal_query_jobs(self):
|
|
popen = FakePopen(returncode=0)
|
|
with (
|
|
patch("obsidian_wiki.mcp_query.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_query.subprocess.Popen", return_value=popen),
|
|
):
|
|
job = self.manager.start_query_job("old query")
|
|
self.manager.refresh_jobs()
|
|
job_dir = self.manager.jobs_root / job["job_id"]
|
|
old_epoch = time.time() - 40 * 24 * 60 * 60
|
|
self._touch_tree(job_dir, old_epoch)
|
|
|
|
result = self.manager.cleanup_runtime(max_age_days=30)
|
|
|
|
self.assertEqual(result["old_jobs_deleted"], 1)
|
|
self.assertFalse(job_dir.exists())
|
|
|
|
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 _write_opencode_skill_fixture(self, root: Path) -> None:
|
|
(root / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
wiki_query = root / ".agents" / "skills" / "wiki-query"
|
|
llm_wiki = root / ".agents" / "skills" / "llm-wiki"
|
|
wiki_query.mkdir(parents=True)
|
|
llm_wiki.mkdir(parents=True)
|
|
(wiki_query / "SKILL.md").write_text(
|
|
"---\nname: wiki-query\ndescription: Answer questions from 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",
|
|
)
|
|
|
|
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="WIKI_QUERY_SKILL_LOADED\n", stderr="")
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
|
|
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()
|