from __future__ import annotations from pathlib import Path import pytest import yaml from services.runtime.connector_appops import ( PUBLIC_TO_UPSTREAM, UPSTREAM_TO_PUBLIC, _validate_endpoint, rewrite_request, rewrite_response, upstream_headers, ) ROOT = Path(__file__).resolve().parents[1] def test_appops_tool_catalog_is_bijective_and_server_namespaced() -> None: assert len(UPSTREAM_TO_PUBLIC) == len(PUBLIC_TO_UPSTREAM) == 29 assert all(name.startswith("server.") for name in PUBLIC_TO_UPSTREAM) assert PUBLIC_TO_UPSTREAM["server.summary.get"] == "get_system_summary" assert PUBLIC_TO_UPSTREAM["server.operation.execute"] == "server.operation.execute" assert PUBLIC_TO_UPSTREAM["server.project.list"] == "gitea.projects.catalog" assert PUBLIC_TO_UPSTREAM["server.project.reconcile"] == "gitea.projects.reconcile" assert PUBLIC_TO_UPSTREAM["server.deployment.apply"] == "deployment.apply" assert PUBLIC_TO_UPSTREAM["server.deployment.rollback"] == "deployment.rollback" assert PUBLIC_TO_UPSTREAM["server.deployment.target.enroll"] == "target.enroll" assert PUBLIC_TO_UPSTREAM["server.ci.runners.status"] == "gitea.actions.runners.status" assert PUBLIC_TO_UPSTREAM["server.appops.deployment.status"] == "appops.deployment.status" assert ( PUBLIC_TO_UPSTREAM["server.appops.self_upgrade.preflight"] == "appops.self_upgrade.preflight" ) assert PUBLIC_TO_UPSTREAM["server.appops.self_upgrade.apply"] == "appops.self_upgrade.apply" assert PUBLIC_TO_UPSTREAM["server.appops.self_upgrade.status"] == "appops.self_upgrade.status" def test_every_public_appops_tool_has_governance_metadata() -> None: canonical_names: set[str] = set() for path in (ROOT / "catalog" / "tools").glob("*.yaml"): document = yaml.safe_load(path.read_text(encoding="utf-8")) spec = document.get("spec") if isinstance(document, dict) else None if isinstance(spec, dict) and spec.get("connectorType") == "appops": name = spec.get("canonicalName") if isinstance(name, str): canonical_names.add(name) assert set(PUBLIC_TO_UPSTREAM) <= canonical_names def test_appops_self_upgrade_manifests_are_fixed_target_and_fingerprint_bound() -> None: specs = {} for path in (ROOT / "catalog" / "tools").glob("*.yaml"): document = yaml.safe_load(path.read_text(encoding="utf-8")) spec = document.get("spec") if isinstance(document, dict) else None if isinstance(spec, dict) and spec.get("canonicalName", "").startswith("server.appops."): specs[spec["canonicalName"]] = spec assert set(specs) == { "server.appops.deployment.status", "server.appops.self_upgrade.preflight", "server.appops.self_upgrade.apply", "server.appops.self_upgrade.status", } for name in ( "server.appops.self_upgrade.preflight", "server.appops.self_upgrade.apply", ): schema = specs[name]["inputSchema"] assert schema["properties"]["repository"]["const"] == "jens/unraid-appops-gateway" assert schema["properties"]["revision"]["pattern"] == "^[0-9a-f]{40}$" assert schema["properties"]["reason"]["minLength"] == 8 assert schema["properties"]["reason"]["maxLength"] == 500 assert schema["properties"]["expected_current_revision"]["pattern"] == "^[0-9a-f]{40}$" apply = specs["server.appops.self_upgrade.apply"] assert apply["riskClass"] == "R3" assert apply["readOnly"] is False assert ( apply["inputSchema"]["properties"]["expected_preflight_fingerprint"]["pattern"] == "^[0-9a-f]{64}$" ) assert { "expected_preflight_fingerprint", "idempotency_key", } <= set(apply["inputSchema"]["required"]) policy = yaml.safe_load( (ROOT / "catalog" / "policies" / "041-appops-governed-operations.yaml").read_text( encoding="utf-8" ) ) assert "server.appops.self_upgrade.apply" in policy["spec"]["match"]["toolPatterns"] def test_appops_proxy_rewrites_only_registered_tool_calls() -> None: request = { "jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "server.operation.preview", "arguments": {"action": "x"}}, } rewritten = rewrite_request(request) assert rewritten["params"]["name"] == "preview_action" deployment_request = { "jsonrpc": "2.0", "id": 8, "method": "tools/call", "params": {"name": "server.deployment.apply", "arguments": {"plan": {}}}, } deployment_rewritten = rewrite_request(deployment_request) assert deployment_rewritten["params"]["name"] == "deployment.apply" reconcile_request = { "jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": "server.project.reconcile", "arguments": {}}, } reconcile_rewritten = rewrite_request(reconcile_request) assert reconcile_rewritten["params"]["name"] == "gitea.projects.reconcile" with pytest.raises(ValueError, match="not registered"): rewrite_request({"method": "tools/call", "params": {"name": "shell.execute"}}) upgrade_request = { "jsonrpc": "2.0", "id": 10, "method": "tools/call", "params": { "name": "server.appops.self_upgrade.apply", "arguments": { "repository": "jens/unraid-appops-gateway", "revision": "a" * 40, "reason": "approved release", "expected_preflight_fingerprint": "b" * 64, "idempotency_key": "upgrade-2026-09-30", "expected_current_revision": "c" * 40, }, }, } upgrade_rewritten = rewrite_request(upgrade_request) assert upgrade_rewritten["params"]["name"] == "appops.self_upgrade.apply" assert upgrade_rewritten["params"]["arguments"] == upgrade_request["params"]["arguments"] for upstream_name in ( "appops.deployment.status", "appops.self_upgrade.preflight", "appops.self_upgrade.apply", "appops.self_upgrade.status", ): with pytest.raises(ValueError, match="not registered"): rewrite_request({"method": "tools/call", "params": {"name": upstream_name}}) def test_appops_proxy_rewrites_discovery_and_drops_unknown_tools() -> None: response = { "jsonrpc": "2.0", "id": 1, "result": { "tools": [ {"name": "list_containers", "description": "containers"}, {"name": "gitea.projects.catalog", "description": "projects"}, {"name": "deployment.preflight", "description": "deployment preflight"}, {"name": "deployment.apply", "description": "typed deployment"}, {"name": "appops.deployment.status", "description": "AppOps revision"}, {"name": "appops.self_upgrade.preflight", "description": "upgrade preflight"}, {"name": "appops.self_upgrade.apply", "description": "upgrade apply"}, {"name": "appops.self_upgrade.status", "description": "upgrade status"}, {"name": "future_unsafe_tool", "description": "unknown"}, ] }, } assert rewrite_response(response, "tools/list")["result"]["tools"] == [ {"name": "server.container.list", "description": "containers"}, {"name": "server.project.list", "description": "projects"}, {"name": "server.deployment.preflight", "description": "deployment preflight"}, {"name": "server.deployment.apply", "description": "typed deployment"}, {"name": "server.appops.deployment.status", "description": "AppOps revision"}, {"name": "server.appops.self_upgrade.preflight", "description": "upgrade preflight"}, {"name": "server.appops.self_upgrade.apply", "description": "upgrade apply"}, {"name": "server.appops.self_upgrade.status", "description": "upgrade status"}, ] def test_appops_proxy_forwards_only_bounded_transport_headers() -> None: assert upstream_headers( { "content-type": "application/json", "mcp-session-id": "session-7", "mcp-protocol-version": "2025-06-18", "cookie": "must-not-leak", "x-itworx-client-id": "must-not-leak-upstream", }, "service-token", has_body=True, ) == { "Authorization": "Bearer service-token", "Accept": "application/json, text/event-stream", "Content-Type": "application/json", "mcp-session-id": "session-7", "mcp-protocol-version": "2025-06-18", } @pytest.mark.parametrize( "endpoint", [ "http://127.0.0.1:1216", "http://198.18.10.150:1216", "https://appops.example.com", ], ) def test_appops_endpoint_accepts_private_http_or_https_origins(endpoint: str) -> None: assert _validate_endpoint(endpoint) == endpoint @pytest.mark.parametrize( "endpoint", [ "http://example.com:1216", "https://user:pass@appops.example", "https://appops.example/path", "file:///data/appops.db", ], ) def test_appops_endpoint_rejects_unsafe_targets(endpoint: str) -> None: with pytest.raises(ValueError): _validate_endpoint(endpoint)