core-api/tests/test_oidc.py

"""Tests for OIDC authentication module."""
import pytest
from unittest.mock import patch, MagicMock
from fastapi import HTTPException

from src.auth.oidc import (
    OIDCConfig,
    oidc_config,
    get_jwks_for_issuer,
    get_current_user,
    _jwks_cache,
)


@pytest.fixture(autouse=True)
def clear_jwks_cache():
    """
    The JWKS cache is module-level state, so a fetch in one test would satisfy
    the next one and hide a regression. Clearing on both sides keeps the tests
    order-independent.
    """
    _jwks_cache.clear()
    yield
    _jwks_cache.clear()


class TestOIDCConfig:
    """Test OIDCConfig class."""

    def test_init_defaults(self):
        """Config should initialize disabled with no issuers or audiences."""
        config = OIDCConfig()

        assert config.enabled is False
        assert config.issuers == []
        assert config.audiences == []

    def test_configure_sets_values(self):
        """configure should set all values."""
        config = OIDCConfig()
        config.configure(
            enabled=True,
            issuers=["https://auth.example.com"],
            audiences=["core-api"],
        )

        assert config.enabled is True
        assert config.issuers == ["https://auth.example.com"]
        assert config.audiences == ["core-api"]

    def test_configure_strips_trailing_slash(self):
        """configure should normalise issuers by dropping the trailing slash."""
        config = OIDCConfig()
        config.configure(
            enabled=True,
            issuers=["https://auth.example.com/"],
            audiences=["core-api"],
        )

        assert config.issuers == ["https://auth.example.com"]

    def test_configure_accepts_multiple_issuers(self):
        """The point of the multi-issuer change: more than one is allowed."""
        config = OIDCConfig()
        config.configure(
            enabled=True,
            issuers=["https://a.example.com/", "https://b.example.com"],
            audiences=["core-api", "other"],
        )

        assert config.issuers == ["https://a.example.com", "https://b.example.com"]
        assert config.audiences == ["core-api", "other"]

    def test_get_jwks_uri_derives_from_issuer(self):
        """The JWKS URI is derived per issuer rather than configured."""
        config = OIDCConfig()

        assert config.get_jwks_uri("https://auth.example.com") == "https://auth.example.com/jwks/"
        assert config.get_jwks_uri("https://auth.example.com/") == "https://auth.example.com/jwks/"

    def test_is_valid_issuer_only_accepts_configured(self):
        """
        An unconfigured issuer must be rejected. This is the security-relevant
        half of multi-issuer support: accepting any issuer would let a token
        from an unrelated identity provider through.
        """
        config = OIDCConfig()
        config.configure(
            enabled=True,
            issuers=["https://auth.example.com"],
            audiences=["core-api"],
        )

        assert config.is_valid_issuer("https://auth.example.com") is True
        assert config.is_valid_issuer("https://auth.example.com/") is True
        assert config.is_valid_issuer("https://evil.example.com") is False


class TestGetJWKSForIssuer:
    """Test get_jwks_for_issuer function."""

    ISSUER = "https://auth.example.com"

    def test_returns_empty_when_disabled(self):
        """Should return an empty dict when OIDC is disabled."""
        original_enabled = oidc_config.enabled
        try:
            oidc_config.enabled = False

            assert get_jwks_for_issuer(self.ISSUER) == {}
        finally:
            oidc_config.enabled = original_enabled

    @patch("src.auth.oidc.httpx.get")
    def test_fetches_jwks_when_enabled(self, mock_get):
        """Should fetch from the issuer's derived JWKS URI."""
        original_enabled = oidc_config.enabled
        try:
            oidc_config.enabled = True

            mock_response = MagicMock()
            mock_response.json.return_value = {"keys": [{"kid": "abc"}]}
            mock_get.return_value = mock_response

            result = get_jwks_for_issuer(self.ISSUER)

            assert result == {"keys": [{"kid": "abc"}]}
            mock_get.assert_called_once()
            assert mock_get.call_args[0][0] == f"{self.ISSUER}/jwks/"
        finally:
            oidc_config.enabled = original_enabled

    @patch("src.auth.oidc.httpx.get")
    def test_caches_per_issuer(self, mock_get):
        """
        A second call for the same issuer must not refetch, and a different
        issuer must. Caching by issuer is the behaviour the multi-issuer change
        introduced, and a shared cache would have served one issuer's keys for
        another — which would be a verification bypass, not just a slow path.
        """
        original_enabled = oidc_config.enabled
        try:
            oidc_config.enabled = True
            mock_response = MagicMock()
            mock_response.json.return_value = {"keys": []}
            mock_get.return_value = mock_response

            get_jwks_for_issuer(self.ISSUER)
            get_jwks_for_issuer(self.ISSUER + "/")   # same issuer, normalised
            assert mock_get.call_count == 1

            get_jwks_for_issuer("https://other.example.com")
            assert mock_get.call_count == 2
        finally:
            oidc_config.enabled = original_enabled

    @patch("src.auth.oidc.httpx.get")
    def test_raises_503_on_fetch_error(self, mock_get):
        """A JWKS fetch failure should surface as 503, not leak the cause."""
        original_enabled = oidc_config.enabled
        try:
            oidc_config.enabled = True
            mock_get.side_effect = Exception("Connection failed")

            with pytest.raises(HTTPException) as exc_info:
                get_jwks_for_issuer(self.ISSUER)

            assert exc_info.value.status_code == 503
        finally:
            oidc_config.enabled = original_enabled


class TestGetCurrentUser:
    """Test get_current_user function."""

    @pytest.mark.asyncio
    async def test_returns_none_when_disabled(self):
        """get_current_user should return None when OIDC disabled."""
        original_enabled = oidc_config.enabled
        try:
            oidc_config.enabled = False

            result = await get_current_user(credentials=None)

            assert result is None
        finally:
            oidc_config.enabled = original_enabled

    @pytest.mark.asyncio
    async def test_raises_401_when_enabled_without_token(self):
        """get_current_user should raise 401 when enabled but no token."""
        original_enabled = oidc_config.enabled
        try:
            oidc_config.enabled = True

            with pytest.raises(HTTPException) as exc_info:
                await get_current_user(credentials=None)

            assert exc_info.value.status_code == 401
        finally:
            oidc_config.enabled = original_enabled


class TestOIDCGlobalConfig:
    """Test global OIDC config."""

    def test_global_config_exists(self):
        """oidc_config should be an OIDCConfig instance."""
        assert isinstance(oidc_config, OIDCConfig)

    def test_global_config_exposes_the_multi_issuer_surface(self):
        """
        Asserts the shape rather than the values, since the live state depends
        on app configuration. These four are what callers depend on.
        """
        assert hasattr(oidc_config, "enabled")
        assert hasattr(oidc_config, "issuers")
        assert hasattr(oidc_config, "audiences")
        assert callable(oidc_config.get_jwks_uri)