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

Schema auth validation tests to improve coverage.
"""

# Third-Party
from pydantic import SecretStr, ValidationError
import pytest

# First-Party
from mcpgateway.config import settings
from mcpgateway.schemas import A2AAgentCreate, A2AAgentUpdate, AdminCreateUserRequest, EmailRegistrationRequest, GatewayCreate, GatewayUpdate, PublicRegistrationRequest
from mcpgateway.utils.services_auth import decode_auth


def test_gateway_create_authheaders_multi_duplicate(caplog):
    caplog.set_level("WARNING", logger="mcpgateway.schemas")
    gateway = GatewayCreate(
        name="gw",
        url="https://example.com",
        auth_type="authheaders",
        auth_headers=[{"key": "X-Token", "value": "a"}, {"key": "X-Token", "value": "b"}],
    )
    decoded = decode_auth(gateway.auth_value)
    assert decoded["X-Token"] == "b"
    assert any("Duplicate header keys detected" in rec.message for rec in caplog.records)


def test_gateway_create_authheaders_invalid_key():
    with pytest.raises(ValueError):
        GatewayCreate(
            name="gw",
            url="https://example.com",
            auth_type="authheaders",
            auth_headers=[{"key": "X:Bad", "value": "v"}],
        )


def test_gateway_create_authheaders_missing_key():
    with pytest.raises(ValueError):
        GatewayCreate(
            name="gw",
            url="https://example.com",
            auth_type="authheaders",
            auth_headers=[{"value": "v"}],
        )


def test_gateway_create_legacy_header():
    gateway = GatewayCreate(
        name="gw",
        url="https://example.com",
        auth_type="authheaders",
        auth_header_key="X-Api-Key",
        auth_header_value="secret",
    )
    decoded = decode_auth(gateway.auth_value)
    assert decoded["X-Api-Key"] == "secret"


def test_gateway_create_query_param_disabled(monkeypatch):
    monkeypatch.setattr(settings, "insecure_allow_queryparam_auth", False)
    with pytest.raises(ValueError):
        GatewayCreate(
            name="gw",
            url="https://example.com",
            auth_type="query_param",
            auth_query_param_key="api_key",
            auth_query_param_value=SecretStr("secret"),
        )


def test_gateway_create_query_param_host_not_allowed(monkeypatch):
    monkeypatch.setattr(settings, "insecure_allow_queryparam_auth", True)
    monkeypatch.setattr(settings, "insecure_queryparam_auth_allowed_hosts", ["allowed.com"])
    with pytest.raises(ValueError):
        GatewayCreate(
            name="gw",
            url="https://bad.com/path",
            auth_type="query_param",
            auth_query_param_key="api_key",
            auth_query_param_value=SecretStr("secret"),
        )


def test_gateway_create_query_param_valid(monkeypatch):
    monkeypatch.setattr(settings, "insecure_allow_queryparam_auth", True)
    monkeypatch.setattr(settings, "insecure_queryparam_auth_allowed_hosts", [])
    gateway = GatewayCreate(
        name="gw",
        url="https://good.com/path",
        auth_type="query_param",
        auth_query_param_key="api_key",
        auth_query_param_value=SecretStr("secret"),
    )
    assert gateway.auth_query_param_key == "api_key"


def test_gateway_update_query_param_missing_value():
    with pytest.raises(ValueError):
        GatewayUpdate(auth_type="query_param", auth_query_param_key="api_key")


def test_a2a_agent_create_auth_basic():
    agent = A2AAgentCreate(
        name="agent",
        endpoint_url="https://example.com",
        auth_type="basic",
        auth_username="user",
        auth_password="pass",
    )
    decoded = decode_auth(agent.auth_value)
    assert decoded["Authorization"].startswith("Basic ")


def test_a2a_agent_create_bearer_missing_token():
    with pytest.raises(ValueError):
        A2AAgentCreate(
            name="agent",
            endpoint_url="https://example.com",
            auth_type="bearer",
        )


def test_a2a_agent_create_authheaders_invalid_key():
    with pytest.raises(ValueError):
        A2AAgentCreate(
            name="agent",
            endpoint_url="https://example.com",
            auth_type="authheaders",
            auth_headers=[{"key": "Bad:Key", "value": "v"}],
        )


def test_a2a_agent_create_query_param_disabled(monkeypatch):
    monkeypatch.setattr(settings, "insecure_allow_queryparam_auth", False)
    with pytest.raises(ValueError):
        A2AAgentCreate(
            name="agent",
            endpoint_url="https://example.com",
            auth_type="query_param",
            auth_query_param_key="api_key",
            auth_query_param_value=SecretStr("secret"),
        )


def test_a2a_agent_create_query_param_host_allowlist(monkeypatch):
    monkeypatch.setattr(settings, "insecure_allow_queryparam_auth", True)
    monkeypatch.setattr(settings, "insecure_queryparam_auth_allowed_hosts", ["allowed.com"])
    with pytest.raises(ValueError):
        A2AAgentCreate(
            name="agent",
            endpoint_url="https://bad.com",
            auth_type="query_param",
            auth_query_param_key="api_key",
            auth_query_param_value=SecretStr("secret"),
        )


def test_a2a_agent_update_query_param_missing_value():
    with pytest.raises(ValueError):
        A2AAgentUpdate(auth_type="query_param", auth_query_param_key="api_key")


# =========================================================================
# auth_type error message consistency tests (PR #3246)
# Verify error messages reference "authheaders" (not "headers").
# =========================================================================


def test_gateway_create_invalid_auth_type_message():
    with pytest.raises(ValidationError) as exc_info:
        GatewayCreate(name="gw", url="https://example.com", auth_type="bogus")
    assert "authheaders" in str(exc_info.value)


def test_gateway_create_authheaders_empty_headers_message():
    with pytest.raises(ValidationError) as exc_info:
        GatewayCreate(name="gw", url="https://example.com", auth_type="authheaders", auth_headers=[{"key": "", "value": "v"}])
    assert "authheaders" in str(exc_info.value).lower()


def test_gateway_create_authheaders_missing_legacy_fields_message():
    with pytest.raises(ValidationError) as exc_info:
        GatewayCreate(name="gw", url="https://example.com", auth_type="authheaders")
    assert "authheaders" in str(exc_info.value).lower()


def test_gateway_update_invalid_auth_type_message():
    with pytest.raises(ValidationError) as exc_info:
        GatewayUpdate(auth_type="bogus")
    assert "authheaders" in str(exc_info.value)


def test_gateway_update_authheaders_empty_headers_message():
    with pytest.raises(ValidationError) as exc_info:
        GatewayUpdate(auth_type="authheaders", auth_headers=[{"key": "", "value": "v"}])
    assert "authheaders" in str(exc_info.value).lower()


def test_gateway_update_authheaders_missing_legacy_fields_message():
    with pytest.raises(ValidationError) as exc_info:
        GatewayUpdate(auth_type="authheaders")
    assert "authheaders" in str(exc_info.value).lower()


def test_a2a_agent_create_invalid_auth_type_message():
    with pytest.raises(ValidationError) as exc_info:
        A2AAgentCreate(name="agent", endpoint_url="https://example.com", auth_type="bogus")
    assert "authheaders" in str(exc_info.value)


def test_a2a_agent_create_authheaders_missing_legacy_fields_message():
    with pytest.raises(ValidationError) as exc_info:
        A2AAgentCreate(name="agent", endpoint_url="https://example.com", auth_type="authheaders")
    assert "authheaders" in str(exc_info.value).lower()


def test_a2a_agent_update_invalid_auth_type_message():
    with pytest.raises(ValidationError) as exc_info:
        A2AAgentUpdate(auth_type="bogus")
    assert "authheaders" in str(exc_info.value)


def test_a2a_agent_update_authheaders_missing_legacy_fields_message():
    with pytest.raises(ValidationError) as exc_info:
        A2AAgentUpdate(auth_type="authheaders")
    assert "authheaders" in str(exc_info.value).lower()


# =========================================================================
# PublicRegistrationRequest Schema Tests
# =========================================================================


def test_public_registration_request_valid():
    """Test PublicRegistrationRequest with valid data."""
    request = PublicRegistrationRequest(
        email="test@example.com",
        password="SecurePass123!",
        full_name="Test User",
    )
    assert request.email == "test@example.com"
    assert request.password == "SecurePass123!"
    assert request.full_name == "Test User"


def test_public_registration_request_password_required():
    """Test PublicRegistrationRequest requires password (not optional)."""
    with pytest.raises(ValueError):
        PublicRegistrationRequest(
            email="test@example.com",
            full_name="Test User",
        )


def test_public_registration_request_password_too_short():
    """Test PublicRegistrationRequest rejects short password."""
    with pytest.raises(ValueError, match="at least 8 characters"):
        PublicRegistrationRequest(
            email="test@example.com",
            password="Short1!",
            full_name="Test User",
        )


def test_public_registration_request_invalid_email():
    """Test PublicRegistrationRequest rejects invalid email."""
    with pytest.raises(ValueError):
        PublicRegistrationRequest(
            email="not-an-email",
            password="SecurePass123!",
            full_name="Test User",
        )


def test_public_registration_request_rejects_admin_fields():
    """Test PublicRegistrationRequest rejects is_admin/is_active/password_change_required (extra=forbid)."""
    with pytest.raises(ValueError):
        PublicRegistrationRequest(
            email="test@example.com",
            password="SecurePass123!",
            full_name="Test User",
            is_admin=True,
        )
    with pytest.raises(ValueError):
        PublicRegistrationRequest(
            email="test@example.com",
            password="SecurePass123!",
            full_name="Test User",
            is_active=False,
        )
    with pytest.raises(ValueError):
        PublicRegistrationRequest(
            email="test@example.com",
            password="SecurePass123!",
            full_name="Test User",
            password_change_required=True,
        )


# =========================================================================
# AdminCreateUserRequest Schema Tests
# =========================================================================


def test_admin_create_user_request_valid():
    """Test AdminCreateUserRequest with password provided."""
    request = AdminCreateUserRequest(
        email="test@example.com",
        password="SecurePass123!",
        full_name="Test User",
    )
    assert request.email == "test@example.com"
    assert request.password == "SecurePass123!"
    assert request.full_name == "Test User"
    assert request.is_admin is False
    assert request.is_active is True
    assert request.password_change_required is False


def test_admin_create_user_request_password_required():
    """Test AdminCreateUserRequest requires password (not optional)."""
    with pytest.raises(ValueError):
        AdminCreateUserRequest(
            email="test@example.com",
            full_name="Test User",
        )


def test_admin_create_user_request_with_all_fields():
    """Test AdminCreateUserRequest with all fields set."""
    request = AdminCreateUserRequest(
        email="complete@example.com",
        password="CompletePass123!",
        full_name="Complete User",
        is_admin=True,
        is_active=False,
        password_change_required=True,
    )
    assert request.email == "complete@example.com"
    assert request.password == "CompletePass123!"
    assert request.full_name == "Complete User"
    assert request.is_admin is True
    assert request.is_active is False
    assert request.password_change_required is True


def test_admin_create_user_request_password_too_short():
    """Test AdminCreateUserRequest rejects short password."""
    with pytest.raises(ValueError, match="at least 8 characters"):
        AdminCreateUserRequest(
            email="test@example.com",
            password="Short1!",
            full_name="Test User",
        )


def test_admin_create_user_request_invalid_email():
    """Test AdminCreateUserRequest rejects invalid email."""
    with pytest.raises(ValueError):
        AdminCreateUserRequest(
            email="not-an-email",
            password="SecurePass123!",
            full_name="Test User",
        )


def test_admin_create_user_request_with_is_active_false():
    """Test AdminCreateUserRequest with is_active=False."""
    request = AdminCreateUserRequest(
        email="inactive@example.com",
        password="SecurePass123!",
        full_name="Inactive User",
        is_active=False,
    )
    assert request.is_active is False


def test_admin_create_user_request_with_pcr_true():
    """Test AdminCreateUserRequest with password_change_required=True."""
    request = AdminCreateUserRequest(
        email="pwchange@example.com",
        password="TempPass123!",
        full_name="PCR User",
        password_change_required=True,
    )
    assert request.password_change_required is True


def test_email_registration_request_deprecated_alias():
    """Test EmailRegistrationRequest is a deprecated alias for AdminCreateUserRequest."""
    assert EmailRegistrationRequest is AdminCreateUserRequest
