import tempfile
from datetime import UTC, datetime, timedelta
import pytest
from core.oauth.schemas import AuthorizationCode, RefreshToken
from core.oauth.storage import JSONStorage
@pytest.fixture
def storage():
"""Create temporary storage for tests"""
with tempfile.TemporaryDirectory() as tmpdir:
yield JSONStorage(tmpdir)
def test_save_and_get_authorization_code(storage):
"""Test authorization code storage"""
code = AuthorizationCode(
code="test_code_123",
client_id="client_1",
redirect_uri="http://localhost:3000/callback",
scope="read write",
code_challenge="challenge_123",
expires_at=datetime.now(UTC) + timedelta(minutes=5),
)
# Save
assert storage.save_authorization_code(code)
# Get
retrieved = storage.get_authorization_code("test_code_123")
assert retrieved is not None
assert retrieved.code == "test_code_123"
assert retrieved.client_id == "client_1"
def test_expired_code_auto_cleanup(storage):
"""Test expired code auto-cleanup"""
code = AuthorizationCode(
code="expired_code",
client_id="client_1",
redirect_uri="http://localhost:3000/callback",
scope="read",
code_challenge="challenge",
expires_at=datetime.now(UTC) - timedelta(seconds=1), # Expired
)
storage.save_authorization_code(code)
# Should return None and cleanup
retrieved = storage.get_authorization_code("expired_code")
assert retrieved is None
def test_refresh_token_revocation(storage):
"""Test refresh token revocation"""
token = RefreshToken(
token="refresh_token_123",
client_id="client_1",
expires_at=datetime.now(UTC) + timedelta(days=7),
)
storage.save_refresh_token(token)
# Revoke
assert storage.revoke_refresh_token("refresh_token_123")
# Should return None
retrieved = storage.get_refresh_token("refresh_token_123")
assert retrieved is None
