200 lines
8.4 KiB
Python
200 lines
8.4 KiB
Python
"""
|
|
Tests for TokenRepository
|
|
|
|
TDD tests for admin token management.
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from inference.data.admin_models import AdminToken
|
|
from inference.data.repositories.token_repository import TokenRepository
|
|
|
|
|
|
class TestTokenRepository:
|
|
"""Tests for TokenRepository."""
|
|
|
|
@pytest.fixture
|
|
def sample_token(self) -> AdminToken:
|
|
"""Create a sample token for testing."""
|
|
return AdminToken(
|
|
token="test-token-123",
|
|
name="Test Token",
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc),
|
|
last_used_at=None,
|
|
expires_at=None,
|
|
)
|
|
|
|
@pytest.fixture
|
|
def expired_token(self) -> AdminToken:
|
|
"""Create an expired token."""
|
|
return AdminToken(
|
|
token="expired-token",
|
|
name="Expired Token",
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=30),
|
|
expires_at=datetime.now(timezone.utc) - timedelta(days=1),
|
|
)
|
|
|
|
@pytest.fixture
|
|
def inactive_token(self) -> AdminToken:
|
|
"""Create an inactive token."""
|
|
return AdminToken(
|
|
token="inactive-token",
|
|
name="Inactive Token",
|
|
is_active=False,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
@pytest.fixture
|
|
def repo(self) -> TokenRepository:
|
|
"""Create a TokenRepository instance."""
|
|
return TokenRepository()
|
|
|
|
def test_is_valid_returns_true_for_active_token(self, repo, sample_token):
|
|
"""Test is_valid returns True for an active, non-expired token."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = sample_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.is_valid("test-token-123")
|
|
|
|
assert result is True
|
|
mock_session.get.assert_called_once_with(AdminToken, "test-token-123")
|
|
|
|
def test_is_valid_returns_false_for_nonexistent_token(self, repo):
|
|
"""Test is_valid returns False for a non-existent token."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = None
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.is_valid("nonexistent-token")
|
|
|
|
assert result is False
|
|
|
|
def test_is_valid_returns_false_for_inactive_token(self, repo, inactive_token):
|
|
"""Test is_valid returns False for an inactive token."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = inactive_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.is_valid("inactive-token")
|
|
|
|
assert result is False
|
|
|
|
def test_is_valid_returns_false_for_expired_token(self, repo, expired_token):
|
|
"""Test is_valid returns False for an expired token."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = expired_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.is_valid("expired-token")
|
|
|
|
assert result is False
|
|
|
|
def test_get_returns_token_when_exists(self, repo, sample_token):
|
|
"""Test get returns token when it exists."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = sample_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.get("test-token-123")
|
|
|
|
assert result is not None
|
|
assert result.token == "test-token-123"
|
|
assert result.name == "Test Token"
|
|
mock_session.expunge.assert_called_once_with(sample_token)
|
|
|
|
def test_get_returns_none_when_not_exists(self, repo):
|
|
"""Test get returns None when token doesn't exist."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = None
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.get("nonexistent-token")
|
|
|
|
assert result is None
|
|
|
|
def test_create_new_token(self, repo):
|
|
"""Test creating a new token."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = None # Token doesn't exist
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
repo.create("new-token", "New Token", expires_at=None)
|
|
|
|
mock_session.add.assert_called_once()
|
|
added_token = mock_session.add.call_args[0][0]
|
|
assert isinstance(added_token, AdminToken)
|
|
assert added_token.token == "new-token"
|
|
assert added_token.name == "New Token"
|
|
|
|
def test_create_updates_existing_token(self, repo, sample_token):
|
|
"""Test create updates an existing token."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = sample_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
repo.create("test-token-123", "Updated Name", expires_at=None)
|
|
|
|
mock_session.add.assert_called_once_with(sample_token)
|
|
assert sample_token.name == "Updated Name"
|
|
assert sample_token.is_active is True
|
|
|
|
def test_update_usage(self, repo, sample_token):
|
|
"""Test updating token last_used_at timestamp."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = sample_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
repo.update_usage("test-token-123")
|
|
|
|
assert sample_token.last_used_at is not None
|
|
mock_session.add.assert_called_once_with(sample_token)
|
|
|
|
def test_deactivate_returns_true_when_token_exists(self, repo, sample_token):
|
|
"""Test deactivate returns True when token exists."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = sample_token
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.deactivate("test-token-123")
|
|
|
|
assert result is True
|
|
assert sample_token.is_active is False
|
|
mock_session.add.assert_called_once_with(sample_token)
|
|
|
|
def test_deactivate_returns_false_when_token_not_exists(self, repo):
|
|
"""Test deactivate returns False when token doesn't exist."""
|
|
with patch("inference.data.repositories.base.get_session_context") as mock_ctx:
|
|
mock_session = MagicMock()
|
|
mock_session.get.return_value = None
|
|
mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
result = repo.deactivate("nonexistent-token")
|
|
|
|
assert result is False
|