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_client_attaches_csrf_only_to_browser_api_mutations(self) -> None: smoke = load_smoke_module() class Response: status = 200 headers = {} def read(self, _limit): return b"{}" class Opener: def __init__(self): self.requests = [] def open(self, request, timeout): self.requests.append((request, timeout)) return Response() client = smoke.Client("http://crank.test", 5) opener = Opener() client.opener = opener client.csrf_token = "a" * 32 client.request_json("POST", "/api/admin/workspaces/ws/operations", {}) client.request_json("POST", "http://mcp.test/v1/ws/agent", {}) self.assertEqual(opener.requests[0][0].get_header("X-csrf-token"), "a" * 32) self.assertIsNone(opener.requests[1][0].get_header("X-csrf-token")) def test_login_keeps_server_issued_csrf_token(self) -> None: smoke = load_smoke_module() class FakeClient: csrf_token = None def request_json(self, *args, **kwargs): return smoke.JsonResponse(200, {}, {"csrf_token": "b" * 32}) client = FakeClient() smoke.login(client, "owner@crank.test", "safe-password") self.assertEqual(client.csrf_token, "b" * 32) 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, kwargs)) if path.endswith("/operations"): return smoke.JsonResponse(200, {}, {"operation_id": "op_safe", "version": 7}) if path.endswith("/operations/op_safe"): return smoke.JsonResponse(200, {"ETag": '"safe-etag"'}, {"id": "op_safe"}) if path.endswith("/agents/agent_safe"): return smoke.JsonResponse(200, {"ETag": '"safe-agent-etag"'}, {"id": "agent_safe"}) 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) binding_request = next(request for request in client.requests if request[1].endswith("/bindings")) self.assertEqual(binding_request[3]["headers"], {"If-Match": '"safe-agent-etag"'}) publish = next(request for request in client.requests if request[1].endswith("/operations/op_safe/publish")) self.assertEqual(publish[3]["headers"], {"If-Match": '"safe-etag"'}) def test_admin_test_run_uses_created_version_and_reports_safe_typed_failure(self) -> None: smoke = load_smoke_module() trace_id = "0123456789abcdef0123456789abcdef" 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, "trace_id": trace_id, "errors": [ { "code": "outbound_target_rejected", "stage": "adapter", "message": "secret-canary", "context": {"url": "https://private.invalid/token-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.assertEqual( str(raised.exception), "stage=adapter code=outbound_target_rejected trace_id=0123456789abcdef0123456789abcdef", ) for forbidden in ("secret-canary", "private.invalid", "token-canary", "context"): self.assertNotIn(forbidden, str(raised.exception)) def test_admin_test_run_fails_closed_for_malformed_diagnostics(self) -> None: smoke = load_smoke_module() valid_trace_id = "0123456789abcdef0123456789abcdef" valid_failure = {"code": "upstream_timeout", "stage": "upstream"} malformed_results = ( None, [], "secret-canary", {"trace_id": valid_trace_id, "errors": [valid_failure]}, {"ok": None, "trace_id": valid_trace_id, "errors": [valid_failure]}, {"ok": 0, "trace_id": valid_trace_id, "errors": [valid_failure]}, {"ok": "false", "trace_id": valid_trace_id, "errors": [valid_failure]}, {"ok": False, "errors": []}, {"ok": False, "errors": {}}, {"ok": False, "trace_id": valid_trace_id, "errors": [None]}, {"ok": False, "errors": ["secret-canary"]}, {"ok": False, "errors": [valid_failure]}, { "ok": False, "trace_id": valid_trace_id[:-1], "errors": [valid_failure], }, { "ok": False, "trace_id": valid_trace_id + "0", "errors": [valid_failure], }, { "ok": False, "trace_id": "0" * 32, "errors": [valid_failure], }, { "ok": False, "trace_id": valid_trace_id.upper(), "errors": [valid_failure], }, { "ok": False, "trace_id": "a" * 31 + "١", "errors": [valid_failure], }, { "ok": False, "trace_id": "a" * 31 + "\n", "errors": [valid_failure], }, { "ok": False, "trace_id": 42, "errors": [valid_failure], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "", "stage": "upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "a" * 65, "stage": "upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "Upstream_timeout", "stage": "upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_таймаут", "stage": "upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": 42, "stage": "upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_timeout", "stage": ""}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_timeout", "stage": "a" * 65}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_timeout", "stage": "Upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_timeout", "stage": "вверх"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_timeout", "stage": 42}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "unsafe-value://secret-canary", "stage": "upstream"}], }, { "ok": False, "trace_id": valid_trace_id, "errors": [{"code": "upstream_timeout", "stage": "unsafe stage secret-canary"}], }, { "ok": False, "trace_id": "not-a-trace-id-secret-canary", "errors": [{"code": "upstream_timeout", "stage": "upstream"}], }, ) class FakeClient: def __init__(self, result): self.result = result def request_json(self, *args, **kwargs): return smoke.JsonResponse(200, {}, self.result) for result in malformed_results: with self.subTest(result=result), self.assertRaises(smoke.SmokeError) as raised: smoke.run_operation_test(FakeClient(result), "ws", "op", 9) self.assertEqual(str(raised.exception), "stage=operation_test code=outcome_not_ok") self.assertNotIn("secret-canary", str(raised.exception)) def test_admin_test_run_accepts_maximum_length_safe_diagnostics(self) -> None: smoke = load_smoke_module() identifier = "a" * 64 trace_id = "0123456789abcdef0123456789abcdef" class FakeClient: def request_json(self, *args, **kwargs): return smoke.JsonResponse( 200, {}, { "ok": False, "trace_id": trace_id, "request_preview": {"credential": "secret-canary"}, "response_preview": {"token": "secret-canary"}, "errors": [{"code": identifier, "stage": identifier}], }, ) with self.assertRaises(smoke.SmokeError) as raised: smoke.run_operation_test(FakeClient(), "ws", "op", 9) rendered = str(raised.exception) self.assertEqual(rendered, f"stage={identifier} code={identifier} trace_id={trace_id}") self.assertLessEqual(len(rendered.encode("ascii")), 256) self.assertNotIn("secret-canary", rendered) def test_admin_test_run_success_ignores_malformed_diagnostics(self) -> None: smoke = load_smoke_module() class FakeClient: def request_json(self, *args, **kwargs): return smoke.JsonResponse( 200, {}, { "ok": True, "trace_id": "secret-canary", "errors": "secret-canary", "request_preview": {"token": "secret-canary"}, }, ) smoke.run_operation_test(FakeClient(), "ws", "op", 9) def test_edit_and_archive_use_fresh_operation_preconditions(self) -> None: smoke = load_smoke_module() class FakeClient: def __init__(self): self.calls = [] self.etag_index = 0 def request_json(self, method, path, payload=None, **kwargs): self.calls.append((method, path, payload, kwargs)) if method == "GET": self.etag_index += 1 return smoke.JsonResponse(200, {"ETag": f'"etag-{self.etag_index}"'}, {}) if method == "PATCH": return smoke.JsonResponse(200, {}, {"version": 2}) if path.endswith("/archive"): return smoke.JsonResponse(200, {}, {"status": "archived"}) raise AssertionError((method, path)) client = FakeClient() version = smoke.edit_and_archive_operation( client, "ws", "op", "safe_operation", "http://admin-api:3001" ) self.assertEqual(version, 2) patch = next(call for call in client.calls if call[0] == "PATCH") archive = next(call for call in client.calls if call[1].endswith("/archive")) self.assertEqual(patch[3]["headers"], {"If-Match": '"etag-1"'}) self.assertEqual(archive[3]["headers"], {"If-Match": '"etag-2"'}) 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()