ea198e8b0d
- 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%
244 lines
7.5 KiB
Python
244 lines
7.5 KiB
Python
"""Tests for UsageStats model (T09).
|
|
|
|
T09: Creare model UsageStats (SQLAlchemy)
|
|
"""
|
|
import pytest
|
|
from datetime import datetime, date
|
|
from decimal import Decimal
|
|
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 TestUsageStatsModelBasics:
|
|
"""Test UsageStats model basic attributes and creation."""
|
|
|
|
def test_usage_stats_model_exists(self):
|
|
"""Test that UsageStats model can be imported."""
|
|
# Assert
|
|
assert UsageStats is not None
|
|
assert hasattr(UsageStats, '__tablename__')
|
|
assert UsageStats.__tablename__ == 'usage_stats'
|
|
|
|
def test_usage_stats_has_required_fields(self):
|
|
"""Test that UsageStats model has all required fields."""
|
|
# Assert
|
|
assert hasattr(UsageStats, 'id')
|
|
assert hasattr(UsageStats, 'api_key_id')
|
|
assert hasattr(UsageStats, 'date')
|
|
assert hasattr(UsageStats, 'model')
|
|
assert hasattr(UsageStats, 'requests_count')
|
|
assert hasattr(UsageStats, 'tokens_input')
|
|
assert hasattr(UsageStats, 'tokens_output')
|
|
assert hasattr(UsageStats, 'cost')
|
|
assert hasattr(UsageStats, 'created_at')
|
|
|
|
def test_usage_stats_create_with_valid_data(self, tmp_path):
|
|
"""Test creating UsageStats with valid data."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_usage.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
# Act
|
|
stats = UsageStats(
|
|
api_key_id=1,
|
|
date=date.today(),
|
|
model="anthropic/claude-3-opus"
|
|
)
|
|
session.add(stats)
|
|
session.flush()
|
|
|
|
# Assert
|
|
assert stats.api_key_id == 1
|
|
assert stats.model == "anthropic/claude-3-opus"
|
|
assert stats.requests_count == 0
|
|
assert stats.tokens_input == 0
|
|
assert stats.tokens_output == 0
|
|
assert stats.cost == 0.0
|
|
|
|
session.close()
|
|
|
|
def test_usage_stats_defaults_are_zero(self, tmp_path):
|
|
"""Test that numeric fields default to zero."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_defaults.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
# Act
|
|
stats = UsageStats(
|
|
api_key_id=1,
|
|
date=date.today(),
|
|
model="gpt-4"
|
|
)
|
|
session.add(stats)
|
|
session.flush()
|
|
|
|
# Assert
|
|
assert stats.requests_count == 0
|
|
assert stats.tokens_input == 0
|
|
assert stats.tokens_output == 0
|
|
assert float(stats.cost) == 0.0
|
|
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUsageStatsConstraints:
|
|
"""Test UsageStats model constraints."""
|
|
|
|
def test_usage_stats_unique_constraint(self):
|
|
"""Test that unique constraint on (api_key_id, date, model) exists."""
|
|
# Assert
|
|
assert hasattr(UsageStats, '__table_args__')
|
|
|
|
def test_usage_stats_api_key_id_index_exists(self):
|
|
"""Test that api_key_id has an index."""
|
|
# Act
|
|
inspector = inspect(UsageStats.__table__)
|
|
indexes = inspector.indexes
|
|
|
|
# Assert
|
|
index_names = [idx.name for idx in indexes]
|
|
assert any('api_key' in name for name in index_names)
|
|
|
|
def test_usage_stats_date_index_exists(self):
|
|
"""Test that date has an index."""
|
|
# Act
|
|
inspector = inspect(UsageStats.__table__)
|
|
indexes = inspector.indexes
|
|
|
|
# Assert
|
|
index_names = [idx.name for idx in indexes]
|
|
assert any('date' in name for name in index_names)
|
|
|
|
def test_usage_stats_model_index_exists(self):
|
|
"""Test that model has an index."""
|
|
# Act
|
|
inspector = inspect(UsageStats.__table__)
|
|
indexes = inspector.indexes
|
|
|
|
# Assert
|
|
index_names = [idx.name for idx in indexes]
|
|
assert any('model' in name for name in index_names)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUsageStatsRelationships:
|
|
"""Test UsageStats model relationships."""
|
|
|
|
def test_usage_stats_has_api_key_relationship(self):
|
|
"""Test that UsageStats has api_key relationship."""
|
|
# Assert
|
|
assert hasattr(UsageStats, 'api_key')
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestUsageStatsDatabaseIntegration:
|
|
"""Integration tests for UsageStats model with database."""
|
|
|
|
def test_usage_stats_persist_and_retrieve(self, tmp_path):
|
|
"""Test persisting and retrieving usage stats from database."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_stats_persist.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
today = date.today()
|
|
|
|
# Act
|
|
stats = UsageStats(
|
|
api_key_id=1,
|
|
date=today,
|
|
model="anthropic/claude-3-opus",
|
|
requests_count=100,
|
|
tokens_input=50000,
|
|
tokens_output=20000,
|
|
cost=Decimal("15.50")
|
|
)
|
|
session.add(stats)
|
|
session.commit()
|
|
|
|
# Retrieve
|
|
retrieved = session.query(UsageStats).first()
|
|
|
|
# Assert
|
|
assert retrieved is not None
|
|
assert retrieved.requests_count == 100
|
|
assert retrieved.tokens_input == 50000
|
|
assert retrieved.tokens_output == 20000
|
|
assert float(retrieved.cost) == 15.50
|
|
|
|
session.close()
|
|
|
|
def test_usage_stats_unique_violation(self, tmp_path):
|
|
"""Test that duplicate (api_key_id, date, model) raises error."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_unique.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
today = date.today()
|
|
|
|
# Create first record
|
|
stats1 = UsageStats(
|
|
api_key_id=1,
|
|
date=today,
|
|
model="gpt-4",
|
|
requests_count=10
|
|
)
|
|
session.add(stats1)
|
|
session.commit()
|
|
|
|
# Try to create duplicate
|
|
stats2 = UsageStats(
|
|
api_key_id=1,
|
|
date=today,
|
|
model="gpt-4",
|
|
requests_count=20
|
|
)
|
|
session.add(stats2)
|
|
|
|
# Assert - Should raise IntegrityError
|
|
with pytest.raises(IntegrityError):
|
|
session.commit()
|
|
|
|
session.close()
|
|
|
|
def test_usage_stats_numeric_precision(self, tmp_path):
|
|
"""Test that cost field stores numeric values correctly."""
|
|
# Arrange
|
|
engine = create_engine(f'sqlite:///{tmp_path}/test_numeric.db')
|
|
Session = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session()
|
|
|
|
# Act
|
|
stats = UsageStats(
|
|
api_key_id=1,
|
|
date=date.today(),
|
|
model="test-model",
|
|
cost=Decimal("123.456789")
|
|
)
|
|
session.add(stats)
|
|
session.commit()
|
|
|
|
# Retrieve
|
|
retrieved = session.query(UsageStats).first()
|
|
|
|
# Assert
|
|
assert retrieved is not None
|
|
assert float(retrieved.cost) > 0
|
|
|
|
session.close()
|