dbt-mcp

"""
Tests for OAuth token models.
"""

import pytest
from pydantic import ValidationError

from dbt_mcp.oauth.token import AccessTokenResponse, DecodedAccessToken


class TestAccessTokenResponse:
    """Test the AccessTokenResponse model."""

    def test_valid_token_response(self):
        """Test creating a valid access token response."""
        token_data = {
            "access_token": "test_access_token",
            "refresh_token": "test_refresh_token",
            "expires_in": 3600,
            "scope": "read write",
            "token_type": "Bearer",
            "expires_at": 1609459200,  # 2021-01-01 00:00:00 UTC
        }

        token_response = AccessTokenResponse(**token_data)

        assert token_response.access_token == "test_access_token"
        assert token_response.refresh_token == "test_refresh_token"
        assert token_response.expires_in == 3600
        assert token_response.scope == "read write"
        assert token_response.token_type == "Bearer"
        assert token_response.expires_at == 1609459200

    def test_missing_required_field(self):
        """Test that missing required fields raise validation errors."""
        incomplete_data = {
            "access_token": "test_access_token",
            "refresh_token": "test_refresh_token",
            # Missing expires_in, scope, token_type, expires_at
        }

        with pytest.raises(ValidationError) as exc_info:
            AccessTokenResponse(**incomplete_data)

        error = exc_info.value
        assert len(error.errors()) >= 4  # At least 4 missing fields

        missing_fields = {
            err["loc"][0] for err in error.errors() if err["type"] == "missing"
        }
        expected_missing = {"expires_in", "scope", "token_type", "expires_at"}
        assert expected_missing.issubset(missing_fields)

    def test_invalid_data_types(self):
        """Test that invalid data types raise validation errors."""
        invalid_data = {
            "access_token": "test_access_token",
            "refresh_token": "test_refresh_token",
            "expires_in": "not_an_int",  # Should be int
            "scope": "read write",
            "token_type": "Bearer",
            "expires_at": "not_an_int",  # Should be int
        }

        with pytest.raises(ValidationError) as exc_info:
            AccessTokenResponse(**invalid_data)

        error = exc_info.value
        # Should have validation errors for expires_in and expires_at
        assert len(error.errors()) >= 2

        invalid_fields = {err["loc"][0] for err in error.errors()}
        assert "expires_in" in invalid_fields
        assert "expires_at" in invalid_fields

    def test_model_dict_conversion(self):
        """Test converting model to dict and back."""
        token_data = {
            "access_token": "test_access_token",
            "refresh_token": "test_refresh_token",
            "expires_in": 3600,
            "scope": "read write",
            "token_type": "Bearer",
            "expires_at": 1609459200,
        }

        token_response = AccessTokenResponse(**token_data)
        token_dict = token_response.model_dump()

        # Should be able to recreate from dict
        recreated_token = AccessTokenResponse(**token_dict)
        assert recreated_token == token_response

    def test_model_json_serialization(self):
        """Test JSON serialization and deserialization."""
        token_data = {
            "access_token": "test_access_token",
            "refresh_token": "test_refresh_token",
            "expires_in": 3600,
            "scope": "read write",
            "token_type": "Bearer",
            "expires_at": 1609459200,
        }

        token_response = AccessTokenResponse(**token_data)
        json_str = token_response.model_dump_json()

        # Should be valid JSON that can be parsed back
        import json

        parsed_data = json.loads(json_str)
        recreated_token = AccessTokenResponse(**parsed_data)
        assert recreated_token == token_response


class TestDecodedAccessToken:
    """Test the DecodedAccessToken model."""

    def test_valid_decoded_token(self):
        """Test creating a valid decoded access token."""
        access_token_response = AccessTokenResponse(
            access_token="test_access_token",
            refresh_token="test_refresh_token",
            expires_in=3600,
            scope="read write",
            token_type="Bearer",
            expires_at=1609459200,
        )

        decoded_claims = {
            "sub": "user123",
            "iss": "https://auth.example.com",
            "exp": 1609459200,
            "iat": 1609455600,
            "scope": "read write",
        }

        decoded_token = DecodedAccessToken(
            access_token_response=access_token_response, decoded_claims=decoded_claims
        )

        assert decoded_token.access_token_response == access_token_response
        assert decoded_token.decoded_claims == decoded_claims
        assert decoded_token.decoded_claims["sub"] == "user123"
        assert decoded_token.decoded_claims["scope"] == "read write"

    def test_empty_decoded_claims(self):
        """Test that empty decoded claims are allowed."""
        access_token_response = AccessTokenResponse(
            access_token="test_access_token",
            refresh_token="test_refresh_token",
            expires_in=3600,
            scope="read write",
            token_type="Bearer",
            expires_at=1609459200,
        )

        decoded_token = DecodedAccessToken(
            access_token_response=access_token_response, decoded_claims={}
        )

        assert decoded_token.access_token_response == access_token_response
        assert decoded_token.decoded_claims == {}

    def test_missing_access_token_response(self):
        """Test that missing access_token_response raises validation error."""
        decoded_claims = {"sub": "user123"}

        with pytest.raises(ValidationError) as exc_info:
            DecodedAccessToken(decoded_claims=decoded_claims)

        error = exc_info.value
        assert len(error.errors()) == 1
        assert error.errors()[0]["loc"] == ("access_token_response",)
        assert error.errors()[0]["type"] == "missing"

    def test_invalid_access_token_response_type(self):
        """Test that invalid access_token_response type raises validation error."""
        with pytest.raises(ValidationError) as exc_info:
            DecodedAccessToken(
                access_token_response="not_a_token_response",  # Should be AccessTokenResponse
                decoded_claims={"sub": "user123"},
            )

        error = exc_info.value
        assert len(error.errors()) >= 1
        # Should have validation error for access_token_response field
        assert any(err["loc"] == ("access_token_response",) for err in error.errors())

    def test_complex_decoded_claims(self):
        """Test with complex nested decoded claims."""
        access_token_response = AccessTokenResponse(
            access_token="test_access_token",
            refresh_token="test_refresh_token",
            expires_in=3600,
            scope="read write",
            token_type="Bearer",
            expires_at=1609459200,
        )

        complex_claims = {
            "sub": "user123",
            "roles": ["admin", "user"],
            "permissions": {"read": ["resource1", "resource2"], "write": ["resource1"]},
            "metadata": {
                "created_at": "2021-01-01T00:00:00Z",
                "last_login": "2021-01-01T12:00:00Z",
            },
        }

        decoded_token = DecodedAccessToken(
            access_token_response=access_token_response, decoded_claims=complex_claims
        )

        assert decoded_token.decoded_claims["roles"] == ["admin", "user"]
        assert decoded_token.decoded_claims["permissions"]["read"] == [
            "resource1",
            "resource2",
        ]
        assert (
            decoded_token.decoded_claims["metadata"]["created_at"]
            == "2021-01-01T00:00:00Z"
        )