200 lines
8.6 KiB
Python
200 lines
8.6 KiB
Python
import importlib.util
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SMOKE = ROOT / "scripts" / "authenticated-product-smoke.py"
|
|
|
|
|
|
def load_smoke_module():
|
|
spec = importlib.util.spec_from_file_location("authenticated_product_smoke", SMOKE)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class AuthenticatedProductSmokeTests(unittest.TestCase):
|
|
def test_operation_payload_uses_internal_upstream(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
payload = smoke.build_operation_payload(
|
|
name="internal_health_smoke_123",
|
|
upstream_base_url="http://admin-api:3001",
|
|
)
|
|
|
|
self.assertEqual(payload["protocol"], "rest")
|
|
self.assertEqual(payload["target"]["method"], "GET")
|
|
self.assertEqual(payload["target"]["base_url"], "http://admin-api:3001")
|
|
self.assertEqual(payload["target"]["path_template"], "/health")
|
|
self.assertNotIn("open-meteo", str(payload).lower())
|
|
self.assertNotIn("frankfurter", str(payload).lower())
|
|
|
|
def test_public_mcp_url_keeps_proxy_prefix(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
self.assertEqual(
|
|
smoke.agent_mcp_url("https://crank.example.com/", "solo", "health-smoke"),
|
|
"https://crank.example.com/mcp/v1/solo/health-smoke",
|
|
)
|
|
self.assertEqual(
|
|
smoke.agent_mcp_url("http://127.0.0.1:3302", "solo", "health-smoke", ""),
|
|
"http://127.0.0.1:3302/v1/solo/health-smoke",
|
|
)
|
|
|
|
def test_resolve_workspace_prefers_authenticated_session_slug(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
class FakeClient:
|
|
def request_json(self, method, path):
|
|
self.request = (method, path)
|
|
return smoke.JsonResponse(
|
|
status=200,
|
|
headers={},
|
|
body={
|
|
"current_workspace_id": "ws_default",
|
|
"memberships": [
|
|
{
|
|
"workspace": {
|
|
"id": "ws_default",
|
|
"slug": "solo",
|
|
}
|
|
}
|
|
],
|
|
},
|
|
)
|
|
|
|
client = FakeClient()
|
|
|
|
self.assertEqual(
|
|
smoke.resolve_workspace(client, "ws_default", "default"),
|
|
("ws_default", "solo"),
|
|
)
|
|
self.assertEqual(client.request, ("GET", "/api/auth/session"))
|
|
|
|
def test_tool_call_payload_targets_health_tool(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
payload = smoke.tools_call_payload("internal_health_smoke", {"probe": "ok"})
|
|
|
|
self.assertEqual(payload["jsonrpc"], "2.0")
|
|
self.assertEqual(payload["method"], "tools/call")
|
|
self.assertEqual(payload["params"]["name"], "internal_health_smoke")
|
|
self.assertEqual(payload["params"]["arguments"], {"probe": "ok"})
|
|
|
|
def test_tool_call_requires_non_error_expected_structured_outcome(self) -> None:
|
|
smoke = load_smoke_module()
|
|
smoke.validate_tool_call_result({"result": {"isError": False, "structuredContent": {"status": "ok"}}})
|
|
for value in (
|
|
{"result": {"isError": True, "structuredContent": {"status": "ok"}}},
|
|
{"result": {"isError": False, "structuredContent": {"status": "failed"}}},
|
|
{"error": {"message": "secret-canary"}},
|
|
[],
|
|
):
|
|
with self.subTest(value=value), self.assertRaises(smoke.SmokeError) as raised:
|
|
smoke.validate_tool_call_result(value)
|
|
self.assertNotIn("secret-canary", str(raised.exception))
|
|
|
|
def test_tools_list_and_key_shape_fail_through_safe_error(self) -> None:
|
|
smoke = load_smoke_module()
|
|
for value in (None, {"result": {"tools": ["bad"]}}, {"result": {"tools": [{"name": 7}]}}):
|
|
with self.subTest(value=value), self.assertRaises(smoke.SmokeError):
|
|
smoke.validate_tools_list(value, "expected")
|
|
|
|
class BadKeyClient:
|
|
def request_json(self, *args, **kwargs):
|
|
return smoke.JsonResponse(200, {}, {"api_key": {}})
|
|
|
|
with self.assertRaises(smoke.SmokeError):
|
|
smoke.create_agent_key(BadKeyClient(), "ws", "agent")
|
|
|
|
def test_session_termination_uses_delete_without_exposing_key(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
class FakeClient:
|
|
def request_json(self, method, url, **kwargs):
|
|
self.call = (method, url, kwargs)
|
|
return smoke.JsonResponse(204, {}, None)
|
|
|
|
client = FakeClient()
|
|
smoke.terminate_mcp_session(client, "https://private.invalid/mcp", "secret-canary", "session-safe")
|
|
self.assertEqual(client.call[0], "DELETE")
|
|
self.assertEqual(client.call[2]["expected"], (204, 404))
|
|
|
|
def test_operation_and_agent_versions_come_from_api_responses(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.requests = []
|
|
|
|
def request_json(self, method, path, payload=None, **kwargs):
|
|
self.requests.append((method, path, payload))
|
|
if path.endswith("/operations"):
|
|
return smoke.JsonResponse(200, {}, {"operation_id": "op_safe", "version": 7})
|
|
if path.endswith("/publish") and "/operations/" in path:
|
|
return smoke.JsonResponse(200, {}, {"published_version": 7})
|
|
if path.endswith("/agents"):
|
|
return smoke.JsonResponse(200, {}, {"agent_id": "agent_safe", "version": 3})
|
|
if path.endswith("/bindings"):
|
|
return smoke.JsonResponse(200, {}, {})
|
|
if path.endswith("/publish") and "/agents/" in path:
|
|
return smoke.JsonResponse(200, {}, {"published_version": 3})
|
|
raise AssertionError(path)
|
|
|
|
client = FakeClient()
|
|
operation_id, operation_version = smoke.create_operation(client, "ws", "safe", "http://admin-api:3001")
|
|
published_operation_version = smoke.publish_operation(client, "ws", operation_id, operation_version)
|
|
agent_id, agent_version = smoke.create_agent(client, "ws", "safe-agent")
|
|
published_agent_version = smoke.bind_and_publish_agent(
|
|
client, "ws", agent_id, agent_version, operation_id, published_operation_version, "safe"
|
|
)
|
|
|
|
self.assertEqual((operation_id, operation_version, published_operation_version), ("op_safe", 7, 7))
|
|
self.assertEqual((agent_id, agent_version, published_agent_version), ("agent_safe", 3, 3))
|
|
binding = next(payload for _, path, payload in client.requests if path.endswith("/bindings"))[0]
|
|
self.assertEqual(binding["operation_version"], 7)
|
|
|
|
def test_admin_test_run_uses_created_version_and_rejects_failed_outcome(self) -> None:
|
|
smoke = load_smoke_module()
|
|
|
|
class FakeClient:
|
|
def __init__(self, ok):
|
|
self.ok = ok
|
|
self.payload = None
|
|
|
|
def request_json(self, method, path, payload=None, **kwargs):
|
|
self.payload = payload
|
|
return smoke.JsonResponse(200, {}, {"ok": self.ok, "errors": [{"message": "secret-canary"}]})
|
|
|
|
passing = FakeClient(True)
|
|
smoke.run_operation_test(passing, "ws", "op", 9)
|
|
self.assertEqual(passing.payload, {"version": 9, "input": {"probe": "ok"}})
|
|
|
|
with self.assertRaises(smoke.SmokeError) as raised:
|
|
smoke.run_operation_test(FakeClient(False), "ws", "op", 9)
|
|
self.assertNotIn("secret-canary", str(raised.exception))
|
|
|
|
def test_http_failure_message_is_bounded_and_redacted(self) -> None:
|
|
smoke = load_smoke_module()
|
|
error = smoke.safe_error("http", "unexpected_status", status=500)
|
|
rendered = str(error)
|
|
self.assertEqual(rendered, "stage=http code=unexpected_status status=500")
|
|
self.assertNotIn("http://", rendered)
|
|
self.assertLessEqual(len(rendered.encode("utf-8")), 256)
|
|
|
|
def test_safe_summary_contains_revisions_but_not_secrets_or_payloads(self) -> None:
|
|
smoke = load_smoke_module()
|
|
summary = smoke.build_safe_summary("op_safe", 7, "agent_safe", 3)
|
|
encoded = json.dumps(summary, sort_keys=True)
|
|
self.assertIn('"operation_version": 7', encoded)
|
|
self.assertIn('"agent_revision": 3', encoded)
|
|
for forbidden in ("secret", "cookie", "authorization", "payload", "session"):
|
|
self.assertNotIn(forbidden, encoded.lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|