172 lines
6.6 KiB
Python
172 lines
6.6 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from nocobase_crm_mcp.nb_client import NbClient, NbCommandError
|
|
|
|
|
|
class NbClientTests(unittest.TestCase):
|
|
def test_list_records_builds_safe_nb_command_and_environment(self) -> None:
|
|
calls: list[dict] = []
|
|
|
|
def runner(cmd, *, cwd, env, timeout, capture_output, text, check):
|
|
calls.append(
|
|
{
|
|
"cmd": cmd,
|
|
"cwd": cwd,
|
|
"env": env,
|
|
"timeout": timeout,
|
|
"capture_output": capture_output,
|
|
"text": text,
|
|
"check": check,
|
|
}
|
|
)
|
|
return subprocess.CompletedProcess(cmd, 0, stdout='{"data":[{"id":1}],"meta":{"count":1}}', stderr="")
|
|
|
|
client = NbClient(env_name="nb-mcp", data_source="main", runner=runner)
|
|
|
|
result = client.list_records(
|
|
"customers",
|
|
filter={"$and": [{"status": {"$eq": "active"}}]},
|
|
fields=["id", "name"],
|
|
sort=["-updatedAt"],
|
|
page=2,
|
|
page_size=20,
|
|
)
|
|
|
|
self.assertEqual(result["data"], [{"id": 1}])
|
|
command = calls[0]["cmd"]
|
|
self.assertEqual(command[:4], ["nb", "api", "resource", "list"])
|
|
self.assertIn("--resource", command)
|
|
self.assertIn("customers", command)
|
|
self.assertIn("--env", command)
|
|
self.assertIn("nb-mcp", command)
|
|
self.assertIn("--yes", command)
|
|
self.assertIn("--json-output", command)
|
|
self.assertIn("--data-source", command)
|
|
self.assertIn("main", command)
|
|
self.assertIn("--filter", command)
|
|
self.assertIn(json.dumps({"$and": [{"status": {"$eq": "active"}}]}, separators=(",", ":")), command)
|
|
self.assertIn("--fields", command)
|
|
self.assertIn("id", command)
|
|
self.assertIn("--sort", command)
|
|
self.assertIn("-updatedAt", command)
|
|
self.assertEqual(calls[0]["env"]["NB_CLI_LOG_DISABLED"], "1")
|
|
self.assertEqual(calls[0]["env"]["NB_SKIP_STARTUP_UPDATE"], "1")
|
|
self.assertTrue(calls[0]["capture_output"])
|
|
self.assertTrue(calls[0]["text"])
|
|
self.assertFalse(calls[0]["check"])
|
|
|
|
def test_get_record_and_query_records_use_expected_commands(self) -> None:
|
|
commands: list[list[str]] = []
|
|
|
|
def runner(cmd, **kwargs):
|
|
commands.append(cmd)
|
|
return subprocess.CompletedProcess(cmd, 0, stdout='{"data":[]}', stderr="")
|
|
|
|
client = NbClient(env_name="nb-mcp", runner=runner)
|
|
client.get_record("customers", 7, fields=["id"])
|
|
client.query_records(
|
|
"customers",
|
|
measures=[{"field": ["id"], "aggregation": "count", "alias": "count"}],
|
|
dimensions=[{"field": ["status"], "alias": "status"}],
|
|
orders=[{"field": ["status"], "alias": "status", "order": "asc"}],
|
|
limit=10,
|
|
timezone="+08:00",
|
|
)
|
|
|
|
self.assertEqual(commands[0][:4], ["nb", "api", "resource", "get"])
|
|
self.assertIn("--filter-by-tk", commands[0])
|
|
self.assertIn("7", commands[0])
|
|
self.assertEqual(commands[1][:4], ["nb", "api", "resource", "query"])
|
|
self.assertIn("--measures", commands[1])
|
|
self.assertIn("--dimensions", commands[1])
|
|
self.assertIn("--orders", commands[1])
|
|
self.assertIn("--timezone", commands[1])
|
|
|
|
def test_list_collection_fields_uses_data_modeling_command(self) -> None:
|
|
commands: list[list[str]] = []
|
|
|
|
def runner(cmd, **kwargs):
|
|
commands.append(cmd)
|
|
return subprocess.CompletedProcess(
|
|
cmd,
|
|
0,
|
|
stdout='{"data":[{"name":"f_customer_name","title":"客户名称","interface":"input"}]}',
|
|
stderr="",
|
|
)
|
|
|
|
result = NbClient(env_name="nb-mcp", runner=runner).list_collection_fields("t_nt3pjadelqi")
|
|
|
|
self.assertEqual(result["data"][0]["name"], "f_customer_name")
|
|
self.assertEqual(
|
|
commands[0][:6],
|
|
["nb", "api", "data-modeling", "collections", "fields", "list"],
|
|
)
|
|
self.assertIn("--collection-name", commands[0])
|
|
self.assertIn("t_nt3pjadelqi", commands[0])
|
|
self.assertIn("--env", commands[0])
|
|
self.assertIn("nb-mcp", commands[0])
|
|
self.assertIn("--yes", commands[0])
|
|
self.assertIn("--json-output", commands[0])
|
|
|
|
def test_get_collection_uses_data_modeling_command(self) -> None:
|
|
commands: list[list[str]] = []
|
|
|
|
def runner(cmd, **kwargs):
|
|
commands.append(cmd)
|
|
return subprocess.CompletedProcess(cmd, 0, stdout='{"data":{"name":"t_nt3pjadelqi"}}', stderr="")
|
|
|
|
NbClient(env_name="nb-mcp", runner=runner).get_collection("t_nt3pjadelqi", appends=["fields"])
|
|
|
|
self.assertEqual(commands[0][:5], ["nb", "api", "data-modeling", "collections", "get"])
|
|
self.assertIn("--filter-by-tk", commands[0])
|
|
self.assertIn("t_nt3pjadelqi", commands[0])
|
|
self.assertIn("--appends", commands[0])
|
|
self.assertIn("fields", commands[0])
|
|
|
|
def test_command_errors_redact_api_token(self) -> None:
|
|
def runner(cmd, **kwargs):
|
|
return subprocess.CompletedProcess(
|
|
cmd,
|
|
1,
|
|
stdout="",
|
|
stderr="Authorization failed for Bearer super-secret-token and --token super-secret-token",
|
|
)
|
|
|
|
client = NbClient(env_name="nb-mcp", api_token="super-secret-token", runner=runner)
|
|
|
|
with self.assertRaises(NbCommandError) as caught:
|
|
client.list_records("customers")
|
|
|
|
message = str(caught.exception)
|
|
self.assertNotIn("super-secret-token", message)
|
|
self.assertIn("<redacted>", message)
|
|
|
|
def test_invalid_json_output_raises_command_error(self) -> None:
|
|
def runner(cmd, **kwargs):
|
|
return subprocess.CompletedProcess(cmd, 0, stdout="HTTP 200", stderr="")
|
|
|
|
client = NbClient(env_name="nb-mcp", runner=runner)
|
|
|
|
with self.assertRaises(NbCommandError):
|
|
client.list_records("customers")
|
|
|
|
def test_environment_overrides_can_change_nb_home_without_shell(self) -> None:
|
|
captured_env = {}
|
|
|
|
def runner(cmd, *, env, **kwargs):
|
|
captured_env.update(env)
|
|
return subprocess.CompletedProcess(cmd, 0, stdout='{"data":[]}', stderr="")
|
|
|
|
with patch.dict(os.environ, {"NB_CLI_ROOT": "/tmp/custom-nb-root"}, clear=True):
|
|
NbClient(env_name="nb-mcp", runner=runner).list_records("customers")
|
|
|
|
self.assertEqual(captured_env["NB_CLI_ROOT"], "/tmp/custom-nb-root")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|