"""Unit tests for ``roles/web-app-n8n/files/python/provision/mcp.py``.""" from __future__ import annotations import contextlib import importlib.util import io import json import unittest from unittest.mock import patch from urllib.parse import parse_qs, urlsplit from . import PROJECT_ROOT SCRIPT_PATH = PROJECT_ROOT / "roles/web-app-n8n/files/python/provision/mcp.py" PAGE_SIZE = 10 WORKFLOW_PAGE_SIZE = 100 MAX_PAGE_SIZE = 250 def paged(items, offset, limit): """Return one page the way n8n slices a listing. Args: items: every record the listing holds. offset: index of the first record on the page. limit: requested page size, capped like n8n caps it. """ return items[offset : offset + min(limit, MAX_PAGE_SIZE)] ENV = { "N8N_BASE": "http://n8n:5678", "N8N_OWNER_EMAIL": "administrator@example.org", "N8N_OWNER_PASSWORD": "x" * 48, "N8N_API_KEY_NAME": "infinito:mcp", "N8N_MCP_PATH": "infinito-mcp", "N8N_MCP_TOKEN": "t" * 48, "N8N_MCP_WORKFLOW": "infinito:mcp-server", "N8N_MCP_CREDENTIAL": "infinito:mcp-bearer", } REFERENCE = {"id": "c1", "name": "infinito:mcp-bearer"} def load_script(overrides=None) -> object: spec = importlib.util.spec_from_file_location("provision_mcp", SCRIPT_PATH) module = importlib.util.module_from_spec(spec) with patch.dict("os.environ", {**ENV, **(overrides or {})}): spec.loader.exec_module(module) return module class FakeApi: """Minimal stand-in for the n8n rest and public API surfaces.""" def __init__(self, keys=(), workflows=(), credentials=()): self.keys = [dict(key) for key in keys] self.workflows = [dict(flow) for flow in workflows] self.credentials = [dict(item) for item in credentials] self.created_workflows = [] self.updated_workflows = [] self.activated_workflows = [] self.deleted_workflows = [] self.written_credentials = [] self.deleted_credentials = [] self.deleted_keys = [] self.logins = 0 self.minted = 0 self.scopes = set() def __call__(self, path, method="GET", payload=None, api_key=None): if path == "/rest/login": self.logins += 1 return 200, {"data": {"id": "owner"}} if path == "/rest/credentials" and method == "GET": return 200, {"data": [dict(item) for item in self.credentials]} if path == "/rest/credentials" and method == "POST": self.minted += 1 new = {"id": f"c{self.minted}", "name": payload["name"]} self.written_credentials.append(payload) self.credentials.append(new) return 200, {"data": new} if path.startswith("/rest/credentials/") and method == "PATCH": self.written_credentials.append(payload) return 200, {"data": {"id": path.rsplit("/", 1)[-1]}} if path.startswith("/rest/credentials/") and method == "DELETE": dead = path.rsplit("/", 1)[-1] self.deleted_credentials.append(dead) self.credentials = [c for c in self.credentials if c["id"] != dead] return 200, {"success": True} if path.startswith("/rest/api-keys?") and method == "GET": query = parse_qs(urlsplit(path).query) label = query["label"][0].lower() items = [dict(key) for key in self.keys if label in key["label"].lower()] skip = int(query.get("skip", ["0"])[0]) take = int(query.get("take", [str(PAGE_SIZE)])[0]) return 200, {"data": {"items": paged(items, skip, take)}} if path == "/rest/api-keys" and method == "POST": self.keys = [{"id": "k1", "label": payload["label"]}] self.scopes = set(payload["scopes"]) return 200, {"data": {"id": "k1", "rawApiKey": "n8n-raw-key"}} if path.startswith("/rest/api-keys/") and method == "DELETE": self.deleted_keys.append(path.rsplit("/", 1)[-1]) return 200, {"success": True} if path.split("?")[0] == "/api/v1/workflows" and method == "GET": query = parse_qs(urlsplit(path).query) name = query.get("name", [""])[0] hits = [dict(flow) for flow in self.workflows if name in flow["name"]] cursor = query.get("cursor", [f"0:{WORKFLOW_PAGE_SIZE}"])[0].split(":") offset, limit = (int(cursor[0]), int(cursor[1])) if "limit" in query: limit = int(query["limit"][0]) page = paged(hits, offset, limit) following = offset + len(page) more = following < len(hits) return 200, { "data": page, "nextCursor": f"{following}:{limit}" if more else None, } if path == "/api/v1/workflows" and method == "POST": self.created_workflows.append(payload) return 200, dict(payload, id="w1", active=False) if path.startswith("/api/v1/workflows/") and method == "PUT": self.updated_workflows.append(payload) return 200, payload if path.endswith("/activate") and method == "POST": if "workflow:activate" not in self.scopes: return 403, '{"message":"Forbidden"}' self.activated_workflows.append(path.split("/")[-2]) return 200, {"active": True} if path.startswith("/api/v1/workflows/") and method == "DELETE": self.deleted_workflows.append(path.rsplit("/", 1)[-1]) return 200, {"success": True} raise AssertionError(f"unexpected {method} {path}") class TestProvisionMcp(unittest.TestCase): def test_a_missing_workflow_is_created_and_activated(self) -> None: module = load_script() api = FakeApi() with patch.object(module, "call", api): module.main() self.assertEqual( ["infinito:mcp-server"], [flow["name"] for flow in api.created_workflows] ) self.assertEqual(["w1"], api.activated_workflows) def test_the_trigger_always_requires_a_bearer(self) -> None: module = load_script() node = module.workflow_body(REFERENCE)["nodes"][0] self.assertEqual("bearerAuth", node["parameters"]["authentication"]) self.assertEqual("@n8n/n8n-nodes-langchain.mcpTrigger", node["type"]) self.assertEqual(2, node["typeVersion"]) def test_the_trigger_carries_the_credential_the_bearer_lives_in(self) -> None: module = load_script() node = module.workflow_body(REFERENCE)["nodes"][0] self.assertEqual({"httpBearerAuth": REFERENCE}, node["credentials"]) def test_the_bearer_is_stored_as_an_http_bearer_credential(self) -> None: module = load_script() api = FakeApi() with patch.object(module, "call", api): module.main() self.assertEqual( [ { "name": "infinito:mcp-bearer", "type": "httpBearerAuth", "data": {"token": "t" * 48}, } ], api.written_credentials, ) def test_an_existing_credential_is_overwritten_not_duplicated(self) -> None: module = load_script() api = FakeApi(credentials=[{"id": "c9", "name": "infinito:mcp-bearer"}]) with patch.object(module, "call", api): module.main() node = api.created_workflows[0]["nodes"][0] self.assertEqual("c9", node["credentials"]["httpBearerAuth"]["id"]) def test_a_rotated_bearer_replaces_the_one_the_credential_carries(self) -> None: module = load_script() api = FakeApi(credentials=[{"id": "c9", "name": "infinito:mcp-bearer"}]) with patch.object(module, "call", api): module.main() self.assertEqual( [{"token": "t" * 48}], [item["data"] for item in api.written_credentials], "n8n never hands an existing credential's secret back, so the " "managed one is overwritten rather than compared; a run that " "skipped the write would leave the revoked bearer in place", ) def test_two_managed_credentials_abort_rather_than_guess(self) -> None: module = load_script() api = FakeApi( credentials=[ {"id": "a", "name": "infinito:mcp-bearer"}, {"id": "b", "name": "infinito:mcp-bearer"}, ] ) with patch.object(module, "call", api), self.assertRaises(SystemExit) as exit_: module.main() self.assertIn("2 credentials", str(exit_.exception)) def test_an_empty_bearer_aborts_before_anything_is_written(self) -> None: module = load_script({"N8N_MCP_TOKEN": ""}) api = FakeApi() with patch.object(module, "call", api), self.assertRaises(SystemExit) as exit_: module.main() self.assertIn("unauthenticated", str(exit_.exception)) self.assertEqual(0, api.logins) def test_a_stale_managed_key_is_replaced_not_guessed(self) -> None: module = load_script() api = FakeApi(keys=[{"id": "old", "label": "infinito:mcp"}]) with patch.object(module, "call", api): module.main() self.assertEqual(["old", "k1"], api.deleted_keys) def test_a_stale_managed_key_behind_a_full_page_is_still_replaced(self) -> None: module = load_script() humans = [{"id": f"h{n}", "label": f"human {n}"} for n in range(PAGE_SIZE)] api = FakeApi(keys=[*humans, {"id": "old", "label": "infinito:mcp"}]) with patch.object(module, "call", api): module.main() self.assertEqual(["old", "k1"], api.deleted_keys) def test_an_unknown_listing_shape_names_its_keys_but_never_a_value(self) -> None: module = load_script() api = FakeApi() def odd(path, method="GET", payload=None, api_key=None): if path.startswith("/rest/api-keys?"): return 200, {"data": {"secret": "t0ps3cret"}} return api(path, method, payload, api_key) with patch.object(module, "call", odd), self.assertRaises(SystemExit) as exit_: module.main() self.assertIn("['secret']", str(exit_.exception)) self.assertNotIn("t0ps3cret", str(exit_.exception)) def test_an_unpaged_listing_of_n8n_1_still_finds_the_stale_key(self) -> None: module = load_script() api = FakeApi() def unpaged(path, method="GET", payload=None, api_key=None): if path.startswith("/rest/api-keys?"): return 200, {"data": [{"id": "old", "label": "infinito:mcp"}]} return api(path, method, payload, api_key) with patch.object(module, "call", unpaged): module.main() self.assertEqual(["old", "k1"], api.deleted_keys) def test_a_stale_managed_key_behind_a_full_filtered_page_is_still_replaced( self, ) -> None: module = load_script() near = [ {"id": f"n{n}", "label": f"infinito:mcp-copy-{n}"} for n in range(MAX_PAGE_SIZE) ] api = FakeApi(keys=[*near, {"id": "old", "label": "infinito:mcp"}]) with patch.object(module, "call", api): module.main() self.assertEqual(["old", "k1"], api.deleted_keys) def test_a_managed_workflow_on_a_later_page_is_updated_not_duplicated( self, ) -> None: module = load_script() near = [ {"id": f"n{n}", "name": f"infinito:mcp-server copy {n}"} for n in range(MAX_PAGE_SIZE) ] managed = {"id": "w9", "name": "infinito:mcp-server", "active": True} api = FakeApi(workflows=[*near, managed]) with patch.object(module, "call", api): module.main() self.assertEqual([], api.created_workflows) self.assertEqual(1, len(api.updated_workflows)) def test_a_repeated_workflow_cursor_aborts_rather_than_loops(self) -> None: module = load_script() api = FakeApi() listings = [] def stuck(path, method="GET", payload=None, api_key=None): if path.startswith("/api/v1/workflows?"): listings.append(path) if len(listings) > 3: raise AssertionError("the workflow listing never ends") return 200, {"data": [], "nextCursor": "same"} return api(path, method, payload, api_key) with ( patch.object(module, "call", stuck), self.assertRaises(SystemExit) as exit_, ): module.main() self.assertIn("nextCursor repeats", str(exit_.exception)) def test_a_key_listing_that_ignores_skip_aborts_rather_than_loops(self) -> None: module = load_script() api = FakeApi() full = [ {"id": f"n{n}", "label": f"infinito:mcp-copy-{n}"} for n in range(MAX_PAGE_SIZE) ] listings = [] def frozen(path, method="GET", payload=None, api_key=None): if path.startswith("/rest/api-keys?"): listings.append(path) if len(listings) > 3: raise AssertionError("the key listing never ends") return 200, {"data": {"items": full}} return api(path, method, payload, api_key) with ( patch.object(module, "call", frozen), self.assertRaises(SystemExit) as exit_, ): module.main() self.assertIn("repeats", str(exit_.exception)) def test_a_key_that_only_contains_the_managed_name_is_left_alone(self) -> None: module = load_script() api = FakeApi(keys=[{"id": "near", "label": "infinito:mcp-legacy"}]) with patch.object(module, "call", api): module.main() self.assertEqual(["k1"], api.deleted_keys) def test_the_run_leaves_no_api_key_behind(self) -> None: module = load_script() api = FakeApi() with patch.object(module, "call", api): module.main() self.assertEqual(["k1"], api.deleted_keys) def test_two_managed_keys_abort_rather_than_guess(self) -> None: module = load_script() api = FakeApi( keys=[ {"id": "a", "label": "infinito:mcp"}, {"id": "b", "label": "infinito:mcp"}, ] ) with patch.object(module, "call", api), self.assertRaises(SystemExit) as exit_: module.main() self.assertIn("2 api keys", str(exit_.exception)) def test_an_existing_inactive_workflow_is_updated(self) -> None: module = load_script() api = FakeApi( workflows=[{"id": "w1", "name": "infinito:mcp-server", "active": False}] ) with patch.object(module, "call", api): module.main() self.assertEqual([], api.created_workflows) self.assertEqual( ["infinito:mcp-server"], [flow["name"] for flow in api.updated_workflows] ) self.assertEqual(["w1"], api.activated_workflows) def test_an_active_workflow_still_gets_its_node_repaired(self) -> None: module = load_script() api = FakeApi( workflows=[ { "id": "w1", "name": "infinito:mcp-server", "active": True, "nodes": [{"parameters": {"authentication": "none"}}], } ] ) with patch.object(module, "call", api): module.main() node = api.updated_workflows[0]["nodes"][0] self.assertEqual("bearerAuth", node["parameters"]["authentication"]) self.assertIn("httpBearerAuth", node["credentials"]) def test_the_only_served_tool_is_the_connected_one(self) -> None: module = load_script() body = module.workflow_body(REFERENCE) tool = body["nodes"][1] self.assertEqual("@n8n/n8n-nodes-langchain.toolCode", tool["type"]) self.assertEqual( { "infinito_health": { "ai_tool": [ [ { "node": "Infinito MCP Trigger", "type": "ai_tool", "index": 0, } ] ] } }, body["connections"], ) def test_the_served_tool_declares_its_input_schema(self) -> None: module = load_script() tool = module.workflow_body(REFERENCE)["nodes"][1] self.assertIs(True, tool["parameters"]["specifyInputSchema"]) self.assertEqual("manual", tool["parameters"]["schemaType"]) schema = json.loads(tool["parameters"]["inputSchema"]) self.assertEqual(["echo"], schema["required"]) self.assertIs(False, schema["additionalProperties"]) def test_the_served_tool_returns_a_string(self) -> None: module = load_script() tool = module.workflow_body(REFERENCE)["nodes"][1] self.assertIn("JSON.stringify", tool["parameters"]["jsCode"]) def test_a_converged_deployment_reruns_unchanged(self) -> None: module = load_script() credential = {"id": "c9", "name": "infinito:mcp-bearer"} api = FakeApi( credentials=[credential], workflows=[ { "id": "w1", "name": "infinito:mcp-server", "active": True, "nodes": module.workflow_body(credential)["nodes"], } ], ) out = io.StringIO() with patch.object(module, "call", api), contextlib.redirect_stdout(out): module.main() self.assertEqual("OK", out.getvalue().strip()) def test_resources_outside_the_managed_names_are_left_alone(self) -> None: module = load_script() api = FakeApi( keys=[{"id": "human-key", "label": "someone else"}], credentials=[{"id": "human-cred", "name": "someone else"}], workflows=[{"id": "human-flow", "name": "someone else", "active": True}], ) with patch.object(module, "call", api): module.main() self.assertNotIn("human-key", api.deleted_keys) self.assertEqual([], api.updated_workflows) self.assertEqual( ["infinito:mcp-server"], [flow["name"] for flow in api.created_workflows] ) self.assertEqual( ["infinito:mcp-bearer"], [item["name"] for item in api.written_credentials] ) def test_two_managed_workflows_abort_rather_than_guess(self) -> None: module = load_script() api = FakeApi( workflows=[ {"id": "a", "name": "infinito:mcp-server"}, {"id": "b", "name": "infinito:mcp-server"}, ] ) with patch.object(module, "call", api), self.assertRaises(SystemExit) as exit_: module.main() self.assertIn("2 workflows", str(exit_.exception)) class TestRecordsRejectsAShapeItCannotRead(unittest.TestCase): def setUp(self) -> None: self.module = load_script() def test_a_list_of_records_passes_through(self) -> None: rows = [{"id": "a"}, {"id": "b"}] self.assertEqual(rows, self.module.records({"data": rows}, "api keys")) self.assertEqual(rows, self.module.records(rows, "api keys")) def test_an_absent_payload_is_empty(self) -> None: self.assertEqual([], self.module.records({"data": None}, "api keys")) self.assertEqual([], self.module.records(None, "api keys")) def test_a_string_body_names_its_type_and_the_endpoint(self) -> None: with self.assertRaises(SystemExit) as exit_: self.module.records("gateway timeout", "api keys") self.assertIn("api keys", str(exit_.exception)) self.assertIn("str", str(exit_.exception)) def test_a_mapping_that_is_not_the_paged_shape_names_its_keys(self) -> None: with self.assertRaises(SystemExit) as exit_: self.module.records({"data": {"data": [{"id": "a"}]}}, "workflows") self.assertIn("workflows", str(exit_.exception)) self.assertIn("data", str(exit_.exception)) def test_a_list_of_strings_names_the_offending_entries(self) -> None: with self.assertRaises(SystemExit) as exit_: self.module.records({"data": ["a", "b"]}, "credentials") self.assertIn("credentials", str(exit_.exception)) self.assertIn("str", str(exit_.exception)) def test_the_failure_never_carries_the_body(self) -> None: with self.assertRaises(SystemExit) as exit_: self.module.records({"data": "s3cret-token"}, "credentials") self.assertNotIn("s3cret-token", str(exit_.exception)) if __name__ == "__main__": unittest.main()