linmihigo
c086f64363
- Add 7 core API endpoints: users, transactions, partners, products, inventory, payments, credit - Implement role-based authentication (admin/write/read-only access) - Add comprehensive database models with proper relationships - Include full test coverage for all endpoints and business logic - Set up Alembic migrations and Docker configuration - Configure FastAPI app with CORS and database integration
254 lines
8.3 KiB
Python
254 lines
8.3 KiB
Python
"""Integration test configuration and fixtures.
|
|
|
|
This module provides fixtures and utilities for integration testing that involve
|
|
real database operations, Alembic migrations, and end-to-end API testing.
|
|
"""
|
|
|
|
import pytest
|
|
import tempfile
|
|
import os
|
|
from pathlib import Path
|
|
from sqlmodel import Session, SQLModel, create_engine, select, text
|
|
from sqlmodel.pool import StaticPool
|
|
from fastapi.testclient import TestClient
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from alembic.script import ScriptDirectory
|
|
from alembic.runtime.environment import EnvironmentContext
|
|
|
|
from app.main import app
|
|
from app.core.db import get_session
|
|
from app.core.config import settings
|
|
from app.schemas.models import User, Partner, Product
|
|
from app.schemas.base import UserRole
|
|
from app.core.auth import get_password_hash
|
|
|
|
|
|
class IntegrationTestConfig:
|
|
"""Configuration for integration tests."""
|
|
|
|
@staticmethod
|
|
def get_test_database_url():
|
|
"""Get test database URL. Uses a separate test database."""
|
|
# Use in-memory SQLite for integration tests to ensure writeability
|
|
return "sqlite:///:memory:"
|
|
|
|
|
|
@pytest.fixture(name="integration_engine", scope="session")
|
|
def integration_engine_fixture():
|
|
"""Create a test engine for integration tests."""
|
|
database_url = IntegrationTestConfig.get_test_database_url()
|
|
|
|
if database_url.startswith("sqlite"):
|
|
# For SQLite, use file-based database for integration tests
|
|
engine = create_engine(
|
|
database_url,
|
|
connect_args={"check_same_thread": False},
|
|
echo=False # Set to True for SQL debugging
|
|
)
|
|
else:
|
|
# For PostgreSQL
|
|
engine = create_engine(database_url, echo=False)
|
|
|
|
return engine
|
|
|
|
|
|
@pytest.fixture(name="integration_session", scope="function")
|
|
def integration_session_fixture(integration_engine):
|
|
"""Create a database session for integration tests with proper cleanup."""
|
|
# Create all tables
|
|
SQLModel.metadata.create_all(integration_engine)
|
|
|
|
with Session(integration_engine) as session:
|
|
yield session
|
|
|
|
# Clean up: drop all tables after each test
|
|
SQLModel.metadata.drop_all(integration_engine)
|
|
|
|
|
|
@pytest.fixture(name="integration_client")
|
|
def integration_client_fixture(integration_session):
|
|
"""Create a test client with integration database session."""
|
|
def get_session_override():
|
|
return integration_session
|
|
|
|
app.dependency_overrides[get_session] = get_session_override
|
|
client = TestClient(app)
|
|
yield client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture(name="alembic_config")
|
|
def alembic_config_fixture(integration_engine):
|
|
"""Create Alembic configuration for migration testing."""
|
|
# Create a temporary alembic.ini for testing
|
|
config = Config()
|
|
config.set_main_option("script_location", "app/alembic")
|
|
config.set_main_option("sqlalchemy.url", str(integration_engine.url))
|
|
|
|
return config
|
|
|
|
|
|
@pytest.fixture(name="migration_context")
|
|
def migration_context_fixture(integration_engine, alembic_config):
|
|
"""Create migration context for testing migrations."""
|
|
script = ScriptDirectory.from_config(alembic_config)
|
|
|
|
def run_migrations(connection, config):
|
|
context = EnvironmentContext(config, script)
|
|
context.configure(
|
|
connection=connection,
|
|
target_metadata=SQLModel.metadata
|
|
)
|
|
with context.begin_transaction():
|
|
context.run_migrations()
|
|
|
|
with integration_engine.connect() as connection:
|
|
yield {
|
|
'connection': connection,
|
|
'config': alembic_config,
|
|
'script': script,
|
|
'run_migrations': lambda: run_migrations(connection, alembic_config)
|
|
}
|
|
|
|
|
|
@pytest.fixture(name="integration_admin_user")
|
|
def integration_admin_user_fixture(integration_session):
|
|
"""Create an admin user for integration tests."""
|
|
admin_user = User(
|
|
username="integration_admin",
|
|
password_hash=get_password_hash("admin_password"),
|
|
role=UserRole.ADMIN
|
|
)
|
|
integration_session.add(admin_user)
|
|
integration_session.commit()
|
|
integration_session.refresh(admin_user)
|
|
return admin_user
|
|
|
|
|
|
@pytest.fixture(name="integration_write_user")
|
|
def integration_write_user_fixture(integration_session):
|
|
"""Create a write user for integration tests."""
|
|
write_user = User(
|
|
username="integration_write",
|
|
password_hash=get_password_hash("write_password"),
|
|
role=UserRole.WRITE
|
|
)
|
|
integration_session.add(write_user)
|
|
integration_session.commit()
|
|
integration_session.refresh(write_user)
|
|
return write_user
|
|
|
|
|
|
@pytest.fixture(name="integration_read_user")
|
|
def integration_read_user_fixture(integration_session):
|
|
"""Create a read-only user for integration tests."""
|
|
read_user = User(
|
|
username="integration_read",
|
|
password_hash=get_password_hash("read_password"),
|
|
role=UserRole.READ_ONLY
|
|
)
|
|
integration_session.add(read_user)
|
|
integration_session.commit()
|
|
integration_session.refresh(read_user)
|
|
return read_user
|
|
|
|
|
|
@pytest.fixture(name="integration_admin_token")
|
|
def integration_admin_token_fixture(integration_client, integration_admin_user):
|
|
"""Get admin authentication headers for integration tests."""
|
|
response = integration_client.post("/api/v1/users/login", json={
|
|
"username": "integration_admin",
|
|
"password": "admin_password"
|
|
})
|
|
assert response.status_code == 200
|
|
token = response.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture(name="integration_write_token")
|
|
def integration_write_token_fixture(integration_client, integration_write_user):
|
|
"""Get write user authentication headers for integration tests."""
|
|
response = integration_client.post("/api/v1/users/login", json={
|
|
"username": "integration_write",
|
|
"password": "write_password"
|
|
})
|
|
assert response.status_code == 200
|
|
token = response.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture(name="integration_read_token")
|
|
def integration_read_token_fixture(integration_client, integration_read_user):
|
|
"""Get read-only user authentication headers for integration tests."""
|
|
response = integration_client.post("/api/v1/users/login", json={
|
|
"username": "integration_read",
|
|
"password": "read_password"
|
|
})
|
|
assert response.status_code == 200
|
|
token = response.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def cleanup_test_database():
|
|
"""Utility function to clean up test database files."""
|
|
test_db_files = ["test_integration.db", "test_integration.db-wal", "test_integration.db-shm"]
|
|
for file_name in test_db_files:
|
|
if os.path.exists(file_name):
|
|
try:
|
|
os.remove(file_name)
|
|
except OSError:
|
|
pass # File might be in use or already deleted
|
|
|
|
|
|
def verify_database_integrity(session: Session) -> dict:
|
|
"""Verify database integrity and return diagnostics."""
|
|
try:
|
|
# Check if we can query basic tables using SQLModel queries
|
|
users = session.exec(select(User)).all()
|
|
partners = session.exec(select(Partner)).all()
|
|
products = session.exec(select(Product)).all()
|
|
|
|
return {
|
|
"status": "healthy",
|
|
"users": len(users),
|
|
"partners": len(partners),
|
|
"products": len(products),
|
|
"tables_accessible": True
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"status": "error",
|
|
"error": str(e),
|
|
"tables_accessible": False
|
|
}
|
|
|
|
|
|
# Pytest configuration for integration tests
|
|
def pytest_configure(config):
|
|
"""Configure pytest for integration tests."""
|
|
config.addinivalue_line(
|
|
"markers", "integration: mark test as integration test"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "slow: mark test as slow running"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "database: mark test as requiring database"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "migration: mark test as testing migrations"
|
|
)
|
|
|
|
|
|
def pytest_runtest_setup(item):
|
|
"""Setup for each integration test."""
|
|
# Clean up any leftover test database files
|
|
cleanup_test_database()
|
|
|
|
|
|
def pytest_runtest_teardown(item):
|
|
"""Teardown for each integration test."""
|
|
# Clean up test database files after each test
|
|
cleanup_test_database()
|