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,71 +1,68 @@
 import logging
+import socket
+from unittest.mock import AsyncMock, patch
 
 import httpx
 import pytest
 
-from endpoint_monitor.checker import EndpointChecker, TargetBlockedError, validate_target
-from endpoint_monitor.logging import JsonFormatter, redact_url
+from app.checker import check_url
+from app.config import Settings
+from app.security import UnsafeDestination, validate_destination
+
+PUBLIC_DNS = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))]
 
 
-async def public_resolver(host: str, port: int) -> list[str]:
-    return ["93.184.216.34"]
+@pytest.mark.asyncio
+async def test_success_and_redirect_targets_are_each_validated():
+    transport = httpx.MockTransport(lambda request: (
+        httpx.Response(302, headers={"location": "https://target.example/final"})
+        if request.url.host == "start.example" else httpx.Response(200)
+    ))
+    validator = AsyncMock()
+    with patch("app.checker.validate_destination", validator):
+        result = await check_url("https://start.example/a?token=x", 1, Settings(), transport)
+    assert result.state == "up"
+    assert validator.await_count == 2
+    assert validator.await_args_list[1].args[0] == "https://target.example/final"
 
 
-@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1", "fc00::1"])
-async def test_dns_answers_block_non_public(address):
-    async def resolver(host: str, port: int) -> list[str]:
-        return [address]
-
-    with pytest.raises(TargetBlockedError, match="non-public"):
-        await validate_target("http://attacker.example/path", resolver)
+@pytest.mark.asyncio
+async def test_private_dns_answer_is_blocked():
+    loop = __import__("asyncio").get_running_loop()
+    with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=[
+        (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80))
+    ])):
+        with pytest.raises(UnsafeDestination, match="non-public"):
+            await validate_destination("http://internal.example")
 
 
-async def test_any_unsafe_dns_answer_blocks_target():
-    async def resolver(host: str, port: int) -> list[str]:
-        return ["93.184.216.34", "127.0.0.1"]
-
-    with pytest.raises(TargetBlockedError):
-        await validate_target("https://mixed.example", resolver)
+@pytest.mark.asyncio
+async def test_redirect_to_private_literal_is_blocked():
+    transport = httpx.MockTransport(lambda request: httpx.Response(
+        302, headers={"location": "http://169.254.169.254/latest"}
+    ))
+    loop = __import__("asyncio").get_running_loop()
+    with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=PUBLIC_DNS)):
+        with pytest.raises(UnsafeDestination):
+            await check_url("https://example.com", 1, Settings(), transport)
 
 
-async def test_redirect_hop_is_resolved_and_blocked():
-    seen = []
-
-    async def resolver(host: str, port: int) -> list[str]:
-        seen.append(host)
-        return ["127.0.0.1"] if host == "internal.example" else ["93.184.216.34"]
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        return httpx.Response(302, headers={"location": "http://internal.example/admin"})
-
-    checker = EndpointChecker(1, 3, resolver, httpx.MockTransport(handler))
-    with pytest.raises(TargetBlockedError):
-        await checker.check("https://public.example/start", "id")
-    assert seen == ["public.example", "internal.example"]
+@pytest.mark.asyncio
+async def test_transport_error_maps_to_error_state():
+    def fail(request):
+        raise httpx.ConnectError("no", request=request)
+    transport = httpx.MockTransport(fail)
+    with patch("app.checker.validate_destination", new=AsyncMock()):
+        result = await check_url("https://example.com", 1, Settings(), transport)
+    assert result.state == "error"
+    assert result.error == "ConnectError"
 
 
-async def test_success_latency_status_and_redaction(caplog):
-    def handler(request: httpx.Request) -> httpx.Response:
-        return httpx.Response(503)
-
-    checker = EndpointChecker(1, 1, public_resolver, httpx.MockTransport(handler))
-    with caplog.at_level(logging.INFO):
-        result = await checker.check("https://example.com/path?token=supersecret#fragment", "abc")
-    assert result.status.value == "down"
-    assert result.http_status == 503
-    assert result.latency_ms >= 0
-    assert result.final_url == "https://example.com/path"
-    assert "supersecret" not in caplog.text
-    assert "fragment" not in caplog.text
-
-
-def test_redact_url_removes_credentials_query_fragment():
-    assert redact_url("https://user:pass@example.com:8443/a?q=secret#x") == "https://example.com:8443/a"
-
-
-def test_json_formatter_is_structured_and_does_not_add_message_secrets():
-    record = logging.LogRecord("x", logging.INFO, __file__, 1, "event", (), None)
-    record.url = redact_url("https://example.com/?api_key=secret")
-    rendered = JsonFormatter().format(record)
-    assert '"event":"event"' in rendered
-    assert "secret" not in rendered
+@pytest.mark.asyncio
+async def test_logs_redact_query(caplog):
+    transport = httpx.MockTransport(lambda request: httpx.Response(200))
+    with patch("app.checker.validate_destination", new=AsyncMock()), caplog.at_level(logging.INFO):
+        await check_url("https://example.com/a?api_key=supersecret", 1, Settings(), transport)
+    text = caplog.text
+    assert "supersecret" not in text
+    assert "REDACTED" in text