"""Behavior checks for HTTP failures and the MCP boundary; never call a remote API.""" import asyncio from datetime import timedelta import unittest import httpx from mcp.shared.memory import create_connected_server_and_client_session from http_api_client import AdapterError, ArtifactClient from http_api_tool import make_server GOOD = {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"} SYNTHETIC_TOKEN = "synthetic-secret-for-test-only" class AdapterTests(unittest.IsolatedAsyncioTestCase): async def test_success_request_shape_timeout_and_client_lifetime(self): requests = [] def handler(request): requests.append(request) return httpx.Response(200, json=GOOD) async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http: result = await ArtifactClient(http, SYNTHETIC_TOKEN).get_artifact("KIT-42") self.assertEqual(result.model_dump(), GOOD) self.assertTrue(http.is_closed) self.assertEqual(len(requests), 1) request = requests[0] self.assertEqual(request.method, "GET") self.assertEqual(str(request.url), "https://inventory.example.invalid/v1/items/KIT-42") self.assertEqual(request.headers["Authorization"], "Bearer " + SYNTHETIC_TOKEN) self.assertNotIn(SYNTHETIC_TOKEN, str(request.url)) self.assertEqual(request.extensions["timeout"], {"connect": 2.0, "read": 5.0, "write": 5.0, "pool": 1.0}) async def test_bad_input_never_reaches_http(self): requests = [] def handler(request): requests.append(request) return httpx.Response(200, json=GOOD) async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http: client = ArtifactClient(http) for item_id in ["../secret", "https://other.invalid", "", "KIT-0", 42]: with self.subTest(item_id=item_id), self.assertRaisesRegex(AdapterError, "INVALID_ID"): await client.get_artifact(item_id) self.assertEqual(requests, []) async def test_statuses_are_safe_and_do_not_retry_or_follow_redirects(self): for status, code in [(401, "AUTH"), (403, "AUTH"), (404, "NOT_FOUND"), (429, "RATE_LIMIT"), (500, "UPSTREAM"), (503, "UPSTREAM"), (302, "HTTP_STATUS"), (400, "HTTP_STATUS")]: requests = [] def handler(request): requests.append(request) return httpx.Response(status, text=SYNTHETIC_TOKEN, headers={"Location": "https://other.invalid/", "Retry-After": "2"}) with self.subTest(status=status): async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http: with self.assertRaises(AdapterError) as caught: await ArtifactClient(http, SYNTHETIC_TOKEN).get_artifact("KIT-42") self.assertEqual(caught.exception.code, code) self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception)) self.assertEqual(len(requests), 1) async def test_timeout_and_connection_failure(self): for exception, code in [(httpx.ReadTimeout, "TIMEOUT"), (httpx.ConnectError, "NETWORK")]: def handler(request): raise exception(SYNTHETIC_TOKEN, request=request) with self.subTest(exception=exception.__name__): async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http: with self.assertRaises(AdapterError) as caught: await ArtifactClient(http).get_artifact("KIT-42") self.assertEqual(caught.exception.code, code) self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception)) async def test_total_deadline_cancels_waiting_handler(self): cancelled = asyncio.Event() async def handler(request): try: await asyncio.Event().wait() finally: cancelled.set() async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http: with self.assertRaisesRegex(AdapterError, "TIMEOUT"): await ArtifactClient(http, deadline=0.03).get_artifact("KIT-42") self.assertTrue(cancelled.is_set()) async def test_invalid_json_schema_and_wrong_record(self): responses = [httpx.Response(200, text="not JSON"), httpx.Response(200, json=[]), httpx.Response(200, json={}), httpx.Response(200, json={**GOOD, "status": "unknown"}), httpx.Response(200, json={**GOOD, "title": 42}), httpx.Response(200, json={**GOOD, "private": SYNTHETIC_TOKEN}), httpx.Response(200, json={**GOOD, "item_id": "KIT-43"})] for response in responses: with self.subTest(body=response.text): async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response), trust_env=False) as http: with self.assertRaises(AdapterError) as caught: await ArtifactClient(http).get_artifact("KIT-42") self.assertEqual(caught.exception.code, "BAD_RESPONSE") self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception)) async def test_mcp_schema_success_bad_input_and_safe_failure(self): requests = [] def handler(request): requests.append(request) return httpx.Response(503, text=SYNTHETIC_TOKEN) if request.url.path.endswith("KIT-43") else httpx.Response(200, json=GOOD) async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http: server = make_server(ArtifactClient(http, SYNTHETIC_TOKEN)) async with create_connected_server_and_client_session(server, read_timeout_seconds=timedelta(seconds=5), raise_exceptions=True) as session: tool = (await session.list_tools()).tools[0] self.assertEqual(tool.name, "get_artifact") self.assertIn("item_id", tool.inputSchema["required"]) self.assertIsNotNone(tool.outputSchema) success = await session.call_tool("get_artifact", {"item_id": "KIT-42"}) self.assertFalse(success.isError) self.assertEqual(success.structuredContent, GOOD) before = len(requests) for arguments in [{}, {"item_id": "../secret"}, {"item_id": 42}]: result = await session.call_tool("get_artifact", arguments) self.assertTrue(result.isError) self.assertEqual(len(requests), before) failure = await session.call_tool("get_artifact", {"item_id": "KIT-43"}) self.assertTrue(failure.isError) self.assertIsNone(failure.structuredContent) self.assertIn("UPSTREAM", failure.content[0].text) self.assertNotIn(SYNTHETIC_TOKEN, failure.model_dump_json()) if __name__ == "__main__": unittest.main(verbosity=2)