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

@@ -1,109 +1,52 @@
-import json
-import logging
-
 import httpx
 import pytest
 
-from app.checker import EndpointChecker, UnsafeTarget
-from app.config import Settings
-from app.logging import StructuredJsonFormatter, redact_url
+from app.checker import EndpointChecker
+
+pytestmark = pytest.mark.anyio
 
 
-async def public_resolver(host: str, port: int) -> list[str]:
-    return ["93.184.216.34"]
-
-
-def checker(handler, resolver=public_resolver, **settings):
-    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
-    return EndpointChecker(client, Settings(**settings), resolver), client
-
-
-async def test_http_status_and_latency():
-    service, client = checker(lambda request: httpx.Response(503, request=request))
-    try:
-        result = await service.check("https://example.com/path")
-    finally:
-        await client.aclose()
-    assert result.state == "down"
-    assert result.status_code == 503
-    assert result.latency_ms is not None and result.latency_ms >= 0
-
-
-async def test_timeout_is_sanitized_error_result():
-    def timeout(request):
-        raise httpx.ReadTimeout("secret low-level detail", request=request)
-    service, client = checker(timeout)
-    try:
-        result = await service.check("https://example.com/?token=secret")
-    finally:
-        await client.aclose()
-    assert result.state == "error"
-    assert result.error == "request timed out"
-    assert "secret" not in result.error
-
-
-async def test_dns_blocks_private_answer_before_outbound_request():
-    calls = 0
+async def test_blocks_private_dns_before_http_call():
+    called = False
     def handler(request):
-        nonlocal calls
-        calls += 1
+        nonlocal called
+        called = True
         return httpx.Response(200, request=request)
-    async def private_resolver(host: str, port: int) -> list[str]:
-        return ["10.0.0.8"]
-    service, client = checker(handler, private_resolver)
-    try:
-        with pytest.raises(UnsafeTarget, match="non-public"):
-            await service.check("http://internal.invalid/")
-    finally:
-        await client.aclose()
-    assert calls == 0
+    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
 
 
-async def test_mixed_dns_answers_are_blocked():
-    async def mixed(host: str, port: int) -> list[str]:
-        return ["93.184.216.34", "127.0.0.1"]
-    service, client = checker(lambda request: httpx.Response(200, request=request), mixed)
-    try:
-        with pytest.raises(UnsafeTarget):
-            await service.check("https://example.com")
-    finally:
-        await client.aclose()
-
-
-async def test_redirect_destination_is_resolved_and_blocked():
+async def test_blocks_redirect_to_private_target():
     calls = []
     def handler(request):
         calls.append(str(request.url))
-        return httpx.Response(302, headers={"Location": "http://127.0.0.1/admin"}, request=request)
-    async def resolver(host: str, port: int) -> list[str]:
-        return ["127.0.0.1"] if host == "127.0.0.1" else ["93.184.216.34"]
-    service, client = checker(handler, resolver)
-    try:
-        with pytest.raises(UnsafeTarget):
-            await service.check("https://example.com/start")
-    finally:
-        await client.aclose()
-    assert calls == ["https://example.com/start"]
+        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"]
 
 
-async def test_redirect_limit_is_error():
+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)
-    service, client = checker(handler, max_redirects=1)
-    try:
-        result = await service.check("https://example.com/start")
-    finally:
-        await client.aclose()
-    assert result.state == "error"
-    assert result.error == "redirect limit exceeded"
-
-
-def test_url_redaction_and_structured_log_are_inspectable():
-    safe = redact_url("https://user:password@example.com/path?token=secret#private")
-    assert "user" not in safe and "password" not in safe and "secret" not in safe
-    record = logging.LogRecord("checker", logging.INFO, __file__, 1,
-                               "done", (), None)
-    record.target = safe
-    payload = json.loads(StructuredJsonFormatter().format(record))
-    assert payload["target"] == "https://example.com/path?<redacted>#<redacted>"
-    assert payload["message"] == "done"
+        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"