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

@@ -0,0 +1,63 @@
+import logging
+
+import httpx
+import pytest
+
+from app.checker import EndpointChecker
+from app.config import Settings
+from app.main import create_app
+from app.models import MonitorState
+from app.store import MonitorStore
+
+pytestmark = pytest.mark.anyio
+
+
+async def public_resolver(host: str, port: int) -> list[str]:
+    return ["93.184.216.34"]
+
+
+async def test_check_updates_current_status() -> None:
+    transport = httpx.MockTransport(lambda request: httpx.Response(204))
+    store = MonitorStore()
+    checker = EndpointChecker(Settings(), resolver=public_resolver, transport=transport)
+    app = create_app(Settings(), store, checker)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        item = (await client.post("/monitors", json={"name": "x", "url": "https://example.com"})).json()
+        checked = await client.post(f"/monitors/{item['id']}/check")
+        current = await client.get(f"/monitors/{item['id']}/status")
+    assert checked.json()["state"] == "up"
+    assert current.json()["http_status"] == 204
+
+
+async def test_dns_and_redirect_ssrf_are_blocked() -> None:
+    called = False
+
+    def redirect(request: httpx.Request) -> httpx.Response:
+        nonlocal called
+        called = True
+        return httpx.Response(302, headers={"location": "http://127.0.0.1/admin"})
+
+    checker = EndpointChecker(Settings(), resolver=public_resolver, transport=httpx.MockTransport(redirect))
+    result = await checker.check("one", "https://example.com")
+    assert called and result.state is MonitorState.BLOCKED
+
+    async def private_resolver(host: str, port: int) -> list[str]:
+        return ["10.0.0.4"]
+
+    called = False
+    blocked = EndpointChecker(Settings(), resolver=private_resolver, transport=httpx.MockTransport(redirect))
+    result = await blocked.check("two", "https://internal.example")
+    assert not called and result.state is MonitorState.BLOCKED
+
+
+async def test_transport_errors_and_log_redaction(caplog: pytest.LogCaptureFixture) -> None:
+    def failure(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("secret?token=visible", request=request)
+
+    checker = EndpointChecker(Settings(), resolver=public_resolver, transport=httpx.MockTransport(failure))
+    with caplog.at_level(logging.INFO, logger="monitor.checker"):
+        result = await checker.check("id", "https://example.com/path?token=visible#frag")
+    assert result.state is MonitorState.ERROR
+    text = caplog.text
+    assert "https://example.com/path" in text
+    assert "token=visible" not in text and "frag" not in text