refactor(mcp): extract order services into independent project
Remove Kingdee sync, business CLI/MCP tools, and internal order-query routing now owned by the separate kingdee-orders-mcp project. Keep Wiki retrieval, file assets and ingest available in code while its deployed service is off. Retain legacy response fields with business access disabled, and document migration ownership and rollback without exposing credentials or data. Validation: 174 Wiki tests pass. Independent order service verified with 36 template parity checks, authenticated HTTP queries, and a real sync.
This commit is contained in:
Codex
File diff suppressed because it is too large
Load Diff
@@ -1,92 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import call, patch
|
||||
|
||||
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
|
||||
if str(SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
import manage_business_sync
|
||||
|
||||
|
||||
class BusinessSyncServiceTests(unittest.TestCase):
|
||||
def test_shell_entrypoint_delegates_to_unified_user_ops_entrypoint(self):
|
||||
entrypoint = SCRIPTS_DIR / "business_sync_service.sh"
|
||||
|
||||
self.assertTrue(entrypoint.is_file())
|
||||
self.assertTrue(os.access(entrypoint, os.X_OK))
|
||||
content = entrypoint.read_text(encoding="utf-8")
|
||||
self.assertIn('exec python3 "$SCRIPT_DIR/obsidian-wiki-ops.py" timer "$@"', content)
|
||||
|
||||
def test_render_service_and_timer_units_for_incremental_sync(self):
|
||||
service = manage_business_sync.render_service_unit(
|
||||
project_root=Path("/opt/obsidian-wiki"),
|
||||
vault_path=Path("/data/bjd-vault"),
|
||||
config_path=Path("/opt/obsidian-wiki/.runtime/kingdee.env"),
|
||||
python_path=Path("/usr/bin/python3"),
|
||||
lookback_hours=48,
|
||||
initial_days=7,
|
||||
)
|
||||
timer = manage_business_sync.render_timer_unit(
|
||||
on_calendar="*-*-* 07:10:00",
|
||||
randomized_delay_sec="15m",
|
||||
)
|
||||
|
||||
self.assertIn("WorkingDirectory=/opt/obsidian-wiki", service)
|
||||
self.assertIn("EnvironmentFile=/opt/obsidian-wiki/.runtime/kingdee.env", service)
|
||||
self.assertIn("/usr/bin/python3 -m obsidian_wiki business sync kingdee", service)
|
||||
self.assertIn("--vault /data/bjd-vault", service)
|
||||
self.assertIn("--config /opt/obsidian-wiki/.runtime/kingdee.env", service)
|
||||
self.assertIn("--incremental", service)
|
||||
self.assertIn("--lookback-hours 48", service)
|
||||
self.assertIn("--initial-days 7", service)
|
||||
self.assertNotIn("KINGDEE_CLIENT_SECRET=", service)
|
||||
self.assertNotIn("KINGDEE_ACCESS_TOKEN=", service)
|
||||
self.assertIn("OnCalendar=*-*-* 07:10:00", timer)
|
||||
self.assertIn("Persistent=true", timer)
|
||||
self.assertIn("RandomizedDelaySec=15m", timer)
|
||||
|
||||
def test_start_timer_writes_units_and_enables_timer(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
project_root = Path(tmp) / "project"
|
||||
vault = Path(tmp) / "vault"
|
||||
config_home = Path(tmp) / "config"
|
||||
project_root.mkdir()
|
||||
vault.mkdir()
|
||||
(project_root / ".runtime").mkdir()
|
||||
(project_root / ".runtime" / "kingdee.env").write_text("KINGDEE_ACGW_IDENTITY=test\n", encoding="utf-8")
|
||||
|
||||
with (
|
||||
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
||||
patch.object(manage_business_sync, "run_systemctl") as run_systemctl,
|
||||
):
|
||||
output = manage_business_sync.start_timer(
|
||||
project_root=project_root,
|
||||
vault_path=vault,
|
||||
python_path=Path("/usr/bin/python3"),
|
||||
)
|
||||
|
||||
paths = manage_business_sync.resolve_service_paths(project_root, config_home=config_home)
|
||||
self.assertTrue(paths.service_unit_path.is_file())
|
||||
self.assertTrue(paths.timer_unit_path.is_file())
|
||||
self.assertIn("Business sync timer is enabled.", output)
|
||||
run_systemctl.assert_has_calls(
|
||||
[
|
||||
call(["daemon-reload"]),
|
||||
call(["enable", "--now", manage_business_sync.TIMER_NAME]),
|
||||
]
|
||||
)
|
||||
|
||||
def test_run_now_starts_service_unit(self):
|
||||
with patch.object(manage_business_sync, "run_systemctl") as run_systemctl:
|
||||
manage_business_sync.run_now()
|
||||
|
||||
run_systemctl.assert_called_once_with(["start", manage_business_sync.SERVICE_NAME])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,275 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
|
||||
from obsidian_wiki.kingdee_api import (
|
||||
DEFAULT_SALORDER_API_URL,
|
||||
ensure_kingdee_config,
|
||||
fetch_kingdee_sales_orders,
|
||||
get_kingdee_access_token,
|
||||
)
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, payload):
|
||||
self.payload = payload
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
def read(self):
|
||||
return json.dumps(self.payload).encode("utf-8")
|
||||
|
||||
|
||||
class RecordingUrlopen:
|
||||
def __init__(self, responses):
|
||||
self.responses = list(responses)
|
||||
self.calls = []
|
||||
|
||||
def __call__(self, request, timeout=60):
|
||||
self.calls.append({"request": request, "timeout": timeout})
|
||||
if not self.responses:
|
||||
raise AssertionError("unexpected HTTP request")
|
||||
return FakeResponse(self.responses.pop(0))
|
||||
|
||||
|
||||
class KingdeeApiTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tempdir = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.tempdir.name)
|
||||
|
||||
def tearDown(self):
|
||||
self.tempdir.cleanup()
|
||||
|
||||
def test_config_command_creates_private_template_without_overwriting(self):
|
||||
config_path = self.root / ".runtime" / "kingdee.env"
|
||||
|
||||
created = ensure_kingdee_config(config_path)
|
||||
config_path.write_text(config_path.read_text(encoding="utf-8") + "\nKINGDEE_TOKEN_URL=https://token.example\n", encoding="utf-8")
|
||||
created_again = ensure_kingdee_config(config_path)
|
||||
|
||||
self.assertTrue(created)
|
||||
self.assertFalse(created_again)
|
||||
self.assertEqual(stat.S_IMODE(config_path.stat().st_mode), 0o600)
|
||||
text = config_path.read_text(encoding="utf-8")
|
||||
self.assertIn(f"KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}", text)
|
||||
self.assertIn("KINGDEE_TOKEN_BODY_JSON=", text)
|
||||
self.assertIn("KINGDEE_TOKEN_URL=https://token.example", text)
|
||||
self.assertIn("是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。", text)
|
||||
self.assertIn("用途：销售订单 API 要求的 x-acgw-identity 请求头。", text)
|
||||
self.assertIn("是否必填：可选，手动测试用；长期定时同步建议使用自动登录配置。", text)
|
||||
self.assertIn("是否必填：可选，默认从 data.access_token 读取。", text)
|
||||
self.assertIn("变量：KINGDEE_CLIENT_ID", text)
|
||||
self.assertIn("变量：KINGDEE_USERNAME", text)
|
||||
self.assertIn("变量：KINGDEE_ACCOUNT_ID", text)
|
||||
self.assertIn("nonce 和 timestamp 会按官方文档自动生成", text)
|
||||
|
||||
def test_default_config_path_accepts_none(self):
|
||||
cwd = Path.cwd()
|
||||
try:
|
||||
os.chdir(self.root)
|
||||
created = ensure_kingdee_config(None)
|
||||
finally:
|
||||
os.chdir(cwd)
|
||||
|
||||
self.assertTrue(created)
|
||||
self.assertTrue((self.root / ".runtime" / "kingdee.env").is_file())
|
||||
|
||||
def test_cli_config_kingdee_writes_template(self):
|
||||
from obsidian_wiki.cli import main
|
||||
|
||||
config_path = self.root / ".runtime" / "kingdee.env"
|
||||
output = StringIO()
|
||||
with redirect_stdout(output):
|
||||
code = main(["business", "config", "kingdee", "--config", str(config_path)])
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.assertTrue(config_path.is_file())
|
||||
self.assertIn("Kingdee config", output.getvalue())
|
||||
|
||||
def test_access_token_is_fetched_and_reused_from_private_cache(self):
|
||||
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
||||
config = {
|
||||
"KINGDEE_TOKEN_URL": "https://token.example/access",
|
||||
"KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc","app_secret":"secret"}',
|
||||
"KINGDEE_TOKEN_HEADERS_JSON": '{"X-App":"demo"}',
|
||||
"KINGDEE_TOKEN_FIELD": "data.access_token",
|
||||
"KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
|
||||
"KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
|
||||
}
|
||||
http = RecordingUrlopen(
|
||||
[
|
||||
{
|
||||
"data": {
|
||||
"access_token": "access-1",
|
||||
"expires_in": 3600000,
|
||||
"refresh_token": "refresh-1",
|
||||
}
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
|
||||
cached = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1100)
|
||||
|
||||
self.assertEqual(token, "access-1")
|
||||
self.assertEqual(cached, "access-1")
|
||||
self.assertEqual(len(http.calls), 1)
|
||||
request = http.calls[0]["request"]
|
||||
self.assertEqual(request.get_method(), "POST")
|
||||
self.assertEqual(request.full_url, "https://token.example/access")
|
||||
self.assertEqual(json.loads(request.data.decode("utf-8"))["app_key"], "abc")
|
||||
self.assertEqual(request.get_header("X-app"), "demo")
|
||||
self.assertEqual(stat.S_IMODE(cache_path.stat().st_mode), 0o600)
|
||||
|
||||
def test_access_token_body_defaults_to_official_fields_with_nonce_and_current_time(self):
|
||||
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
||||
config = {
|
||||
"KINGDEE_TOKEN_URL": "https://token.example/access",
|
||||
"KINGDEE_CLIENT_ID": "thirdapp",
|
||||
"KINGDEE_CLIENT_SECRET": "secret",
|
||||
"KINGDEE_USERNAME": "zhangSan",
|
||||
"KINGDEE_ACCOUNT_ID": "1355633519610561531",
|
||||
"KINGDEE_LANGUAGE": "zh_CN",
|
||||
"KINGDEE_TOKEN_FIELD": "data.access_token",
|
||||
}
|
||||
http = RecordingUrlopen([{"data": {"access_token": "access-1"}}])
|
||||
|
||||
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
|
||||
|
||||
self.assertEqual(token, "access-1")
|
||||
body = json.loads(http.calls[0]["request"].data.decode("utf-8"))
|
||||
self.assertEqual(body["client_id"], "thirdapp")
|
||||
self.assertEqual(body["client_secret"], "secret")
|
||||
self.assertEqual(body["username"], "zhangSan")
|
||||
self.assertEqual(body["accountId"], "1355633519610561531")
|
||||
self.assertEqual(body["language"], "zh_CN")
|
||||
self.assertRegex(body["nonce"], r"^[0-9a-f]{32}$")
|
||||
self.assertEqual(body["timestamp"], "1970-01-01T00:16:40.000Z")
|
||||
|
||||
def test_expired_token_uses_refresh_endpoint(self):
|
||||
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
||||
cache_path.parent.mkdir(parents=True)
|
||||
cache_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"access_token": "old-token",
|
||||
"refresh_token": "refresh-1",
|
||||
"expires_at": 999,
|
||||
"obtained_at": 100,
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
config = {
|
||||
"KINGDEE_TOKEN_URL": "https://token.example/access",
|
||||
"KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc"}',
|
||||
"KINGDEE_REFRESH_URL": "https://token.example/refresh",
|
||||
"KINGDEE_REFRESH_BODY_JSON": '{"refresh_token":"{refresh_token}"}',
|
||||
"KINGDEE_TOKEN_FIELD": "data.access_token",
|
||||
"KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
|
||||
"KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
|
||||
}
|
||||
http = RecordingUrlopen(
|
||||
[
|
||||
{
|
||||
"data": {
|
||||
"access_token": "new-token",
|
||||
"expires_in": 3600000,
|
||||
"refresh_token": "refresh-2",
|
||||
}
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
|
||||
|
||||
self.assertEqual(token, "new-token")
|
||||
request = http.calls[0]["request"]
|
||||
self.assertEqual(request.full_url, "https://token.example/refresh")
|
||||
self.assertEqual(json.loads(request.data.decode("utf-8"))["refresh_token"], "refresh-1")
|
||||
|
||||
def test_sales_order_fetch_posts_filters_headers_and_paginates(self):
|
||||
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
||||
cache_path.parent.mkdir(parents=True)
|
||||
cache_path.write_text(
|
||||
json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
config = {
|
||||
"KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
|
||||
"KINGDEE_ACGW_IDENTITY": "identity-1",
|
||||
"KINGDEE_PAGE_SIZE": "2",
|
||||
}
|
||||
page_1 = {
|
||||
"status": True,
|
||||
"errorCode": "",
|
||||
"message": "",
|
||||
"data": {
|
||||
"pageNo": 1,
|
||||
"pageSize": 2,
|
||||
"rows": [{"id": "1", "billno": "SO-001"}, {"id": "2", "billno": "SO-002"}],
|
||||
"totalCount": 3,
|
||||
},
|
||||
}
|
||||
page_2 = {
|
||||
"status": True,
|
||||
"errorCode": "",
|
||||
"message": "",
|
||||
"data": {
|
||||
"pageNo": 2,
|
||||
"pageSize": 2,
|
||||
"rows": [{"id": "3", "billno": "SO-003"}],
|
||||
"totalCount": 3,
|
||||
},
|
||||
}
|
||||
http = RecordingUrlopen([page_1, page_2])
|
||||
|
||||
rows, payload = fetch_kingdee_sales_orders(
|
||||
config,
|
||||
filters={"start_bizdate": "2026-06-01", "billstatus": ["C"]},
|
||||
token_cache_path=cache_path,
|
||||
urlopen=http,
|
||||
now_epoch=1000,
|
||||
)
|
||||
|
||||
self.assertEqual([row["billno"] for row in rows], ["SO-001", "SO-002", "SO-003"])
|
||||
self.assertEqual(payload["pages_fetched"], 2)
|
||||
self.assertEqual(len(http.calls), 2)
|
||||
first = http.calls[0]["request"]
|
||||
self.assertEqual(first.get_method(), "POST")
|
||||
self.assertEqual(first.get_header("Accesstoken"), "access-1")
|
||||
self.assertEqual(first.get_header("X-acgw-identity"), "identity-1")
|
||||
body = json.loads(first.data.decode("utf-8"))
|
||||
self.assertEqual(body["data"]["start_bizdate"], "2026-06-01")
|
||||
self.assertEqual(body["data"]["billstatus"], ["C"])
|
||||
self.assertEqual(body["pageNo"], 1)
|
||||
self.assertEqual(body["pageSize"], 2)
|
||||
|
||||
def test_sales_order_api_failure_reports_code_and_message(self):
|
||||
cache_path = self.root / ".runtime" / "kingdee-token.json"
|
||||
cache_path.parent.mkdir(parents=True)
|
||||
cache_path.write_text(
|
||||
json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
config = {
|
||||
"KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
|
||||
"KINGDEE_ACGW_IDENTITY": "identity-1",
|
||||
}
|
||||
http = RecordingUrlopen([{"status": False, "errorCode": "E001", "message": "bad filter"}])
|
||||
|
||||
with self.assertRaisesRegex(RuntimeError, "E001.*bad filter"):
|
||||
fetch_kingdee_sales_orders(config, filters={}, token_cache_path=cache_path, urlopen=http, now_epoch=1000)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,144 +0,0 @@
|
||||
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])
|
||||
+4
-12
@@ -131,18 +131,10 @@ class McpQueryTests(unittest.TestCase):
|
||||
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.assertNotIn("obsidian-wiki business", prompt)
|
||||
self.assertIn("independent kingdee-orders MCP", prompt)
|
||||
self.assertIn("Do not read legacy order databases", 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)
|
||||
|
||||
@@ -200,7 +192,7 @@ class McpQueryTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(result["answer_markdown"], result["answer"])
|
||||
self.assertEqual(result["citations"], ["Mini FAKRA", "BJD"])
|
||||
self.assertTrue(result["used_business_facts"])
|
||||
self.assertFalse(result["used_business_facts"])
|
||||
self.assertTrue(result["used_wiki"])
|
||||
self.assertEqual(result["data_freshness"], ["2026-06-02"])
|
||||
self.assertEqual(result["confidence"], "high")
|
||||
|
||||
@@ -1081,9 +1081,9 @@ class RemoteMcpServerTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(capabilities["server_name"], "wiki-file-assets")
|
||||
self.assertIn("server_version", capabilities)
|
||||
self.assertEqual(capabilities["schema_version"], 4)
|
||||
self.assertEqual(capabilities["capabilities_schema_version"], 4)
|
||||
self.assertEqual(capabilities["tools_version"], 4)
|
||||
self.assertEqual(capabilities["schema_version"], 5)
|
||||
self.assertEqual(capabilities["capabilities_schema_version"], 5)
|
||||
self.assertEqual(capabilities["tools_version"], 5)
|
||||
self.assertEqual(capabilities["enabled_tools"], ["file-assets", "ingest", "query"])
|
||||
self.assertIn("file_index", capabilities)
|
||||
self.assertIn("record_count", capabilities["file_index"])
|
||||
|
||||
@@ -15,7 +15,7 @@ import manage_mcp_service
|
||||
|
||||
|
||||
class McpServiceConfigTests(unittest.TestCase):
|
||||
def test_set_business_only_preserves_connection_and_restarts_once_without_opencode(self):
|
||||
def test_set_file_assets_only_preserves_connection_and_restarts_once_without_opencode(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
config_home = root / "config"
|
||||
@@ -31,11 +31,11 @@ class McpServiceConfigTests(unittest.TestCase):
|
||||
patch.object(manage_mcp_service, "resolve_opencode_path", side_effect=AssertionError("no OpenCode")),
|
||||
patch("builtins.print"),
|
||||
):
|
||||
result = manage_mcp_service.main(["tools", "set", "business", "--project-root", str(root)])
|
||||
result = manage_mcp_service.main(["tools", "set", "file-assets", "--project-root", str(root)])
|
||||
self.assertEqual(result, 0)
|
||||
restart.assert_called_once()
|
||||
config = manage_mcp_service.saved_service_config(root)
|
||||
self.assertEqual(config.enabled_tools, ("business",))
|
||||
self.assertEqual(config.enabled_tools, ("file-assets",))
|
||||
self.assertEqual(config.token, "existing-token")
|
||||
self.assertEqual(config.public_base_url, "https://orders.example.test/mcp")
|
||||
self.assertEqual(config.port, 8765)
|
||||
@@ -851,7 +851,7 @@ class McpServiceConfigTests(unittest.TestCase):
|
||||
self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=low\n", env_content)
|
||||
self.assertNotIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=", env_content)
|
||||
self.assertIn("Enabled tool groups:\nfile-assets,query", output)
|
||||
self.assertIn("Use ask_wiki for ordinary wiki, business facts, or hybrid questions.", output)
|
||||
self.assertIn("Use ask_wiki for wiki questions.", output)
|
||||
prepare_runtime.assert_called_once_with(project_root, uv_path=Path("/home/nac/.local/bin/uv"))
|
||||
run_systemctl.assert_has_calls(
|
||||
[
|
||||
|
||||
+2
-20
@@ -21,25 +21,7 @@ def load_user_ops():
|
||||
|
||||
|
||||
class UserOpsTests(unittest.TestCase):
|
||||
def test_kingdee_sync_forwards_to_business_cli(self):
|
||||
user_ops = load_user_ops()
|
||||
|
||||
with patch.object(user_ops, "obsidian_cli_main", return_value=0) as cli_main:
|
||||
status = user_ops.main(["kingdee", "sync", "--vault", "/data/vault", "--incremental"])
|
||||
|
||||
self.assertEqual(status, 0)
|
||||
cli_main.assert_called_once_with(
|
||||
["business", "sync", "kingdee", "--vault", "/data/vault", "--incremental"]
|
||||
)
|
||||
|
||||
def test_timer_install_alias_installs_and_starts_business_timer(self):
|
||||
user_ops = load_user_ops()
|
||||
|
||||
with patch.object(user_ops.manage_business_sync, "main", return_value=0) as timer_main:
|
||||
status = user_ops.main(["timer", "install", "--vault", "/data/vault"])
|
||||
|
||||
self.assertEqual(status, 0)
|
||||
timer_main.assert_called_once_with(["start", "--vault", "/data/vault"])
|
||||
|
||||
def test_mcp_commands_forward_to_mcp_service_manager(self):
|
||||
user_ops = load_user_ops()
|
||||
@@ -59,8 +41,8 @@ class UserOpsTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(status, 0)
|
||||
text = output.getvalue()
|
||||
self.assertIn("kingdee sync", text)
|
||||
self.assertIn("timer install", text)
|
||||
self.assertNotIn("kingdee sync", text)
|
||||
self.assertNotIn("timer install", text)
|
||||
self.assertIn("mcp start", text)
|
||||
|
||||
|
||||
|
||||
@@ -1,31 +1,10 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class WikiSkillInstructionTests(unittest.TestCase):
|
||||
def test_wiki_query_documents_hybrid_business_and_wiki_route(self):
|
||||
text = Path(".skills/wiki-query/SKILL.md").read_text(encoding="utf-8")
|
||||
|
||||
self.assertIn("Hybrid Business + Wiki Route", text)
|
||||
self.assertNotIn("spawn two subagents", text)
|
||||
self.assertIn("two independent tracks", text)
|
||||
self.assertIn("parallel subagents are optional", text)
|
||||
self.assertIn("run the same tracks sequentially", text)
|
||||
self.assertIn("customer/account health", text)
|
||||
self.assertIn("product/material situation", text)
|
||||
self.assertIn("salesperson/team activity", text)
|
||||
self.assertIn("project progress with commercial signal", text)
|
||||
self.assertIn("delivery risk", text)
|
||||
self.assertIn("anomaly explanation", text)
|
||||
self.assertIn("TOP ranking context", text)
|
||||
self.assertIn("quality or complaint impact", text)
|
||||
self.assertIn("ordinary wiki-query", text)
|
||||
self.assertIn("business facts query", text)
|
||||
self.assertIn("fuse the two results", text)
|
||||
self.assertIn("customer_summary", text)
|
||||
self.assertIn("business sql", text)
|
||||
self.assertIn("Do not use hybrid mode when filtered mode is active", text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
def test_wiki_query_keeps_order_access_in_separate_project(self):
|
||||
text = Path(".skills/wiki-query/SKILL.md").read_text()
|
||||
self.assertIn("independent `kingdee-orders-mcp` project", text)
|
||||
self.assertIn("must not read legacy order databases", text)
|
||||
self.assertNotIn("obsidian-wiki business", text)
|
||||
self.assertNotIn("Hybrid Business + Wiki Route", text)
|
||||
|
||||
Reference in New Issue
Block a user