import asyncio import hashlib import os import sqlite3 import tempfile import unittest from pathlib import Path from unittest.mock import patch from obsidian_wiki import business_facts as facts, mcp_server from obsidian_wiki.mcp_business import BUSINESS_TOOLS, BusinessQueries class BusinessMcpTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.db = self.root / "orders.sqlite" facts.sync_kingdee_orders( self.root, [ {"id": "external", "billno": "SO-1", "bizdate": "2026-06-03", "customer_name": "External customer", "org_name": "Factory", "totalallamount": 100, "operator_operatorname": "Alice"}, {"id": "internal", "billno": "SO-2", "bizdate": "2026-06-04", "customer_name": "Sales company", "org_name": "Factory", "totalallamount": 500}, ], db_path=self.db, synced_at="2026-06-05T00:00:00Z", ) self.queries = BusinessQueries(self.db, defaults={"internal_parties": "Factory,Sales company"}) def test_templates_preserve_intercompany_basis_and_do_not_modify_database(self): # Checkpoint the sync transaction so the file digest covers the current data. with sqlite3.connect(self.db) as conn: conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") before = hashlib.sha256(self.db.read_bytes()).hexdigest() result = self.queries.template("sales_amount", {"year": 2026, "month": 6}) self.assertEqual(result["data"]["sales_amount"], 100) self.assertEqual(result["last_sync"], "2026-06-05T00:00:00Z") all_orders = self.queries.template("sales_amount", { "year": 2026, "month": 6, "include_internal_orders": True, }) self.assertEqual(all_orders["data"]["sales_amount"], 600) self.assertEqual(hashlib.sha256(self.db.read_bytes()).hexdigest(), before) with facts.read_only_queries(): with facts.connect(self.db) as conn: with self.assertRaises(sqlite3.OperationalError): conn.execute("DELETE FROM orders") def test_every_existing_template_runs_through_read_only_adapter(self): for name in facts.BUSINESS_TEMPLATE_NAMES: with self.subTest(template=name): result = self.queries.template(name, { "year": 2026, "month": 6, "customer": "External", "product": "P1", "staff": "Alice", }) self.assertEqual(result["template"], name) self.assertIn("last_sync", result) def test_sql_keeps_guards_row_limit_and_freshness(self): result = self.queries.sql("SELECT order_no FROM orders ORDER BY order_no", limit=1) self.assertEqual(result["rows"], [{"order_no": "SO-1"}]) self.assertTrue(result["is_stale"]) self.assertTrue(result["warnings"]) for sql in ["DELETE FROM orders", "SELECT raw_json FROM orders", "SELECT * FROM orders", "SELECT * FROM raw_snapshots", "PRAGMA database_list", "SELECT order_no FROM orders; SELECT 1"]: with self.subTest(sql=sql), self.assertRaises(RuntimeError): self.queries.sql(sql) def test_missing_database_is_not_created(self): db = self.root / "missing" / "orders.sqlite" queries = BusinessQueries(db) self.assertFalse(queries.status()["available"]) with self.assertRaises(RuntimeError): queries.template("order_count", {"year": 2026, "month": 6}) self.assertFalse(db.parent.exists()) def test_database_path_and_server_filters_cannot_be_overridden_by_client(self): for params in [{"db_path": "/tmp/other"}, {"internal_parties": ""}, {"internal_party_ids": []}]: with self.subTest(params=params), self.assertRaises(ValueError): self.queries.template("order_count", params) def test_read_only_context_does_not_change_normal_sync(self): self.queries.status() facts.sync_kingdee_orders(self.root, [ {"id": "new", "billno": "SO-3", "bizdate": "2026-06-06", "totalallamount": 20}, ], db_path=self.db) self.assertEqual(self.queries.status()["stored_orders"], 3) def test_business_only_mcp_has_no_wiki_tools_resources_or_job_managers(self): with ( patch.object(mcp_server, "IngestManager", side_effect=AssertionError("no ingest")), patch.object(mcp_server, "QueryManager", side_effect=AssertionError("no OpenCode")), patch.object(mcp_server, "_file_index_capabilities", side_effect=AssertionError("no file index")), ): server = mcp_server.create_server( vault_path=self.root / "nonexistent-vault", enable_file_assets=False, enable_business=True, business_db_path=self.db, ) names = {tool.name for tool in asyncio.run(server.list_tools())} self.assertEqual(names, {*BUSINESS_TOOLS, "get_mcp_capabilities"}) self.assertEqual(asyncio.run(server.list_resources()), []) self.assertEqual(asyncio.run(server.list_resource_templates()), []) self.assertEqual(server._obsidian_job_managers, ()) async def check(): result = await server.call_tool("get_mcp_capabilities", {}) # FastMCP returns content blocks plus structured content for dict outputs. capabilities = result[1] self.assertEqual(capabilities["enabled_tools"], ["business"]) self.assertFalse(capabilities["business"]["requires_opencode"]) self.assertFalse(capabilities["query"]["enabled"]) self.assertFalse(capabilities["ingest"]["enabled"]) self.assertFalse(capabilities["file_assets"]["enabled"]) self.assertFalse(capabilities["safety"]["arbitrary_shell"]) response = await server.call_tool("query_business_template", { "template_name": "order_count", "params": {"year": 2026, "month": 6}, }) self.assertIn("order_count", response[1]["data"]) asyncio.run(check()) self.assertFalse((self.root / "nonexistent-vault").exists()) def test_business_tools_are_opt_in(self): server = mcp_server.create_server(vault_path=self.root, enable_file_assets=False) self.assertEqual({t.name for t in asyncio.run(server.list_tools())}, {"get_mcp_capabilities"}) with patch.dict(os.environ, {"OBSIDIAN_MCP_ENABLED_TOOLS": "business"}): self.assertEqual(mcp_server.enabled_tool_names(), {"business"}) def test_http_entrypoint_passes_business_group_and_separate_database(self): with ( patch.dict(os.environ, {"OBSIDIAN_MCP_ENABLED_TOOLS": "business"}), patch.object(mcp_server, "run_http_server") as run, ): self.assertEqual(mcp_server.main([ "--vault", str(self.root), "--business-db", str(self.db), "--transport", "streamable-http", ]), 0) self.assertTrue(run.call_args.kwargs["enable_business"]) self.assertEqual(run.call_args.kwargs["business_db_path"], str(self.db)) for name in ("enable_file_assets", "enable_query", "enable_ingest"): self.assertFalse(run.call_args.kwargs[name])