From bc3c6cca8bb983e17da955087f8eaf24a126d62c Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 16:07:13 +0000 Subject: [PATCH] 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 | 102 ++++++++++++++++++++++++++---------------- 1 file changed, 63 insertions(+), 39 deletions(-) diff --git a/tests/test_checker.py b/tests/test_checker.py index 3d7b09a..a97c0c2 100644 --- a/tests/test_checker.py +++ b/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