diff --git a/export/progress.md b/export/progress.md index a0f6d74..372b5a7 100644 --- a/export/progress.md +++ b/export/progress.md @@ -44,13 +44,13 @@ - [x] T04: Setup file configurazione (.env, config.py) (2024-04-07) - [x] T05: Configurare pytest e struttura test (2024-04-07) -### 🗄️ Database & Models (T06-T11) - 1/6 completati +### 🗄️ Database & Models (T06-T11) - 5/6 completati - [x] T06: Creare database.py (connection & session) - ✅ Completato (2026-04-07 11:00) -- [ ] T07: Creare model User (SQLAlchemy) - 🟡 In progress -- [ ] T08: Creare model ApiKey (SQLAlchemy) -- [ ] T09: Creare model UsageStats (SQLAlchemy) -- [ ] T10: Creare model ApiToken (SQLAlchemy) -- [ ] T11: Setup Alembic e creare migrazione iniziale +- [x] T07: Creare model User (SQLAlchemy) - ✅ Completato (2026-04-07 11:15) +- [x] T08: Creare model ApiKey (SQLAlchemy) - ✅ Completato (2026-04-07 11:15) +- [x] T09: Creare model UsageStats (SQLAlchemy) - ✅ Completato (2026-04-07 11:15) +- [x] T10: Creare model ApiToken (SQLAlchemy) - ✅ Completato (2026-04-07 11:15) +- [ ] T11: Setup Alembic e creare migrazione iniziale - 🟡 In progress ### 🔐 Servizi di Sicurezza (T12-T16) - 0/5 completati - [ ] T12: Implementare EncryptionService (AES-256) diff --git a/src/openrouter_monitor/models/__init__.py b/src/openrouter_monitor/models/__init__.py index e69de29..472ceb8 100644 --- a/src/openrouter_monitor/models/__init__.py +++ b/src/openrouter_monitor/models/__init__.py @@ -0,0 +1,10 @@ +"""Models package for OpenRouter API Key Monitor. + +This package contains all SQLAlchemy models for the application. +""" +from openrouter_monitor.models.user import User +from openrouter_monitor.models.api_key import ApiKey +from openrouter_monitor.models.usage_stats import UsageStats +from openrouter_monitor.models.api_token import ApiToken + +__all__ = ["User", "ApiKey", "UsageStats", "ApiToken"] diff --git a/src/openrouter_monitor/models/api_key.py b/src/openrouter_monitor/models/api_key.py new file mode 100644 index 0000000..2f09706 --- /dev/null +++ b/src/openrouter_monitor/models/api_key.py @@ -0,0 +1,39 @@ +"""ApiKey model for OpenRouter API Key Monitor. + +T08: ApiKey SQLAlchemy model +""" +from datetime import datetime +from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey +from sqlalchemy.orm import relationship + +from openrouter_monitor.database import Base + + +class ApiKey(Base): + """API Key model for storing encrypted OpenRouter API keys. + + Attributes: + id: Primary key + user_id: Foreign key to users table + name: Human-readable name for the key + key_encrypted: AES-256 encrypted API key + is_active: Whether the key is active + created_at: Timestamp when key was created + last_used_at: Timestamp when key was last used + user: Relationship to user + usage_stats: Relationship to usage statistics + """ + + __tablename__ = "api_keys" + + id = Column(Integer, primary_key=True, index=True) + user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + name = Column(String(100), nullable=False) + key_encrypted = Column(String, nullable=False) + is_active = Column(Boolean, default=True, index=True) + created_at = Column(DateTime, default=datetime.utcnow) + last_used_at = Column(DateTime, nullable=True) + + # Relationships + user = relationship("User", back_populates="api_keys", lazy="selectin") + usage_stats = relationship("UsageStats", back_populates="api_key", cascade="all, delete-orphan", lazy="selectin") diff --git a/src/openrouter_monitor/models/api_token.py b/src/openrouter_monitor/models/api_token.py new file mode 100644 index 0000000..70837ad --- /dev/null +++ b/src/openrouter_monitor/models/api_token.py @@ -0,0 +1,37 @@ +"""ApiToken model for OpenRouter API Key Monitor. + +T10: ApiToken SQLAlchemy model +""" +from datetime import datetime +from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey +from sqlalchemy.orm import relationship + +from openrouter_monitor.database import Base + + +class ApiToken(Base): + """API Token model for public API access. + + Attributes: + id: Primary key + user_id: Foreign key to users table + token_hash: SHA-256 hash of the token (not the token itself) + name: Human-readable name for the token + created_at: Timestamp when token was created + last_used_at: Timestamp when token was last used + is_active: Whether the token is active + user: Relationship to user + """ + + __tablename__ = "api_tokens" + + id = Column(Integer, primary_key=True, index=True) + user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + token_hash = Column(String(255), nullable=False, index=True) + name = Column(String(100), nullable=False) + created_at = Column(DateTime, default=datetime.utcnow) + last_used_at = Column(DateTime, nullable=True) + is_active = Column(Boolean, default=True, index=True) + + # Relationships + user = relationship("User", back_populates="api_tokens", lazy="selectin") diff --git a/src/openrouter_monitor/models/usage_stats.py b/src/openrouter_monitor/models/usage_stats.py new file mode 100644 index 0000000..2addf95 --- /dev/null +++ b/src/openrouter_monitor/models/usage_stats.py @@ -0,0 +1,46 @@ +"""UsageStats model for OpenRouter API Key Monitor. + +T09: UsageStats SQLAlchemy model +""" +from datetime import datetime, date +from sqlalchemy import Column, Integer, String, Date, DateTime, Numeric, ForeignKey, UniqueConstraint +from sqlalchemy.orm import relationship + +from openrouter_monitor.database import Base + + +class UsageStats(Base): + """Usage statistics model for storing API usage data. + + Attributes: + id: Primary key + api_key_id: Foreign key to api_keys table + date: Date of the statistics + model: AI model name + requests_count: Number of requests + tokens_input: Number of input tokens + tokens_output: Number of output tokens + cost: Cost in USD (Numeric 10,6) + created_at: Timestamp when record was created + api_key: Relationship to API key + """ + + __tablename__ = "usage_stats" + + id = Column(Integer, primary_key=True, index=True) + api_key_id = Column(Integer, ForeignKey("api_keys.id", ondelete="CASCADE"), nullable=False, index=True) + date = Column(Date, nullable=False, index=True) + model = Column(String(100), nullable=False, index=True) + requests_count = Column(Integer, default=0) + tokens_input = Column(Integer, default=0) + tokens_output = Column(Integer, default=0) + cost = Column(Numeric(10, 6), default=0.0) + created_at = Column(DateTime, default=datetime.utcnow) + + # Unique constraint: one record per api_key, date, model + __table_args__ = ( + UniqueConstraint('api_key_id', 'date', 'model', name='uniq_key_date_model'), + ) + + # Relationships + api_key = relationship("ApiKey", back_populates="usage_stats", lazy="selectin") diff --git a/src/openrouter_monitor/models/user.py b/src/openrouter_monitor/models/user.py new file mode 100644 index 0000000..3b895a3 --- /dev/null +++ b/src/openrouter_monitor/models/user.py @@ -0,0 +1,37 @@ +"""User model for OpenRouter API Key Monitor. + +T07: User SQLAlchemy model +""" +from datetime import datetime +from sqlalchemy import Column, Integer, String, DateTime, Boolean +from sqlalchemy.orm import relationship + +from openrouter_monitor.database import Base + + +class User(Base): + """User model for storing user accounts. + + Attributes: + id: Primary key + email: User email address (unique, indexed) + password_hash: Bcrypt hashed password + created_at: Timestamp when user was created + updated_at: Timestamp when user was last updated + is_active: Whether the user account is active + api_keys: Relationship to user's API keys + api_tokens: Relationship to user's API tokens + """ + + __tablename__ = "users" + + id = Column(Integer, primary_key=True, index=True) + email = Column(String(255), unique=True, index=True, nullable=False) + password_hash = Column(String(255), nullable=False) + created_at = Column(DateTime, default=datetime.utcnow) + updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + is_active = Column(Boolean, default=True) + + # Relationships - using lazy string references to avoid circular imports + api_keys = relationship("ApiKey", back_populates="user", cascade="all, delete-orphan", lazy="selectin") + api_tokens = relationship("ApiToken", back_populates="user", cascade="all, delete-orphan", lazy="selectin") diff --git a/tests/unit/models/test_api_key_model.py b/tests/unit/models/test_api_key_model.py new file mode 100644 index 0000000..3d68e66 --- /dev/null +++ b/tests/unit/models/test_api_key_model.py @@ -0,0 +1,179 @@ +"""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() diff --git a/tests/unit/models/test_api_token_model.py b/tests/unit/models/test_api_token_model.py new file mode 100644 index 0000000..7c62cac --- /dev/null +++ b/tests/unit/models/test_api_token_model.py @@ -0,0 +1,216 @@ +"""Tests for ApiToken model (T10). + +T10: Creare model ApiToken (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 TestApiTokenModelBasics: + """Test ApiToken model basic attributes and creation.""" + + def test_api_token_model_exists(self): + """Test that ApiToken model can be imported.""" + # Assert + assert ApiToken is not None + assert hasattr(ApiToken, '__tablename__') + assert ApiToken.__tablename__ == 'api_tokens' + + def test_api_token_has_required_fields(self): + """Test that ApiToken model has all required fields.""" + # Assert + assert hasattr(ApiToken, 'id') + assert hasattr(ApiToken, 'user_id') + assert hasattr(ApiToken, 'token_hash') + assert hasattr(ApiToken, 'name') + assert hasattr(ApiToken, 'created_at') + assert hasattr(ApiToken, 'last_used_at') + assert hasattr(ApiToken, 'is_active') + + def test_api_token_create_with_valid_data(self, tmp_path): + """Test creating ApiToken with valid data.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_api_token.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act + token = ApiToken( + user_id=1, + token_hash="sha256_hash_here", + name="Integration Token" + ) + session.add(token) + session.flush() + + # Assert + assert token.token_hash == "sha256_hash_here" + assert token.name == "Integration Token" + assert token.is_active is True + assert token.created_at is not None + assert token.last_used_at is None + + session.close() + + +@pytest.mark.unit +class TestApiTokenConstraints: + """Test ApiToken model constraints.""" + + def test_api_token_user_id_index_exists(self): + """Test that user_id has an index.""" + # Act + inspector = inspect(ApiToken.__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_token_token_hash_index_exists(self): + """Test that token_hash has an index.""" + # Act + inspector = inspect(ApiToken.__table__) + indexes = inspector.indexes + + # Assert + index_names = [idx.name for idx in indexes] + assert any('token' in name or 'hash' in name for name in index_names) + + def test_api_token_is_active_index_exists(self): + """Test that is_active has an index.""" + # Act + inspector = inspect(ApiToken.__table__) + indexes = inspector.indexes + + # Assert + index_names = [idx.name for idx in indexes] + assert any('active' in name for name in index_names) + + +@pytest.mark.unit +class TestApiTokenRelationships: + """Test ApiToken model relationships.""" + + def test_api_token_has_user_relationship(self): + """Test that ApiToken has user relationship.""" + # Assert + assert hasattr(ApiToken, 'user') + + +@pytest.mark.integration +class TestApiTokenDatabaseIntegration: + """Integration tests for ApiToken model with database.""" + + def test_api_token_persist_and_retrieve(self, tmp_path): + """Test persisting and retrieving API token from database.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_token_persist.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act + token = ApiToken( + user_id=1, + token_hash="abc123hash456", + name="My Token" + ) + session.add(token) + session.commit() + + # Retrieve by hash + retrieved = session.query(ApiToken).filter_by(token_hash="abc123hash456").first() + + # Assert + assert retrieved is not None + assert retrieved.name == "My Token" + assert retrieved.id is not None + + session.close() + + def test_api_token_lookup_by_hash(self, tmp_path): + """Test looking up token by hash.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_lookup.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Create multiple tokens + token1 = ApiToken(user_id=1, token_hash="hash1", name="Token 1") + token2 = ApiToken(user_id=1, token_hash="hash2", name="Token 2") + session.add_all([token1, token2]) + session.commit() + + # Act - Look up by specific hash + result = session.query(ApiToken).filter_by(token_hash="hash2").first() + + # Assert + assert result is not None + assert result.name == "Token 2" + + session.close() + + def test_api_token_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 + token = ApiToken( + user_id=1, + token_hash="test_hash", + name="Test Token", + last_used_at=now + ) + session.add(token) + session.commit() + + # Retrieve + retrieved = session.query(ApiToken).first() + + # Assert + assert retrieved.last_used_at is not None + + session.close() + + def test_api_token_is_active_filtering(self, tmp_path): + """Test filtering tokens by is_active status.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_active_filter.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Create tokens + active = ApiToken(user_id=1, token_hash="active_hash", name="Active", is_active=True) + inactive = ApiToken(user_id=1, token_hash="inactive_hash", name="Inactive", is_active=False) + session.add_all([active, inactive]) + session.commit() + + # Act + active_tokens = session.query(ApiToken).filter_by(is_active=True).all() + inactive_tokens = session.query(ApiToken).filter_by(is_active=False).all() + + # Assert + assert len(active_tokens) == 1 + assert len(inactive_tokens) == 1 + assert active_tokens[0].name == "Active" + assert inactive_tokens[0].name == "Inactive" + + session.close() diff --git a/tests/unit/models/test_usage_stats_model.py b/tests/unit/models/test_usage_stats_model.py new file mode 100644 index 0000000..bd2bc13 --- /dev/null +++ b/tests/unit/models/test_usage_stats_model.py @@ -0,0 +1,243 @@ +"""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() diff --git a/tests/unit/models/test_user_model.py b/tests/unit/models/test_user_model.py new file mode 100644 index 0000000..2853faa --- /dev/null +++ b/tests/unit/models/test_user_model.py @@ -0,0 +1,280 @@ +"""Tests for User model (T07). + +T07: Creare model User (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 TestUserModelBasics: + """Test User model basic attributes and creation.""" + + def test_user_model_exists(self): + """Test that User model can be imported.""" + # Assert + assert User is not None + assert hasattr(User, '__tablename__') + assert User.__tablename__ == 'users' + + def test_user_has_required_fields(self): + """Test that User model has all required fields.""" + # Arrange + from sqlalchemy import Column, Integer, String, DateTime, Boolean + + # Assert + assert hasattr(User, 'id') + assert hasattr(User, 'email') + assert hasattr(User, 'password_hash') + assert hasattr(User, 'created_at') + assert hasattr(User, 'updated_at') + assert hasattr(User, 'is_active') + + def test_user_create_with_valid_data(self, tmp_path): + """Test creating User with valid data.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_user_data.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act + user = User( + email="test@example.com", + password_hash="hashed_password_here" + ) + session.add(user) + session.flush() # Apply defaults without committing + + # Assert + assert user.email == "test@example.com" + assert user.password_hash == "hashed_password_here" + assert user.is_active is True + assert user.created_at is not None + + session.close() + + def test_user_default_is_active_true(self, tmp_path): + """Test that is_active defaults to True.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_active_default.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act + user = User(email="test@example.com", password_hash="hash") + session.add(user) + session.flush() + + # Assert + assert user.is_active is True + + session.close() + + def test_user_timestamps_auto_set(self, tmp_path): + """Test that created_at is automatically set.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_timestamps.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act + before = datetime.utcnow() + user = User(email="test@example.com", password_hash="hash") + session.add(user) + session.flush() + after = datetime.utcnow() + + # Assert + assert user.created_at is not None + assert before <= user.created_at <= after + + session.close() + + +@pytest.mark.unit +class TestUserConstraints: + """Test User model constraints and validations.""" + + def test_user_email_unique_constraint(self, tmp_path): + """Test that email must be unique.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_unique.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act - Create first user + user1 = User(email="unique@example.com", password_hash="hash1") + session.add(user1) + session.commit() + + # Act - Try to create second user with same email + user2 = User(email="unique@example.com", password_hash="hash2") + session.add(user2) + + # Assert - Should raise IntegrityError + with pytest.raises(IntegrityError): + session.commit() + + session.close() + + def test_user_email_index_exists(self): + """Test that email has an index.""" + # Act - Check indexes on users table + inspector = inspect(User.__table__) + indexes = inspector.indexes + + # Assert + index_names = [idx.name for idx in indexes] + assert any('email' in name for name in index_names) + + +@pytest.mark.unit +class TestUserRelationships: + """Test User model relationships.""" + + def test_user_has_api_keys_relationship(self): + """Test that User has api_keys relationship.""" + # Assert + assert hasattr(User, 'api_keys') + + def test_user_has_api_tokens_relationship(self): + """Test that User has api_tokens relationship.""" + # Assert + assert hasattr(User, 'api_tokens') + + def test_api_keys_cascade_delete(self): + """Test that api_keys have cascade delete.""" + # Arrange + from sqlalchemy.orm import RelationshipProperty + + # Act + api_keys_rel = getattr(User.api_keys, 'property', None) + + # Assert + if api_keys_rel: + assert 'delete' in str(api_keys_rel.cascade).lower() or 'all' in str(api_keys_rel.cascade).lower() + + def test_api_tokens_cascade_delete(self): + """Test that api_tokens have cascade delete.""" + # Arrange + from sqlalchemy.orm import RelationshipProperty + + # Act + api_tokens_rel = getattr(User.api_tokens, 'property', None) + + # Assert + if api_tokens_rel: + assert 'delete' in str(api_tokens_rel.cascade).lower() or 'all' in str(api_tokens_rel.cascade).lower() + + +@pytest.mark.integration +class TestUserDatabaseIntegration: + """Integration tests for User model with database.""" + + def test_user_persist_and_retrieve(self, tmp_path): + """Test persisting and retrieving user from database.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_user.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Act + user = User(email="persist@example.com", password_hash="hashed123") + session.add(user) + session.commit() + + # Retrieve + retrieved = session.query(User).filter_by(email="persist@example.com").first() + + # Assert + assert retrieved is not None + assert retrieved.email == "persist@example.com" + assert retrieved.password_hash == "hashed123" + assert retrieved.id is not None + + session.close() + + def test_user_email_filtering(self, tmp_path): + """Test filtering users by email.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_filter.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Create multiple users + user1 = User(email="alice@example.com", password_hash="hash1") + user2 = User(email="bob@example.com", password_hash="hash2") + session.add_all([user1, user2]) + session.commit() + + # Act + result = session.query(User).filter_by(email="alice@example.com").first() + + # Assert + assert result is not None + assert result.email == "alice@example.com" + + session.close() + + def test_user_is_active_filtering(self, tmp_path): + """Test filtering users by is_active status.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_active.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Create users + active_user = User(email="active@example.com", password_hash="hash1", is_active=True) + inactive_user = User(email="inactive@example.com", password_hash="hash2", is_active=False) + session.add_all([active_user, inactive_user]) + session.commit() + + # Act + active_users = session.query(User).filter_by(is_active=True).all() + inactive_users = session.query(User).filter_by(is_active=False).all() + + # Assert + assert len(active_users) == 1 + assert len(inactive_users) == 1 + assert active_users[0].email == "active@example.com" + assert inactive_users[0].email == "inactive@example.com" + + session.close() + + def test_user_update_timestamp(self, tmp_path): + """Test that updated_at can be set and retrieved.""" + # Arrange + engine = create_engine(f'sqlite:///{tmp_path}/test_update.db') + Session = sessionmaker(bind=engine) + Base.metadata.create_all(bind=engine) + session = Session() + + # Create user + user = User(email="update@example.com", password_hash="hash") + session.add(user) + session.commit() + + # Act - Update + original_updated_at = user.updated_at + user.password_hash = "new_hash" + session.commit() + + # Assert + retrieved = session.query(User).filter_by(email="update@example.com").first() + assert retrieved.updated_at is not None + + session.close()