"""Tests for Alembic migrations (T11). T11: Setup Alembic e migrazione iniziale """ import pytest import os import tempfile from pathlib import Path @pytest.mark.unit class TestAlembicInitialization: """Test Alembic initialization and configuration.""" def test_alembic_ini_exists(self): """Test that alembic.ini file exists.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent alembic_ini_path = project_root / "alembic.ini" # Assert assert alembic_ini_path.exists(), "alembic.ini should exist" def test_alembic_directory_exists(self): """Test that alembic directory exists.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent alembic_dir = project_root / "alembic" # Assert assert alembic_dir.exists(), "alembic directory should exist" assert alembic_dir.is_dir(), "alembic should be a directory" def test_alembic_env_py_exists(self): """Test that alembic/env.py file exists.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent env_py_path = project_root / "alembic" / "env.py" # Assert assert env_py_path.exists(), "alembic/env.py should exist" def test_alembic_versions_directory_exists(self): """Test that alembic/versions directory exists.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent versions_dir = project_root / "alembic" / "versions" # Assert assert versions_dir.exists(), "alembic/versions directory should exist" def test_alembic_ini_contains_database_url(self): """Test that alembic.ini contains DATABASE_URL configuration.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent alembic_ini_path = project_root / "alembic.ini" # Act with open(alembic_ini_path, 'r') as f: content = f.read() # Assert assert "sqlalchemy.url" in content, "alembic.ini should contain sqlalchemy.url" def test_alembic_env_py_imports_base(self): """Test that alembic/env.py imports Base from models.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent env_py_path = project_root / "alembic" / "env.py" # Act with open(env_py_path, 'r') as f: content = f.read() # Assert assert "Base" in content or "target_metadata" in content, \ "alembic/env.py should reference Base or target_metadata" @pytest.mark.integration class TestAlembicMigrations: """Test Alembic migration functionality.""" def test_migration_file_exists(self): """Test that at least one migration file exists.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent versions_dir = project_root / "alembic" / "versions" # Act migration_files = list(versions_dir.glob("*.py")) # Assert assert len(migration_files) > 0, "At least one migration file should exist" def test_migration_contains_create_tables(self): """Test that migration contains table creation commands.""" # Arrange project_root = Path(__file__).parent.parent.parent.parent versions_dir = project_root / "alembic" / "versions" # Get the first migration file migration_files = list(versions_dir.glob("*.py")) if not migration_files: pytest.skip("No migration files found") migration_file = migration_files[0] # Act with open(migration_file, 'r') as f: content = f.read() # Assert assert "upgrade" in content, "Migration should contain upgrade function" assert "downgrade" in content, "Migration should contain downgrade function" def test_alembic_upgrade_creates_tables(self, tmp_path): """Test that alembic upgrade creates all required tables.""" # Arrange import subprocess import sys # Create a temporary database db_path = tmp_path / "test_alembic.db" # Set up environment with test database env = os.environ.copy() env['DATABASE_URL'] = f"sqlite:///{db_path}" env['SECRET_KEY'] = "test-secret-key-min-32-characters-long" env['ENCRYPTION_KEY'] = "test-32-byte-encryption-key!!" # Change to project root project_root = Path(__file__).parent.parent.parent.parent # Act - Run alembic upgrade result = subprocess.run( [sys.executable, "-m", "alembic", "upgrade", "head"], cwd=project_root, env=env, capture_output=True, text=True ) # Assert assert result.returncode == 0, f"Alembic upgrade failed: {result.stderr}" # Verify database file exists assert db_path.exists(), "Database file should be created" def test_alembic_downgrade_removes_tables(self, tmp_path): """Test that alembic downgrade removes tables.""" # Arrange import subprocess import sys # Create a temporary database db_path = tmp_path / "test_alembic_downgrade.db" # Set up environment with test database env = os.environ.copy() env['DATABASE_URL'] = f"sqlite:///{db_path}" env['SECRET_KEY'] = "test-secret-key-min-32-characters-long" env['ENCRYPTION_KEY'] = "test-32-byte-encryption-key!!" # Change to project root project_root = Path(__file__).parent.parent.parent.parent # Act - First upgrade subprocess.run( [sys.executable, "-m", "alembic", "upgrade", "head"], cwd=project_root, env=env, capture_output=True, text=True ) # Then downgrade result = subprocess.run( [sys.executable, "-m", "alembic", "downgrade", "-1"], cwd=project_root, env=env, capture_output=True, text=True ) # Assert assert result.returncode == 0, f"Alembic downgrade failed: {result.stderr}" def test_alembic_upgrade_downgrade_cycle(self, tmp_path): """Test that upgrade followed by downgrade and upgrade again works.""" # Arrange import subprocess import sys # Create a temporary database db_path = tmp_path / "test_alembic_cycle.db" # Set up environment with test database env = os.environ.copy() env['DATABASE_URL'] = f"sqlite:///{db_path}" env['SECRET_KEY'] = "test-secret-key-min-32-characters-long" env['ENCRYPTION_KEY'] = "test-32-byte-encryption-key!!" # Change to project root project_root = Path(__file__).parent.parent.parent.parent # Act - Upgrade result1 = subprocess.run( [sys.executable, "-m", "alembic", "upgrade", "head"], cwd=project_root, env=env, capture_output=True, text=True ) # Downgrade result2 = subprocess.run( [sys.executable, "-m", "alembic", "downgrade", "-1"], cwd=project_root, env=env, capture_output=True, text=True ) # Upgrade again result3 = subprocess.run( [sys.executable, "-m", "alembic", "upgrade", "head"], cwd=project_root, env=env, capture_output=True, text=True ) # Assert assert result1.returncode == 0, "First upgrade failed" assert result2.returncode == 0, "Downgrade failed" assert result3.returncode == 0, "Second upgrade failed" @pytest.mark.integration class TestDatabaseTables: """Test that database tables are created correctly.""" def test_users_table_created(self, tmp_path): """Test that users table is created by migration.""" # Arrange import subprocess import sys from sqlalchemy import create_engine, inspect # Create a temporary database db_path = tmp_path / "test_tables.db" # Set up environment with test database env = os.environ.copy() env['DATABASE_URL'] = f"sqlite:///{db_path}" env['SECRET_KEY'] = "test-secret-key-min-32-characters-long" env['ENCRYPTION_KEY'] = "test-32-byte-encryption-key!!" # Change to project root project_root = Path(__file__).parent.parent.parent.parent # Act - Run alembic upgrade subprocess.run( [sys.executable, "-m", "alembic", "upgrade", "head"], cwd=project_root, env=env, capture_output=True, text=True ) # Verify tables engine = create_engine(f"sqlite:///{db_path}") inspector = inspect(engine) tables = inspector.get_table_names() # Assert assert "users" in tables, "users table should be created" assert "api_keys" in tables, "api_keys table should be created" assert "usage_stats" in tables, "usage_stats table should be created" assert "api_tokens" in tables, "api_tokens table should be created" engine.dispose() def test_alembic_version_table_created(self, tmp_path): """Test that alembic_version table is created.""" # Arrange import subprocess import sys from sqlalchemy import create_engine, inspect # Create a temporary database db_path = tmp_path / "test_version.db" # Set up environment with test database env = os.environ.copy() env['DATABASE_URL'] = f"sqlite:///{db_path}" env['SECRET_KEY'] = "test-secret-key-min-32-characters-long" env['ENCRYPTION_KEY'] = "test-32-byte-encryption-key!!" # Change to project root project_root = Path(__file__).parent.parent.parent.parent # Act - Run alembic upgrade subprocess.run( [sys.executable, "-m", "alembic", "upgrade", "head"], cwd=project_root, env=env, capture_output=True, text=True ) # Verify tables engine = create_engine(f"sqlite:///{db_path}") inspector = inspect(engine) tables = inspector.get_table_names() # Assert assert "alembic_version" in tables, "alembic_version table should be created" engine.dispose()