a8151fbe66
Backend Tests: - Add pytest configuration with async support (conftest.py) - Add model tests: User, Conversation, Message, TokenUsage, Session, UserMemory - Add configuration tests: Settings validation and environment variables - Add API tests: Health endpoint and future endpoint stubs - Add database tests: Connection, transactions, query execution Phase 1 Foundation: - FastAPI application structure with main.py - SQLAlchemy async models for all entities - Alembic migrations setup - Configuration management via Pydantic Settings - Logging system (English only) - Docker multi-stage builds for backend - Docker Compose orchestration (PostgreSQL, Redis, backend) - Frontend React + TypeScript structure - Dark & Gold theme CSS implementation - Environment configuration examples All code and comments in English as per requirements. Tests cover model relationships, cascade deletes, and constraints. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
253 lines
6.8 KiB
Python
253 lines
6.8 KiB
Python
"""
|
|
Tests for SQLAlchemy models
|
|
|
|
Tests model creation, relationships, and constraints
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
from sqlalchemy import select
|
|
|
|
from app.models.user import User
|
|
from app.models.conversation import Conversation
|
|
from app.models.message import Message
|
|
from app.models.token_usage import TokenUsage
|
|
from app.models.session import Session
|
|
from app.models.user_memory import UserMemory
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user(db_session):
|
|
"""Test creating a user"""
|
|
user = User(
|
|
azure_ad_id="test-azure-id-123",
|
|
email="test@example.com",
|
|
display_name="Test User",
|
|
given_name="Test",
|
|
surname="User",
|
|
role="user",
|
|
)
|
|
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
assert user.id is not None
|
|
assert user.email == "test@example.com"
|
|
assert user.azure_ad_id == "test-azure-id-123"
|
|
assert user.is_active is True
|
|
assert user.role == "user"
|
|
assert isinstance(user.created_at, datetime)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_conversation_relationship(db_session):
|
|
"""Test user-conversation relationship"""
|
|
# Create user
|
|
user = User(
|
|
azure_ad_id="test-azure-id-456",
|
|
email="user@example.com",
|
|
display_name="User Test",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Create conversation
|
|
conversation = Conversation(
|
|
user_id=user.id,
|
|
title="Test Conversation",
|
|
)
|
|
db_session.add(conversation)
|
|
await db_session.commit()
|
|
await db_session.refresh(conversation)
|
|
|
|
# Verify relationship
|
|
assert conversation.user_id == user.id
|
|
assert conversation.title == "Test Conversation"
|
|
assert isinstance(conversation.created_at, datetime)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_message_relationship(db_session):
|
|
"""Test conversation-message relationship"""
|
|
# Create user and conversation
|
|
user = User(
|
|
azure_ad_id="test-azure-id-789",
|
|
email="messages@example.com",
|
|
display_name="Message User",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
conversation = Conversation(user_id=user.id, title="Message Test")
|
|
db_session.add(conversation)
|
|
await db_session.commit()
|
|
await db_session.refresh(conversation)
|
|
|
|
# Create messages
|
|
user_message = Message(
|
|
conversation_id=conversation.id,
|
|
role="user",
|
|
content="Hello, bot!",
|
|
token_count=3,
|
|
)
|
|
assistant_message = Message(
|
|
conversation_id=conversation.id,
|
|
role="assistant",
|
|
content="Hello! How can I help you?",
|
|
token_count=7,
|
|
)
|
|
|
|
db_session.add_all([user_message, assistant_message])
|
|
await db_session.commit()
|
|
|
|
# Fetch conversation with messages
|
|
result = await db_session.execute(
|
|
select(Conversation).where(Conversation.id == conversation.id)
|
|
)
|
|
conv = result.scalar_one()
|
|
|
|
assert len(conv.messages) == 2
|
|
assert conv.messages[0].role == "user"
|
|
assert conv.messages[1].role == "assistant"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_token_usage_tracking(db_session):
|
|
"""Test token usage model"""
|
|
user = User(
|
|
azure_ad_id="test-token-user",
|
|
email="token@example.com",
|
|
display_name="Token User",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Create token usage record
|
|
token_usage = TokenUsage(
|
|
user_id=user.id,
|
|
prompt_tokens=100,
|
|
completion_tokens=50,
|
|
total_tokens=150,
|
|
model="gpt-5-nano-2025-08-07",
|
|
cost_usd=0.015,
|
|
operation_type="chat",
|
|
)
|
|
|
|
db_session.add(token_usage)
|
|
await db_session.commit()
|
|
await db_session.refresh(token_usage)
|
|
|
|
assert token_usage.prompt_tokens == 100
|
|
assert token_usage.completion_tokens == 50
|
|
assert token_usage.total_tokens == 150
|
|
assert token_usage.model == "gpt-5-nano-2025-08-07"
|
|
assert float(token_usage.cost_usd) == 0.015
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_memory(db_session):
|
|
"""Test user memory model"""
|
|
user = User(
|
|
azure_ad_id="test-memory-user",
|
|
email="memory@example.com",
|
|
display_name="Memory User",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Create memory
|
|
memory = UserMemory(
|
|
user_id=user.id,
|
|
memory_key="office_location",
|
|
memory_value="Sydney office",
|
|
importance_score=0.8,
|
|
)
|
|
|
|
db_session.add(memory)
|
|
await db_session.commit()
|
|
await db_session.refresh(memory)
|
|
|
|
assert memory.memory_key == "office_location"
|
|
assert memory.memory_value == "Sydney office"
|
|
assert float(memory.importance_score) == 0.8
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_creation(db_session):
|
|
"""Test session model"""
|
|
user = User(
|
|
azure_ad_id="test-session-user",
|
|
email="session@example.com",
|
|
display_name="Session User",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Create session
|
|
from datetime import timedelta
|
|
|
|
session = Session(
|
|
user_id=user.id,
|
|
access_token_hash="hashed_token_123",
|
|
expires_at=datetime.utcnow() + timedelta(hours=1),
|
|
ip_address="127.0.0.1",
|
|
user_agent="Test Agent",
|
|
)
|
|
|
|
db_session.add(session)
|
|
await db_session.commit()
|
|
await db_session.refresh(session)
|
|
|
|
assert session.user_id == user.id
|
|
assert session.access_token_hash == "hashed_token_123"
|
|
assert session.is_active is True
|
|
assert session.ip_address == "127.0.0.1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cascade_delete(db_session):
|
|
"""Test cascade delete functionality"""
|
|
# Create user with conversation and messages
|
|
user = User(
|
|
azure_ad_id="test-cascade-user",
|
|
email="cascade@example.com",
|
|
display_name="Cascade User",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
conversation = Conversation(user_id=user.id, title="Cascade Test")
|
|
db_session.add(conversation)
|
|
await db_session.commit()
|
|
await db_session.refresh(conversation)
|
|
|
|
message = Message(
|
|
conversation_id=conversation.id,
|
|
role="user",
|
|
content="Test message",
|
|
)
|
|
db_session.add(message)
|
|
await db_session.commit()
|
|
|
|
# Delete user - should cascade
|
|
await db_session.delete(user)
|
|
await db_session.commit()
|
|
|
|
# Verify conversation is deleted
|
|
result = await db_session.execute(
|
|
select(Conversation).where(Conversation.id == conversation.id)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
# Verify message is deleted
|
|
result = await db_session.execute(
|
|
select(Message).where(Message.id == message.id)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|