Spaces:
Sleeping
Sleeping
File size: 3,732 Bytes
db4ba8d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """
Tests for JWT and authentication dependencies
"""
from unittest.mock import MagicMock, patch
import pytest
from src import dependencies
from src.dependencies import KEYCLOAK_JWKS_TTL, get_keycloak_jwks
@pytest.fixture(autouse=True)
def reset_jwks_cache():
dependencies._keycloak_jwks = None
dependencies._keycloak_jwks_time = 0
yield
dependencies._keycloak_jwks = None
dependencies._keycloak_jwks_time = 0
def test_keycloak_jwks_cached_with_ttl():
"""Test that Keycloak JWKS is cached with TTL."""
with patch("src.dependencies.httpx.Client") as mock_client:
# Mock HTTP response
mock_response = MagicMock()
mock_response.json = MagicMock(return_value={"keys": ["test-key-1"]})
mock_client.return_value.__enter__.return_value.get.return_value = mock_response
# First call should fetch from HTTP
jwks1 = get_keycloak_jwks()
assert jwks1 == {"keys": ["test-key-1"]}
# Call count should be 1
assert mock_client.return_value.__enter__.return_value.get.call_count == 1
# Second call within TTL should use cache (no new HTTP call)
jwks2 = get_keycloak_jwks()
assert jwks2 == {"keys": ["test-key-1"]}
assert mock_client.return_value.__enter__.return_value.get.call_count == 1 # Still 1
def test_keycloak_jwks_refreshes_after_ttl():
"""Test that Keycloak JWKS is refreshed after TTL expires."""
with patch("src.dependencies.httpx.Client") as mock_client:
with patch("src.dependencies.time.time") as mock_time:
# Mock HTTP responses
mock_response1 = MagicMock()
mock_response1.json = MagicMock(return_value={"keys": ["old-key"]})
mock_response2 = MagicMock()
mock_response2.json = MagicMock(return_value={"keys": ["new-key"]})
mock_client.return_value.__enter__.return_value.get.side_effect = [mock_response1, mock_response2]
# Mock time progression
mock_time.side_effect = [0, 0, KEYCLOAK_JWKS_TTL + 1] # First call, second call, then TTL expires
# First call
jwks1 = get_keycloak_jwks()
assert jwks1 == {"keys": ["old-key"]}
# Second call (still within TTL)
jwks2 = get_keycloak_jwks()
assert jwks2 == {"keys": ["old-key"]}
# Third call (after TTL expires) - should refresh
jwks3 = get_keycloak_jwks()
assert jwks3 == {"keys": ["new-key"]}
assert mock_client.return_value.__enter__.return_value.get.call_count == 2
def test_keycloak_jwks_handles_http_errors():
"""Test that JWKS fetch handles HTTP errors."""
with patch("src.dependencies.httpx.Client") as mock_client:
# Mock HTTP error
mock_client.return_value.__enter__.return_value.get.side_effect = Exception("Connection refused")
# Should raise the exception
with pytest.raises(Exception, match="Connection refused"):
get_keycloak_jwks()
def test_current_user_immutable():
"""Test that CurrentUser object stores all required fields."""
from src.dependencies import CurrentUser
user = CurrentUser(
sub="user-123",
email="user@example.com",
full_name="Test User",
roles=["operator"],
tier="sme",
company_id="company-123",
raw_token="jwt-token"
)
# Should have all fields
assert user.sub == "user-123"
assert user.email == "user@example.com"
assert user.roles == ["operator"]
assert user.tier == "sme"
assert user.company_id == "company-123"
|