decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project.

tests/test_checker.py

@@ -2,51 +2,75 @@ import httpx
 import pytest
 
 from app.checker import EndpointChecker
-
-pytestmark = pytest.mark.anyio
+from app.config import Settings
+from app.models import MonitorState
+from app.security import UnsafeTarget
 
 
-async def test_blocks_private_dns_before_http_call():
+async def public_resolver(host: str, port: int) -> list[str]:
+    return ["93.184.216.34"]
+
+
+@pytest.mark.anyio
+async def test_http_results_and_redirect_are_checked() -> None:
+    seen: list[str] = []
+
+    async def resolver(host: str, port: int) -> list[str]:
+        seen.append(host)
+        return ["93.184.216.34"]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.host == "example.com":
+            return httpx.Response(302, headers={"location": "https://other.example/final"})
+        return httpx.Response(503)
+
+    checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=resolver)
+    result = await checker.check("id", "https://example.com/start?token=secret")
+    assert result.state is MonitorState.DOWN
+    assert result.http_status == 503
+    assert result.latency_ms is not None
+    assert seen == ["example.com", "other.example"]
+
+
+@pytest.mark.anyio
+async def test_transport_error_becomes_error_status() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectTimeout("timed out", request=request)
+
+    checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=public_resolver)
+    result = await checker.check("id", "https://example.com")
+    assert result.state is MonitorState.ERROR
+    assert "timed out" in (result.error or "")
+
+
+@pytest.mark.anyio
+async def test_private_dns_is_blocked_before_transport() -> None:
     called = False
-    def handler(request):
+
+    async def private_resolver(host: str, port: int) -> list[str]:
+        return ["127.0.0.1"]
+
+    def handler(request: httpx.Request) -> httpx.Response:
         nonlocal called
         called = True
-        return httpx.Response(200, request=request)
-    async def private(host, port):
-        return ["127.0.0.1"]
-    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
-        result = await EndpointChecker(client, 1, 2, 100, private).check("id", "http://public.test")
-    assert result.status == "error"
-    assert result.error == "non_public_target"
-    assert not called
+        return httpx.Response(200)
+
+    checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=private_resolver)
+    with pytest.raises(UnsafeTarget):
+        await checker.check("id", "https://internal.example")
+    assert called is False
 
 
-async def test_blocks_redirect_to_private_target():
-    calls = []
-    def handler(request):
-        calls.append(str(request.url))
-        return httpx.Response(302, headers={"location": "http://internal.test/admin"}, request=request)
-    async def resolver(host, port):
-        return ["10.0.0.1"] if host == "internal.test" else ["93.184.216.34"]
-    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
-        result = await EndpointChecker(client, 1, 2, 100, resolver).check("id", "https://example.com")
-    assert result.error == "non_public_target"
-    assert calls == ["https://example.com"]
+@pytest.mark.anyio
+async def test_redirect_to_private_address_is_blocked() -> None:
+    calls = 0
 
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(302, headers={"location": "http://127.0.0.1/admin"})
 
-async def test_rejects_if_any_dns_answer_is_private():
-    async def mixed(host, port):
-        return ["93.184.216.34", "169.254.169.254"]
-    async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200))) as client:
-        result = await EndpointChecker(client, 1, 1, 100, mixed).check("id", "http://example.com")
-    assert result.error == "non_public_target"
-
-
-async def test_redirect_limit():
-    def handler(request):
-        return httpx.Response(302, headers={"location": "/again"}, request=request)
-    async def public(host, port):
-        return ["93.184.216.34"]
-    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
-        result = await EndpointChecker(client, 1, 1, 100, public).check("id", "https://example.com")
-    assert result.error == "too_many_redirects"
+    checker = EndpointChecker(Settings(), transport=httpx.MockTransport(handler), resolver=public_resolver)
+    with pytest.raises(UnsafeTarget):
+        await checker.check("id", "https://example.com")
+    assert calls == 1