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_security.py

@@ -0,0 +1,43 @@
+import io
+import json
+import logging
+
+import httpx
+
+from app.checker import EndpointChecker
+from app.logging import JsonFormatter
+from app.models import Monitor, MonitorCreate
+from app.security import redacted_url
+
+
+async def test_dns_resolution_blocks_private_address_before_transport() -> None:
+    calls = 0
+
+    async def private_resolver(_: str, __: int) -> list[str]:
+        return ["10.0.0.7"]
+
+    async def transport_handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(200)
+
+    monitor = Monitor(**MonitorCreate(name="private", url="https://internal.example").model_dump())
+    checker = EndpointChecker(
+        1, 2, resolver=private_resolver, transport=httpx.MockTransport(transport_handler)
+    )
+    result = await checker.check(monitor)
+    assert result.state == "blocked"
+    assert calls == 0
+
+
+def test_url_and_json_logs_redact_query_and_credentials() -> None:
+    assert redacted_url("https://user:pass@example.com/a?token=secret#x") == "https://example.com/a"
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    handler.setFormatter(JsonFormatter())
+    record = logging.LogRecord("test", logging.INFO, "", 0, "check", (), None)
+    record.event_data = {"url": redacted_url("https://example.com/a?token=secret")}
+    handler.emit(record)
+    payload = json.loads(stream.getvalue())
+    assert payload["url"] == "https://example.com/a"
+    assert "secret" not in stream.getvalue()