# -*- coding: utf-8 -*-
"""Location: ./tests/unit/mcpgateway/utils/test_passthrough_headers.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

Unit tests for HTTP header passthrough functionality.
This module contains comprehensive unit tests for the passthrough_headers utility
module, covering all scenarios including configuration priorities, conflict
detection, case sensitivity, and security features.
"""

# Standard
import logging
from unittest.mock import Mock, patch

# Third-Party
import pytest

# First-Party
from mcpgateway.cache.global_config_cache import global_config_cache
from mcpgateway.db import Gateway as DbGateway
from mcpgateway.db import GlobalConfig
from mcpgateway.utils.passthrough_headers import get_passthrough_headers


class TestPassthroughHeaders:
    """Test suite for HTTP header passthrough functionality."""

    def setup_method(self):
        """Clear the global config cache before each test to ensure isolation."""
        global_config_cache.invalidate()

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_basic_header_passthrough_global_config(self, mock_settings):
        """Test basic header passthrough with global configuration."""
        mock_settings.enable_header_passthrough = True

        # Mock database and global config
        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Tenant-Id", "X-Trace-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"x-tenant-id": "acme-corp", "x-trace-id": "trace-456", "user-agent": "TestClient/1.0"}  # Not in allowed headers
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        expected = {"Content-Type": "application/json", "X-Tenant-Id": "acme-corp", "X-Trace-Id": "trace-456"}
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_gateway_specific_override(self, mock_settings):
        """Test that gateway-specific headers override global configuration."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Tenant-Id", "X-Trace-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        # Gateway with custom headers
        mock_gateway = Mock(spec=DbGateway)
        mock_gateway.passthrough_headers = ["X-Custom-Header"]
        mock_gateway.auth_type = None

        request_headers = {"x-custom-header": "custom-value", "x-tenant-id": "should-be-ignored", "x-trace-id": "also-ignored"}  # Not in gateway config
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        expected = {"Content-Type": "application/json", "X-Custom-Header": "custom-value"}
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_authorization_conflict_basic_auth(self, mock_settings, caplog):
        """Test that Authorization header is blocked when gateway uses basic auth."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["Authorization", "X-Tenant-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        mock_gateway = Mock(spec=DbGateway)
        mock_gateway.passthrough_headers = ["Authorization", "X-Tenant-Id"]
        mock_gateway.auth_type = "basic"
        mock_gateway.name = "test-gateway"

        request_headers = {"authorization": "Bearer should-be-blocked", "x-tenant-id": "acme-corp"}
        base_headers = {"Content-Type": "application/json"}

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        # Authorization should be blocked, X-Tenant-Id should pass through
        expected = {"Content-Type": "application/json", "X-Tenant-Id": "acme-corp"}
        assert result == expected

        # Check warning was logged
        assert any("Skipping Authorization header passthrough due to basic auth" in record.message for record in caplog.records)

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_authorization_conflict_bearer_auth(self, mock_settings, caplog):
        """Test that Authorization header is blocked when gateway uses bearer auth."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["Authorization"]
        mock_db.query.return_value.first.return_value = mock_global_config

        mock_gateway = Mock(spec=DbGateway)
        mock_gateway.passthrough_headers = None  # Use global
        mock_gateway.auth_type = "bearer"
        mock_gateway.name = "bearer-gateway"

        request_headers = {"authorization": "Bearer should-be-blocked"}
        base_headers = {"Content-Type": "application/json"}

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        # Only base headers should remain
        expected = {"Content-Type": "application/json"}
        assert result == expected

        # Check warning was logged
        assert any("Skipping Authorization header passthrough due to bearer auth" in record.message for record in caplog.records)

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_base_header_conflict_prevention(self, mock_settings, caplog):
        """Test that request headers don't override base headers."""
        mock_settings.enable_header_passthrough = True
        mock_settings.enable_overwrite_base_headers = False

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["Content-Type", "X-Tenant-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"content-type": "text/plain", "x-tenant-id": "acme-corp"}  # Conflicts with base header  # Should pass through
        base_headers = {"Content-Type": "application/json"}

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Base header preserved, tenant ID added
        expected = {"Content-Type": "application/json", "X-Tenant-Id": "acme-corp"}
        assert result == expected

        # Check conflict warning was logged
        assert any("conflicts with pre-defined headers" in record.message for record in caplog.records)

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_case_insensitive_header_matching(self, mock_settings):
        """Test that header matching works with lowercase request headers."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Tenant-ID", "Authorization"]
        mock_db.query.return_value.first.return_value = mock_global_config

        # Request headers are expected to be normalized to lowercase
        request_headers = {"x-tenant-id": "mixed-case-value", "authorization": "bearer lowercase-header"}  # Lowercase key
        base_headers: dict[str, str] = {}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Headers should preserve config case in output keys
        expected = {"X-Tenant-ID": "mixed-case-value", "Authorization": "bearer lowercase-header"}
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_missing_request_headers(self, mock_settings, caplog):
        """Test behavior when configured headers are missing from request."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Missing", "X-Present"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"x-present": "present-value"}
        base_headers = {"Content-Type": "application/json"}

        with caplog.at_level(logging.WARNING):
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Only present header should be included
        expected = {"Content-Type": "application/json", "X-Present": "present-value"}
        assert result == expected

        # Check debug message for missing header
        with caplog.at_level(logging.DEBUG, logger="mcpgateway.utils.passthrough_headers"):
            # Re-run to capture debug messages
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        assert any("X-Missing not found in request headers" in record.message for record in caplog.records)

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_empty_allowed_headers(self, mock_settings):
        """Test behavior with empty allowed headers configuration."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = []
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"x-tenant-id": "should-be-ignored"}
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Only base headers should remain
        expected = {"Content-Type": "application/json"}
        assert result == expected

    def test_none_allowed_headers(self):
        """Test behavior when allowed headers is None."""
        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = None
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"x-tenant-id": "should-be-ignored"}
        base_headers = {"Content-Type": "application/json"}

        # Mock settings fallback
        with patch("mcpgateway.utils.passthrough_headers.settings") as mock_settings:
            mock_settings.default_passthrough_headers = ["X-Default"]

            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Should fall back to settings, but request doesn't have X-Default
        expected = {"Content-Type": "application/json"}
        assert result == expected

    def test_no_global_config_fallback_to_settings(self):
        """Test fallback to settings when no global config exists."""
        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None  # No global config

        request_headers = {"x-default": "default-value"}
        base_headers = {"Content-Type": "application/json"}

        # Mock settings fallback
        with patch("mcpgateway.utils.passthrough_headers.settings") as mock_settings:
            mock_settings.default_passthrough_headers = ["X-Default"]

            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        expected = {"Content-Type": "application/json", "X-Default": "default-value"}
        assert result == expected

    def test_empty_request_headers(self):
        """Test behavior with empty request headers."""
        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Tenant-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers: dict[str, str] = {}
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Only base headers should remain
        expected = {"Content-Type": "application/json"}
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_no_auth_gateway_passes_authorization_when_feature_disabled(self, mock_settings):
        """When gateway.auth_type == 'none', the client's Authorization header
        should be passed through even if ENABLE_HEADER_PASSTHROUGH is False.
        This behavior is handled before the main allowlist processing.
        """
        # Feature disabled globally
        mock_settings.enable_header_passthrough = False

        mock_db = Mock()
        # No global config needed for this early path
        mock_db.query.return_value.first.return_value = None

        request_headers = {"authorization": "Bearer client-token"}
        base_headers: dict[str, str] = {}

        mock_gateway = Mock(spec=DbGateway)
        mock_gateway.passthrough_headers = None
        mock_gateway.auth_type = "none"

        result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        # Authorization should be present because gateway is configured with auth_type 'none'
        assert result.get("Authorization") == "Bearer client-token"

    def test_none_request_headers(self):
        """Test behavior with None request headers."""
        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Tenant-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = None
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)  # type: ignore[arg-type]

        # Only base headers should remain
        expected = {"Content-Type": "application/json"}
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_base_headers_not_modified(self, mock_settings):
        """Test that original base_headers dictionary is not modified."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Tenant-Id"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"x-tenant-id": "acme-corp"}
        base_headers = {"Content-Type": "application/json"}
        original_base = base_headers.copy()

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Original base_headers should not be modified
        assert base_headers == original_base

        # Result should include both base and passthrough headers
        assert "Content-Type" in result
        assert "X-Tenant-Id" in result

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_multiple_auth_type_conflicts(self, mock_settings, caplog):
        """Test various auth type conflict scenarios."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["Authorization"]
        mock_db.query.return_value.first.return_value = mock_global_config

        request_headers = {"authorization": "Bearer token"}
        base_headers: dict[str, str] = {}

        # Test with different auth types. Include the string "none" which should
        # allow passthrough of the client's Authorization header (special-case handled
        # before the main passthrough allowlist logic).
        auth_types = ["basic", "bearer", "api-key", None, "none"]

        for auth_type in auth_types:
            caplog.clear()
            mock_gateway = Mock(spec=DbGateway)
            mock_gateway.passthrough_headers = None
            mock_gateway.auth_type = auth_type
            mock_gateway.name = f"gateway-{auth_type or 'none'}"

            with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
                result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

            if auth_type in ["basic", "bearer"]:
                # Authorization should be blocked
                assert "Authorization" not in result
                assert any("Skipping Authorization header passthrough" in record.message for record in caplog.records)
            else:
                # Authorization should pass through
                assert result.get("Authorization") == "Bearer token"

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_complex_mixed_scenario(self, mock_settings):
        """Test complex scenario with multiple headers, conflicts, and overrides."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["Authorization", "X-Global", "X-Conflict"]
        mock_db.query.return_value.first.return_value = mock_global_config

        mock_gateway = Mock(spec=DbGateway)
        mock_gateway.passthrough_headers = ["X-Gateway", "X-Conflict", "Authorization"]
        mock_gateway.auth_type = "basic"  # Will block Authorization
        mock_gateway.name = "complex-gateway"

        request_headers = {
            "authorization": "Bearer token",  # Blocked by basic auth
            "x-global": "global-value",  # Not in gateway config, ignored
            "x-gateway": "gateway-value",  # Should pass through
            "x-conflict": "conflict-value",  # Should pass through (in both configs)
            "x-random": "random-value",  # Not configured, ignored
        }
        base_headers = {"Content-Type": "application/json", "User-Agent": "MCPGateway/1.0"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        expected = {"Content-Type": "application/json", "User-Agent": "MCPGateway/1.0", "X-Gateway": "gateway-value", "X-Conflict": "conflict-value"}
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_database_query_called_correctly(self, mock_settings):
        """Test that database is queried correctly for GlobalConfig."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = []
        mock_db.query.return_value.first.return_value = mock_global_config

        # Force DB mode regardless of CI env vars (PASSTHROUGH_HEADERS_SOURCE can be "env").
        with patch("mcpgateway.config.settings.passthrough_headers_source", "db"):
            get_passthrough_headers({}, {}, mock_db)

        # Verify database was queried for GlobalConfig
        # Import at assertion-time to avoid stale class identity if mcpgateway.db
        # was reloaded earlier in the session.
        from mcpgateway.db import GlobalConfig as DbGlobalConfig

        mock_db.query.assert_called_once_with(DbGlobalConfig)
        mock_db.query.return_value.first.assert_called_once()

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_logging_levels(self, mock_settings, caplog):
        """Test that appropriate log levels are used for different scenarios."""
        mock_settings.enable_header_passthrough = True
        mock_settings.enable_overwrite_base_headers = False

        mock_db = Mock()
        mock_global_config = Mock(spec=GlobalConfig)
        mock_global_config.passthrough_headers = ["X-Missing", "Authorization", "X-Conflict"]
        mock_db.query.return_value.first.return_value = mock_global_config

        mock_gateway = Mock(spec=DbGateway)
        mock_gateway.passthrough_headers = None
        mock_gateway.auth_type = "basic"
        mock_gateway.name = "test-gateway"

        request_headers = {"authorization": "Bearer token", "x-conflict": "request-value"}  # Will be blocked by basic auth  # Will conflict with base header
        base_headers = {"X-Conflict": "base-value"}  # Conflicts with x-conflict

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        # Should have warnings for: missing header, auth conflict, base header conflict
        warning_messages = [record.message for record in caplog.records if record.levelno == logging.WARNING]

        assert len(warning_messages) == 2  # Only auth conflict and base header conflict
        assert any("due to basic auth" in msg for msg in warning_messages)
        assert any("conflicts with pre-defined headers" in msg for msg in warning_messages)

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_enable_overwrite_base_headers(self, mock_settings):
        """Test that enable_overwrite_base_headers allows overriding base headers."""
        mock_settings.enable_header_passthrough = True
        mock_settings.enable_overwrite_base_headers = True  # Enable override
        mock_settings.default_passthrough_headers = ["Content-Type", "X-Tenant-Id"]

        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None

        request_headers = {"content-type": "text/plain", "x-tenant-id": "acme-corp"}
        base_headers = {"Content-Type": "application/json", "User-Agent": "MCPGateway"}

        with patch("mcpgateway.utils.passthrough_headers.logger") as mock_logger:
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Should override Content-Type and add X-Tenant-Id
        expected = {"Content-Type": "text/plain", "User-Agent": "MCPGateway", "X-Tenant-Id": "acme-corp"}
        assert result == expected

        # Should log debug message about override being enabled
        mock_logger.debug.assert_any_call("Overwriting base headers is enabled via ENABLE_OVERWRITE_BASE_HEADERS flag")

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_disable_overwrite_base_headers_prevents_conflicts(self, mock_settings, caplog):
        """Test that when overwrite is disabled, base header conflicts are prevented."""
        mock_settings.enable_header_passthrough = True
        mock_settings.enable_overwrite_base_headers = False  # Disable override (default)
        mock_settings.default_passthrough_headers = ["Content-Type", "X-Tenant-Id"]

        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None

        request_headers = {"content-type": "text/plain", "x-tenant-id": "acme-corp"}
        base_headers = {"Content-Type": "application/json", "User-Agent": "MCPGateway"}

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Should preserve base Content-Type and add X-Tenant-Id
        expected = {"Content-Type": "application/json", "User-Agent": "MCPGateway", "X-Tenant-Id": "acme-corp"}
        assert result == expected

        # Should log warning about conflict
        assert any("conflicts with pre-defined headers" in record.message for record in caplog.records)
