From 001c05adec8d19eb85f8f8ecc1e2d5739598ee97 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 16:07:14 +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_logging.py | 33 ++++++++++++++++++++------------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/tests/test_logging.py b/tests/test_logging.py index 0360a57..672dedc 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -1,19 +1,26 @@ -import json import logging -from app.logging_config import JsonFormatter, redact_url +import httpx +import pytest + +from app.checker import EndpointChecker +from app.config import Settings -def test_redact_url_removes_credentials_query_and_fragment(): - assert redact_url("https://user:pass@example.com:8443/a?token=secret#x") == "https://example.com:8443/a" +async def public_resolver(host: str, port: int) -> list[str]: + return ["93.184.216.34"] -def test_json_formatter_never_emits_url_secrets(): - record = logging.LogRecord("test", logging.INFO, "", 0, "event", (), None) - record.url = "https://user:pass@example.com/a?token=secret#fragment" - record.monitor_id = "123" - output = JsonFormatter().format(record) - parsed = json.loads(output) - assert parsed["url"] == "https://example.com/a" - assert "secret" not in output - assert "pass" not in output +@pytest.mark.anyio +async def test_logs_redact_credentials_query_and_fragment(caplog: pytest.LogCaptureFixture) -> None: + checker = EndpointChecker( + Settings(), + transport=httpx.MockTransport(lambda request: httpx.Response(200)), + resolver=public_resolver, + ) + with caplog.at_level(logging.INFO, logger="endpoint_monitor.checker"): + await checker.check("monitor-1", "https://example.com/path?token=topsecret#fragment") + record = next(record for record in caplog.records if getattr(record, "event", None) == "check_completed") + assert record.url == "https://example.com/path" + assert "topsecret" not in record.getMessage() + assert "topsecret" not in record.url