275 lines
8.1 KiB
Python
275 lines
8.1 KiB
Python
"""
|
|
Token Repository Integration Tests
|
|
|
|
Tests TokenRepository with real database operations.
|
|
"""
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
import pytest
|
|
|
|
from backend.data.repositories.token_repository import TokenRepository
|
|
|
|
|
|
class TestTokenCreate:
|
|
"""Tests for token creation."""
|
|
|
|
def test_create_new_token(self, patched_session):
|
|
"""Test creating a new admin token."""
|
|
repo = TokenRepository()
|
|
|
|
repo.create(
|
|
token="new-test-token-abc123",
|
|
name="New Test Admin",
|
|
)
|
|
|
|
token = repo.get("new-test-token-abc123")
|
|
assert token is not None
|
|
assert token.token == "new-test-token-abc123"
|
|
assert token.name == "New Test Admin"
|
|
assert token.is_active is True
|
|
assert token.expires_at is None
|
|
|
|
def test_create_token_with_expiration(self, patched_session):
|
|
"""Test creating token with expiration date."""
|
|
repo = TokenRepository()
|
|
expiry = datetime.now(timezone.utc) + timedelta(days=30)
|
|
|
|
repo.create(
|
|
token="expiring-token-xyz789",
|
|
name="Expiring Token",
|
|
expires_at=expiry,
|
|
)
|
|
|
|
token = repo.get("expiring-token-xyz789")
|
|
assert token is not None
|
|
assert token.expires_at is not None
|
|
|
|
def test_create_updates_existing_token(self, patched_session, admin_token):
|
|
"""Test creating with existing token updates it."""
|
|
repo = TokenRepository()
|
|
new_expiry = datetime.now(timezone.utc) + timedelta(days=60)
|
|
|
|
repo.create(
|
|
token=admin_token.token,
|
|
name="Updated Admin Name",
|
|
expires_at=new_expiry,
|
|
)
|
|
|
|
token = repo.get(admin_token.token)
|
|
assert token is not None
|
|
assert token.name == "Updated Admin Name"
|
|
assert token.is_active is True
|
|
|
|
|
|
class TestTokenValidation:
|
|
"""Tests for token validation."""
|
|
|
|
def test_is_valid_active_token(self, patched_session, admin_token):
|
|
"""Test that active token is valid."""
|
|
repo = TokenRepository()
|
|
|
|
result = repo.is_valid(admin_token.token)
|
|
|
|
assert result is True
|
|
|
|
def test_is_valid_nonexistent_token(self, patched_session):
|
|
"""Test that nonexistent token is invalid."""
|
|
repo = TokenRepository()
|
|
|
|
result = repo.is_valid("nonexistent-token-12345")
|
|
|
|
assert result is False
|
|
|
|
def test_is_valid_deactivated_token(self, patched_session, admin_token):
|
|
"""Test that deactivated token is invalid."""
|
|
repo = TokenRepository()
|
|
|
|
repo.deactivate(admin_token.token)
|
|
result = repo.is_valid(admin_token.token)
|
|
|
|
assert result is False
|
|
|
|
def test_is_valid_expired_token(self, patched_session):
|
|
"""Test that expired token is invalid."""
|
|
repo = TokenRepository()
|
|
past_expiry = datetime.now(timezone.utc) - timedelta(days=1)
|
|
|
|
repo.create(
|
|
token="expired-token-test",
|
|
name="Expired Token",
|
|
expires_at=past_expiry,
|
|
)
|
|
|
|
result = repo.is_valid("expired-token-test")
|
|
|
|
assert result is False
|
|
|
|
def test_is_valid_not_yet_expired_token(self, patched_session):
|
|
"""Test that not-yet-expired token is valid."""
|
|
repo = TokenRepository()
|
|
future_expiry = datetime.now(timezone.utc) + timedelta(days=7)
|
|
|
|
repo.create(
|
|
token="valid-expiring-token",
|
|
name="Valid Expiring Token",
|
|
expires_at=future_expiry,
|
|
)
|
|
|
|
result = repo.is_valid("valid-expiring-token")
|
|
|
|
assert result is True
|
|
|
|
|
|
class TestTokenGet:
|
|
"""Tests for token retrieval."""
|
|
|
|
def test_get_existing_token(self, patched_session, admin_token):
|
|
"""Test getting an existing token."""
|
|
repo = TokenRepository()
|
|
|
|
token = repo.get(admin_token.token)
|
|
|
|
assert token is not None
|
|
assert token.token == admin_token.token
|
|
assert token.name == admin_token.name
|
|
|
|
def test_get_nonexistent_token(self, patched_session):
|
|
"""Test getting a token that doesn't exist."""
|
|
repo = TokenRepository()
|
|
|
|
token = repo.get("nonexistent-token-xyz")
|
|
|
|
assert token is None
|
|
|
|
|
|
class TestTokenDeactivate:
|
|
"""Tests for token deactivation."""
|
|
|
|
def test_deactivate_existing_token(self, patched_session, admin_token):
|
|
"""Test deactivating an existing token."""
|
|
repo = TokenRepository()
|
|
|
|
result = repo.deactivate(admin_token.token)
|
|
|
|
assert result is True
|
|
token = repo.get(admin_token.token)
|
|
assert token is not None
|
|
assert token.is_active is False
|
|
|
|
def test_deactivate_nonexistent_token(self, patched_session):
|
|
"""Test deactivating a token that doesn't exist."""
|
|
repo = TokenRepository()
|
|
|
|
result = repo.deactivate("nonexistent-token-abc")
|
|
|
|
assert result is False
|
|
|
|
def test_reactivate_deactivated_token(self, patched_session, admin_token):
|
|
"""Test reactivating a deactivated token via create."""
|
|
repo = TokenRepository()
|
|
|
|
# Deactivate first
|
|
repo.deactivate(admin_token.token)
|
|
assert repo.is_valid(admin_token.token) is False
|
|
|
|
# Reactivate via create
|
|
repo.create(
|
|
token=admin_token.token,
|
|
name="Reactivated Admin",
|
|
)
|
|
|
|
assert repo.is_valid(admin_token.token) is True
|
|
|
|
|
|
class TestTokenUsageTracking:
|
|
"""Tests for token usage tracking."""
|
|
|
|
def test_update_usage(self, patched_session, admin_token):
|
|
"""Test updating token last used timestamp."""
|
|
repo = TokenRepository()
|
|
|
|
# Initially last_used_at might be None
|
|
initial_token = repo.get(admin_token.token)
|
|
initial_last_used = initial_token.last_used_at
|
|
|
|
repo.update_usage(admin_token.token)
|
|
|
|
updated_token = repo.get(admin_token.token)
|
|
assert updated_token.last_used_at is not None
|
|
if initial_last_used:
|
|
assert updated_token.last_used_at >= initial_last_used
|
|
|
|
def test_update_usage_nonexistent_token(self, patched_session):
|
|
"""Test updating usage for nonexistent token does nothing."""
|
|
repo = TokenRepository()
|
|
|
|
# Should not raise, just does nothing
|
|
repo.update_usage("nonexistent-token-usage")
|
|
|
|
token = repo.get("nonexistent-token-usage")
|
|
assert token is None
|
|
|
|
|
|
class TestTokenWorkflow:
|
|
"""Tests for complete token workflows."""
|
|
|
|
def test_full_token_lifecycle(self, patched_session):
|
|
"""Test complete token lifecycle: create, validate, use, deactivate."""
|
|
repo = TokenRepository()
|
|
token_str = "lifecycle-test-token"
|
|
|
|
# 1. Create token
|
|
repo.create(token=token_str, name="Lifecycle Token")
|
|
assert repo.is_valid(token_str) is True
|
|
|
|
# 2. Use token
|
|
repo.update_usage(token_str)
|
|
token = repo.get(token_str)
|
|
assert token.last_used_at is not None
|
|
|
|
# 3. Update token info
|
|
new_expiry = datetime.now(timezone.utc) + timedelta(days=90)
|
|
repo.create(
|
|
token=token_str,
|
|
name="Updated Lifecycle Token",
|
|
expires_at=new_expiry,
|
|
)
|
|
token = repo.get(token_str)
|
|
assert token.name == "Updated Lifecycle Token"
|
|
|
|
# 4. Deactivate token
|
|
result = repo.deactivate(token_str)
|
|
assert result is True
|
|
assert repo.is_valid(token_str) is False
|
|
|
|
# 5. Reactivate token
|
|
repo.create(token=token_str, name="Reactivated Token")
|
|
assert repo.is_valid(token_str) is True
|
|
|
|
def test_multiple_tokens(self, patched_session):
|
|
"""Test managing multiple tokens."""
|
|
repo = TokenRepository()
|
|
|
|
# Create multiple tokens
|
|
tokens = [
|
|
("token-a", "Admin A"),
|
|
("token-b", "Admin B"),
|
|
("token-c", "Admin C"),
|
|
]
|
|
|
|
for token_str, name in tokens:
|
|
repo.create(token=token_str, name=name)
|
|
|
|
# Verify all are valid
|
|
for token_str, _ in tokens:
|
|
assert repo.is_valid(token_str) is True
|
|
|
|
# Deactivate one
|
|
repo.deactivate("token-b")
|
|
|
|
# Verify states
|
|
assert repo.is_valid("token-a") is True
|
|
assert repo.is_valid("token-b") is False
|
|
assert repo.is_valid("token-c") is True
|