test_auth.py

"""Tests for authentication module."""

# ruff: noqa: S105, S106
# S105/S106: Test credentials use dummy values.

from __future__ import annotations

import base64
import datetime
import json
import sys
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from aiohttp import ClientSession

try:
    from botocore.exceptions import ClientError

    HAS_NEXITY_DEPS = True
except ImportError:
    HAS_NEXITY_DEPS = False

from pyoverkiz.auth.base import AuthContext
from pyoverkiz.auth.credentials import (
    LocalTokenCredentials,
    RexelOAuthCodeCredentials,
    TokenCredentials,
    UsernamePasswordCredentials,
)
from pyoverkiz.auth.factory import (
    _ensure_credentials,
    build_auth_strategy,
)
from pyoverkiz.auth.strategies import (
    BearerTokenAuthStrategy,
    CozytouchAuthStrategy,
    LocalTokenAuthStrategy,
    NexityAuthStrategy,
    RexelAuthStrategy,
    SessionLoginStrategy,
    SomfyAuthStrategy,
    _decode_jwt_payload,
)
from pyoverkiz.enums import APIType, Server
from pyoverkiz.exceptions import InvalidTokenError, NexityBadCredentialsError
from pyoverkiz.models import ServerConfig


class TestAuthContext:
    """Test AuthContext functionality."""

    def test_not_expired_no_expiration(self):
        """Test that context without expiration is not expired."""
        context = AuthContext(access_token="test_token")
        assert not context.is_expired()

    def test_not_expired_future_expiration(self):
        """Test that context with future expiration is not expired."""
        future = datetime.datetime.now(datetime.UTC) + datetime.timedelta(hours=1)
        context = AuthContext(access_token="test_token", expires_at=future)
        assert not context.is_expired()

    def test_expired_past_expiration(self):
        """Test that context with past expiration is expired."""
        past = datetime.datetime.now(datetime.UTC) - datetime.timedelta(hours=1)
        context = AuthContext(access_token="test_token", expires_at=past)
        assert context.is_expired()

    def test_expired_with_skew(self):
        """Test that context respects skew time."""
        # Expires in 3 seconds, but default skew is 5
        soon = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=3)
        context = AuthContext(access_token="test_token", expires_at=soon)
        assert context.is_expired()

    def test_not_expired_with_custom_skew(self):
        """Test that custom skew time can be provided."""
        soon = datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=3)
        context = AuthContext(access_token="test_token", expires_at=soon)
        assert not context.is_expired(skew_seconds=1)


class TestCredentials:
    """Test credential dataclasses."""

    def test_username_password_credentials(self):
        """Test UsernamePasswordCredentials creation."""
        creds = UsernamePasswordCredentials("user@example.com", "password123")
        assert creds.username == "user@example.com"
        assert creds.password == "password123"

    def test_token_credentials(self):
        """Test TokenCredentials creation."""
        creds = TokenCredentials("my_token_123")
        assert creds.token == "my_token_123"

    def test_local_token_credentials(self):
        """Test LocalTokenCredentials creation."""
        creds = LocalTokenCredentials("local_token_456")
        assert creds.token == "local_token_456"
        assert isinstance(creds, TokenCredentials)

    def test_rexel_oauth_credentials(self):
        """Test RexelOAuthCodeCredentials creation."""
        creds = RexelOAuthCodeCredentials("auth_code_xyz", "http://redirect.uri")
        assert creds.code == "auth_code_xyz"
        assert creds.redirect_uri == "http://redirect.uri"


