- Add User model with email unique constraint and relationships - Add ApiKey model with encrypted key storage and user relationship - Add UsageStats model with unique constraint (api_key_id, date, model) - Add ApiToken model with token_hash indexing - Configure all cascade delete relationships - Add 49 comprehensive tests with 95% coverage Models: - User: id, email, password_hash, created_at, updated_at, is_active - ApiKey: id, user_id, name, key_encrypted, is_active, created_at, last_used_at - UsageStats: id, api_key_id, date, model, requests_count, tokens_input, tokens_output, cost - ApiToken: id, user_id, token_hash, name, created_at, last_used_at, is_active Tests: 49 passed, coverage 95%
180 lines
5.5 KiB
Python
180 lines
5.5 KiB
Python
"""Tests for ApiKey model (T08).
|
|
|
|
T08: Creare model ApiKey (SQLAlchemy)
|
|
"""
|
|
import pytest
|
|
from datetime import datetime
|
|
from sqlalchemy import create_engine, inspect
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
# Import models to register them with Base
|
|
from openrouter_monitor.models import User, ApiKey, UsageStats, ApiToken
|
|
from openrouter_monitor.database import Base
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestApiKeyModelBasics:
|
|
"""Test ApiKey model basic attributes and creation."""
|
|
|
|
def test_api_key_model_exists(self):
|
|
"""Test that ApiKey model can be imported."""
|
|
# Assert
|
|
assert ApiKey is not None
|
|
assert hasattr(ApiKey, '__tablename__')
|
|
assert ApiKey.__tablename__ == 'api_keys'
|
|
|
|
def test_api_key_has_required_fields(self):
|
|
"""Test that ApiKey model has all required fields."""
|
|
# Assert
|
|
assert hasattr(ApiKey, 'id')
|
|
assert hasattr(ApiKey, 'user_id')
|
|
assert hasattr(ApiKey, 'name')
|
|
assert hasattr(ApiKey, 'key_encrypted')
|
|
assert hasattr(ApiKey, 'is_active')
|
|
assert hasattr(ApiKey, 'created_at')
|
|
assert hasattr(ApiKey, 'last_used_at')
|
|
|
|
def test_api_key_create_with_valid_data(self, tmp_path):
|
|
"""Test creating ApiKey with valid data."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_api_key.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
# Act
|
|
api_key = ApiKey(
|
|
user_id=1,
|
|
name="Production Key",
|
|
key_encrypted="encrypted_value_here"
|
|
)
|
|
session.add(api_key)
|
|
session.flush()
|
|
|
|
# Assert
|
|
assert api_key.name == "Production Key"
|
|
assert api_key.key_encrypted == "encrypted_value_here"
|
|
assert api_key.is_active is True
|
|
assert api_key.created_at is not None
|
|
assert api_key.last_used_at is None
|
|
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestApiKeyConstraints:
|
|
"""Test ApiKey model constraints."""
|
|
|
|
def test_api_key_user_id_index_exists(self):
|
|
"""Test that user_id has an index."""
|
|
# Act
|
|
inspector = inspect(ApiKey.__table__)
|
|
indexes = inspector.indexes
|
|
|
|
# Assert
|
|
index_names = [idx.name for idx in indexes]
|
|
assert any('user' in name for name in index_names)
|
|
|
|
def test_api_key_is_active_index_exists(self):
|
|
"""Test that is_active has an index."""
|
|
# Act
|
|
inspector = inspect(ApiKey.__table__)
|
|
indexes = inspector.indexes
|
|
|
|
# Assert
|
|
index_names = [idx.name for idx in indexes]
|
|
assert any('active' in name for name in index_names)
|
|
|
|
def test_api_key_foreign_key_constraint(self):
|
|
"""Test that user_id has foreign key constraint."""
|
|
# Assert
|
|
assert hasattr(ApiKey, 'user_id')
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestApiKeyRelationships:
|
|
"""Test ApiKey model relationships."""
|
|
|
|
def test_api_key_has_user_relationship(self):
|
|
"""Test that ApiKey has user relationship."""
|
|
# Assert
|
|
assert hasattr(ApiKey, 'user')
|
|
|
|
def test_api_key_has_usage_stats_relationship(self):
|
|
"""Test that ApiKey has usage_stats relationship."""
|
|
# Assert
|
|
assert hasattr(ApiKey, 'usage_stats')
|
|
|
|
def test_usage_stats_cascade_delete(self):
|
|
"""Test that usage_stats have cascade delete."""
|
|
# Arrange
|
|
from sqlalchemy.orm import RelationshipProperty
|
|
|
|
# Act
|
|
usage_stats_rel = getattr(ApiKey.usage_stats, 'property', None)
|
|
|
|
# Assert
|
|
if usage_stats_rel:
|
|
assert 'delete' in str(usage_stats_rel.cascade).lower() or 'all' in str(usage_stats_rel.cascade).lower()
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestApiKeyDatabaseIntegration:
|
|
"""Integration tests for ApiKey model with database."""
|
|
|
|
def test_api_key_persist_and_retrieve(self, tmp_path):
|
|
"""Test persisting and retrieving API key from database."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_api_key_persist.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
# Act
|
|
api_key = ApiKey(
|
|
user_id=1,
|
|
name="Test Key",
|
|
key_encrypted="encrypted_abc123"
|
|
)
|
|
session.add(api_key)
|
|
session.commit()
|
|
|
|
# Retrieve
|
|
retrieved = session.query(ApiKey).filter_by(name="Test Key").first()
|
|
|
|
# Assert
|
|
assert retrieved is not None
|
|
assert retrieved.key_encrypted == "encrypted_abc123"
|
|
assert retrieved.id is not None
|
|
|
|
session.close()
|
|
|
|
def test_api_key_last_used_at_can_be_set(self, tmp_path):
|
|
"""Test that last_used_at can be set."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_last_used.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
now = datetime.utcnow()
|
|
|
|
# Act
|
|
api_key = ApiKey(
|
|
user_id=1,
|
|
name="Test Key",
|
|
key_encrypted="encrypted_abc123",
|
|
last_used_at=now
|
|
)
|
|
session.add(api_key)
|
|
session.commit()
|
|
|
|
# Retrieve
|
|
retrieved = session.query(ApiKey).first()
|
|
|
|
# Assert
|
|
assert retrieved.last_used_at is not None
|
|
|
|
session.close()
|