leocrm/tests/test_observability.py

"""Tests for observability & trace_id correlation (B.14).

Covers:
- trace_id is generated per request and present in response header
- trace_id is set in structlog contextvars
- Sensitive fields are not logged (sanitize_dict)
"""

from __future__ import annotations

import pytest
import structlog

from app.core.monitoring import sanitize_dict, _sanitize_sensitive_fields, _SENSITIVE_FIELDS


class TestTraceIdPropagation:
    """Tests for trace_id generation and propagation."""

    @pytest.mark.asyncio
    async def test_trace_id_in_response_header(self, client):
        """Every response should have X-Trace-Id header."""
        response = await client.get("/api/v1/health")
        assert response.status_code == 200
        assert "x-trace-id" in response.headers
        trace_id = response.headers["x-trace-id"]
        assert len(trace_id) == 8
        # Should be hex characters
        int(trace_id, 16)  # Raises ValueError if not hex

    @pytest.mark.asyncio
    async def test_trace_id_unique_per_request(self, client):
        """Each request should get a unique trace_id."""
        r1 = await client.get("/api/v1/health")
        r2 = await client.get("/api/v1/health")
        t1 = r1.headers.get("x-trace-id")
        t2 = r2.headers.get("x-trace-id")
        assert t1 is not None
        assert t2 is not None
        assert t1 != t2

    @pytest.mark.asyncio
    async def test_trace_id_in_structlog_contextvars(self, client):
        """trace_id should be set in structlog contextvars during request processing."""
        # The middleware sets trace_id in contextvars during request handling.
        # We verify this by checking that the response has the header (which
        # is set from the same trace_id variable that was bound to contextvars).
        response = await client.get("/api/v1/health")
        assert "x-trace-id" in response.headers
        # After request completes, contextvars should be cleared
        ctx = structlog.contextvars.get_contextvars()
        assert "trace_id" not in ctx or ctx.get("trace_id") is None

    @pytest.mark.asyncio
    async def test_trace_id_consistent_header_and_body(self, client):
        """trace_id in error response body should match X-Trace-Id header."""
        response = await client.get("/api/v1/contacts/00000000-0000-0000-0000-000000000000")
        if response.status_code >= 400:
            body = response.json()
            header_trace = response.headers.get("x-trace-id")
            body_trace = body.get("trace_id")
            # Both should be present and match
            if header_trace and body_trace:
                assert header_trace == body_trace


class TestSensitiveFieldSanitization:
    """Tests for sensitive field redaction in logs."""

    def test_sanitize_dict_redacts_password(self):
        data = {"username": "admin", "password": "secret123"}
        result = sanitize_dict(data)
        assert result["username"] == "admin"
        assert result["password"] == "***REDACTED***"

    def test_sanitize_dict_redacts_api_key(self):
        data = {"api_key": "sk-abc123", "model": "gpt-4"}
        result = sanitize_dict(data)
        assert result["api_key"] == "***REDACTED***"
        assert result["model"] == "gpt-4"

    def test_sanitize_dict_redacts_token(self):
        data = {"token": "bearer xyz", "user_id": "123"}
        result = sanitize_dict(data)
        assert result["token"] == "***REDACTED***"
        assert result["user_id"] == "123"

    def test_sanitize_dict_redacts_nested(self):
        data = {"config": {"password": "secret", "host": "localhost"}}
        result = sanitize_dict(data)
        assert result["config"]["password"] == "***REDACTED***"
        assert result["config"]["host"] == "localhost"

    def test_sanitize_dict_extra_sensitive(self):
        data = {"custom_secret": "value", "name": "test"}
        result = sanitize_dict(data, extra_sensitive={"custom_secret"})
        assert result["custom_secret"] == "***REDACTED***"
        assert result["name"] == "test"

    def test_sanitize_dict_case_insensitive(self):
        data = {"Password": "secret", "API_KEY": "key123"}
        result = sanitize_dict(data)
        assert result["Password"] == "***REDACTED***"
        assert result["API_KEY"] == "***REDACTED***"

    def test_sanitize_processor_redacts(self):
        """The structlog processor should redact sensitive fields."""
        event_dict = {"password": "secret", "event": "login", "user": "admin"}
        result = _sanitize_sensitive_fields(None, "info", event_dict)
        assert result["password"] == "***REDACTED***"
        assert result["event"] == "login"
        assert result["user"] == "admin"

    def test_sensitive_fields_include_common_names(self):
        """Common sensitive field names should be in the redaction set."""
        expected = {"password", "api_key", "token", "secret", "authorization"}
        assert expected.issubset(_SENSITIVE_FIELDS)

    def test_sanitize_dict_preserves_non_sensitive(self):
        data = {"method": "GET", "path": "/api/v1/health", "status": 200}
        result = sanitize_dict(data)
        assert result == data