class TestAuthFactory:
    """Test authentication factory functions."""

    def test_ensure_credentials_username_password_valid(self):
        """Test that valid username/password credentials pass validation."""
        creds = UsernamePasswordCredentials("user", "pass")
        result = _ensure_credentials(creds, UsernamePasswordCredentials)
        assert result is creds

    def test_ensure_credentials_username_password_invalid(self):
        """Test that invalid credentials raise TypeError."""
        creds = TokenCredentials("token")
        with pytest.raises(TypeError, match="UsernamePasswordCredentials are required"):
            _ensure_credentials(creds, UsernamePasswordCredentials)

    def test_ensure_credentials_token_valid(self):
        """Test that valid token credentials pass validation."""
        creds = TokenCredentials("token")
        result = _ensure_credentials(creds, TokenCredentials)
        assert result is creds

    def test_ensure_credentials_token_local_valid(self):
        """Test that LocalTokenCredentials also pass token validation."""
        creds = LocalTokenCredentials("local_token")
        result = _ensure_credentials(creds, TokenCredentials)
        assert result is creds

    def test_ensure_credentials_token_invalid(self):
        """Test that invalid credentials raise TypeError."""
        creds = UsernamePasswordCredentials("user", "pass")
        with pytest.raises(TypeError, match="TokenCredentials are required"):
            _ensure_credentials(creds, TokenCredentials)

    def test_ensure_credentials_rexel_valid(self):
        """Test that valid Rexel credentials pass validation."""
        creds = RexelOAuthCodeCredentials("code", "uri")
        result = _ensure_credentials(creds, RexelOAuthCodeCredentials)
        assert result is creds

    def test_ensure_credentials_rexel_invalid(self):
        """Test that invalid credentials raise TypeError."""
        creds = UsernamePasswordCredentials("user", "pass")
        with pytest.raises(TypeError, match="RexelOAuthCodeCredentials are required"):
            _ensure_credentials(creds, RexelOAuthCodeCredentials)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_somfy(self):
        """Test building Somfy auth strategy."""
        server_config = ServerConfig(
            server=Server.SOMFY_EUROPE,
            name="Somfy",
            endpoint="https://api.somfy.com",
            manufacturer="Somfy",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, SomfyAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_cozytouch(self):
        """Test building Cozytouch auth strategy."""
        server_config = ServerConfig(
            server=Server.ATLANTIC_COZYTOUCH,
            name="Cozytouch",
            endpoint="https://api.cozytouch.com",
            manufacturer="Atlantic",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, CozytouchAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_nexity(self):
        """Test building Nexity auth strategy."""
        server_config = ServerConfig(
            server=Server.NEXITY,
            name="Nexity",
            endpoint="https://api.nexity.com",
            manufacturer="Nexity",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, NexityAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_rexel(self):
        """Test building Rexel auth strategy."""
        server_config = ServerConfig(
            server=Server.REXEL,
            name="Rexel",
            endpoint="https://api.rexel.com",
            manufacturer="Rexel",
            api_type=APIType.CLOUD,
        )
        credentials = RexelOAuthCodeCredentials("code", "http://redirect.uri")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, RexelAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_local_token(self):
        """Test building local token auth strategy."""
        server_config = ServerConfig(
            server=None,
            name="Local",
            endpoint="https://gateway.local",
            manufacturer="Overkiz",
            api_type=APIType.LOCAL,
        )
        credentials = LocalTokenCredentials("local_token")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, LocalTokenAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_local_bearer(self):
        """Test building local bearer token auth strategy."""
        server_config = ServerConfig(
            server=None,
            name="Local",
            endpoint="https://gateway.local",
            manufacturer="Overkiz",
            api_type=APIType.LOCAL,
        )
        credentials = TokenCredentials("bearer_token")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, BearerTokenAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_cloud_bearer(self):
        """Test building cloud bearer token auth strategy."""
        server_config = ServerConfig(
            server=Server.SOMFY_OCEANIA,
            name="Somfy Oceania",
            endpoint="https://api.somfy.com.au",
            manufacturer="Somfy",
            api_type=APIType.CLOUD,
        )
        credentials = TokenCredentials("bearer_token")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, BearerTokenAuthStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_session_login(self):
        """Test building generic session login auth strategy."""
        server_config = ServerConfig(
            server=Server.SOMFY_OCEANIA,
            name="Somfy Oceania",
            endpoint="https://api.somfy.com.au",
            manufacturer="Somfy",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = build_auth_strategy(
            server_config=server_config,
            credentials=credentials,
            session=session,
            ssl_context=True,
        )

        assert isinstance(strategy, SessionLoginStrategy)

    @pytest.mark.asyncio
    async def test_build_auth_strategy_wrong_credentials_type(self):
        """Test that wrong credentials type raises TypeError."""
        server_config = ServerConfig(
            server=Server.SOMFY_EUROPE,
            name="Somfy",
            endpoint="https://api.somfy.com",
            manufacturer="Somfy",
            api_type=APIType.CLOUD,
        )
        credentials = TokenCredentials("token")  # Wrong type for Somfy
        session = AsyncMock(spec=ClientSession)

        with pytest.raises(TypeError, match="UsernamePasswordCredentials are required"):
            build_auth_strategy(
                server_config=server_config,
                credentials=credentials,
                session=session,
                ssl_context=True,
            )


class TestSessionLoginStrategy:
    """Test SessionLoginStrategy."""

    @pytest.mark.asyncio
    async def test_login_success(self):
        """Test successful login with 200 response."""
        server_config = ServerConfig(
            server=Server.SOMFY_OCEANIA,
            name="Test",
            endpoint="https://api.test.com/",
            manufacturer="Test",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        mock_response = MagicMock()
        mock_response.status = 200
        mock_response.json = AsyncMock(return_value={"success": True})
        mock_response.__aenter__ = AsyncMock(return_value=mock_response)
        mock_response.__aexit__ = AsyncMock(return_value=None)
        session.post = MagicMock(return_value=mock_response)

        strategy = SessionLoginStrategy(credentials, session, server_config, True)
        await strategy.login()

        session.post.assert_called_once()

    @pytest.mark.asyncio
    async def test_login_204_no_content(self):
        """Test login with 204 No Content response."""
        server_config = ServerConfig(
            server=Server.SOMFY_OCEANIA,
            name="Test",
            endpoint="https://api.test.com/",
            manufacturer="Test",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        mock_response = MagicMock()
        mock_response.status = 204
        mock_response.json = AsyncMock()
        mock_response.__aenter__ = AsyncMock(return_value=mock_response)
        mock_response.__aexit__ = AsyncMock(return_value=None)
        session.post = MagicMock(return_value=mock_response)

        strategy = SessionLoginStrategy(credentials, session, server_config, True)
        await strategy.login()

        # Should not call json() for 204 response
        assert not mock_response.json.called

    @pytest.mark.asyncio
    async def test_refresh_if_needed_no_refresh(self):
        """Test that refresh_if_needed returns False when no refresh needed."""
        server_config = ServerConfig(
            server=Server.SOMFY_OCEANIA,
            name="Test",
            endpoint="https://api.test.com/",
            manufacturer="Test",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = SessionLoginStrategy(credentials, session, server_config, True)
        result = await strategy.refresh_if_needed()

        assert not result

    def test_auth_headers_no_token(self):
        """Test that auth headers return empty dict when no token."""
        server_config = ServerConfig(
            server=Server.SOMFY_OCEANIA,
            name="Test",
            endpoint="https://api.test.com/",
            manufacturer="Test",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = SessionLoginStrategy(credentials, session, server_config, True)
        headers = strategy.auth_headers()

        assert headers == {}


class TestBearerTokenAuthStrategy:
    """Test BearerTokenAuthStrategy."""

    @pytest.mark.asyncio
    async def test_login_no_op(self):
        """Test that login is a no-op for bearer tokens."""
        server_config = ServerConfig(
            server=None,
            name="Test",
            endpoint="https://api.test.com/",
            manufacturer="Test",
            api_type=APIType.CLOUD,
        )
        credentials = TokenCredentials("my_bearer_token")
        session = AsyncMock(spec=ClientSession)

        strategy = BearerTokenAuthStrategy(credentials, session, server_config, True)
        result = await strategy.login()

        # Login should be a no-op
        assert result is None

    def test_auth_headers_with_token(self):
        """Test that auth headers include Bearer token."""
        server_config = ServerConfig(
            server=None,
            name="Test",
            endpoint="https://api.test.com/",
            manufacturer="Test",
            api_type=APIType.CLOUD,
        )
        credentials = TokenCredentials("my_bearer_token")
        session = AsyncMock(spec=ClientSession)

        strategy = BearerTokenAuthStrategy(credentials, session, server_config, True)
        headers = strategy.auth_headers()

        assert headers == {"Authorization": "Bearer my_bearer_token"}


class TestNexityAuthStrategy:
    """Tests for Nexity auth error mapping behavior."""

    def test_boto3_not_imported_at_module_load(self):
        """Verify boto3 and warrant_lite are lazy-imported, not at module load."""
        saved = {}
        for mod in ("boto3", "botocore", "warrant_lite"):
            saved[mod] = sys.modules.pop(mod, None)

        try:
            import importlib

            import pyoverkiz.auth.strategies

            importlib.reload(pyoverkiz.auth.strategies)

            assert "boto3" not in sys.modules
            assert "warrant_lite" not in sys.modules
        finally:
            for mod, value in saved.items():
                if value is not None:
                    sys.modules[mod] = value

    @pytest.mark.asyncio
    async def test_login_raises_import_error_without_nexity_extra(self):
        """Login raises ImportError with install hint when nexity extra is missing."""
        server_config = ServerConfig(
            server=Server.NEXITY,
            name="Nexity",
            endpoint="https://api.nexity.com",
            manufacturer="Nexity",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        strategy = NexityAuthStrategy(credentials, session, server_config, True)

        with (
            patch.dict(sys.modules, {"boto3": None}),
            pytest.raises(ImportError, match="pyoverkiz\\[nexity\\]"),
        ):
            await strategy.login()

    @pytest.mark.asyncio
    @pytest.mark.skipif(not HAS_NEXITY_DEPS, reason="nexity extra not installed")
    async def test_login_maps_invalid_credentials_client_error(self):
        """Map Cognito bad-credential errors to NexityBadCredentialsError."""
        server_config = ServerConfig(
            server=Server.NEXITY,
            name="Nexity",
            endpoint="https://api.nexity.com",
            manufacturer="Nexity",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        bad_credentials_error = ClientError(
            error_response={"Error": {"Code": "NotAuthorizedException"}},
            operation_name="InitiateAuth",
        )
        warrant_instance = MagicMock()
        warrant_instance.authenticate_user.side_effect = bad_credentials_error

        with (
            patch("boto3.client", return_value=MagicMock()),
            patch("warrant_lite.WarrantLite", return_value=warrant_instance),
        ):
            strategy = NexityAuthStrategy(credentials, session, server_config, True)
            with pytest.raises(NexityBadCredentialsError):
                await strategy.login()

    @pytest.mark.asyncio
    @pytest.mark.skipif(not HAS_NEXITY_DEPS, reason="nexity extra not installed")
    async def test_login_propagates_non_auth_client_error(self):
        """Propagate non-auth Cognito errors to preserve failure context."""
        server_config = ServerConfig(
            server=Server.NEXITY,
            name="Nexity",
            endpoint="https://api.nexity.com",
            manufacturer="Nexity",
            api_type=APIType.CLOUD,
        )
        credentials = UsernamePasswordCredentials("user", "pass")
        session = AsyncMock(spec=ClientSession)

        service_error = ClientError(
            error_response={"Error": {"Code": "InternalErrorException"}},
            operation_name="InitiateAuth",
        )
        warrant_instance = MagicMock()
        warrant_instance.authenticate_user.side_effect = service_error

        with (
            patch("boto3.client", return_value=MagicMock()),
            patch("warrant_lite.WarrantLite", return_value=warrant_instance),
        ):
            strategy = NexityAuthStrategy(credentials, session, server_config, True)
            with pytest.raises(ClientError, match="InternalErrorException"):
                await strategy.login()


class TestRexelAuthStrategy:
    """Tests for Rexel auth specifics."""

    @pytest.mark.asyncio
    async def test_exchange_token_error_response(self):
        """Ensure OAuth error payloads raise InvalidTokenError before parsing access token."""
        server_config = ServerConfig(
            server=Server.REXEL,
            name="Rexel",
            endpoint="https://api.rexel.com",
            manufacturer="Rexel",
            api_type=APIType.CLOUD,
        )
        credentials = RexelOAuthCodeCredentials("code", "https://redirect")
        session = AsyncMock(spec=ClientSession)

        mock_response = MagicMock()
        mock_response.status = 400
        mock_response.json = AsyncMock(
            return_value={"error": "invalid_grant", "error_description": "bad grant"}
        )
        mock_response.__aenter__ = AsyncMock(return_value=mock_response)
        mock_response.__aexit__ = AsyncMock(return_value=None)
        session.post = MagicMock(return_value=mock_response)

        strategy = RexelAuthStrategy(credentials, session, server_config, True)

        with pytest.raises(InvalidTokenError, match="bad grant"):
            await strategy._exchange_token({"grant_type": "authorization_code"})

    def test_ensure_consent_missing(self):
        """Raising when JWT consent claim is missing or incorrect."""
        payload_segment = (
            base64.urlsafe_b64encode(json.dumps({"consent": "other"}).encode())
            .decode()
            .rstrip("=")
        )
        token = f"header.{payload_segment}.sig"

        with pytest.raises(InvalidTokenError, match="Consent is missing"):
            RexelAuthStrategy._ensure_consent(token)

    def test_decode_jwt_payload_invalid_format(self):
        """Malformed tokens raise InvalidTokenError during decoding."""
        with pytest.raises(InvalidTokenError):
            _decode_jwt_payload("invalid.token")