Petru Paler
7d7cd2bb57
- Add migrations.py with create_migration, run_migrations, get_db_version - Implement CLI migrate and create-migration commands - Use FastMigrate's sequential 0001-*.sql naming convention - Add comprehensive unit, integration, and E2E tests (35 tests) - Add fresh_db_path and temp_migrations_dir test fixtures 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
368 lines
14 KiB
Python
368 lines
14 KiB
Python
# ABOUTME: Tests for the migration framework.
|
|
# ABOUTME: Validates migration discovery, creation, and application.
|
|
|
|
import pytest
|
|
|
|
|
|
class TestGetNextMigrationIndex:
|
|
"""Test discovery of next migration index."""
|
|
|
|
def test_empty_directory_returns_1(self, temp_migrations_dir):
|
|
"""First migration should be index 1."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
result = get_next_migration_index(temp_migrations_dir)
|
|
assert result == 1
|
|
|
|
def test_nonexistent_directory_returns_1(self, tmp_path):
|
|
"""Non-existent directory should return 1."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
nonexistent = tmp_path / "does_not_exist"
|
|
result = get_next_migration_index(nonexistent)
|
|
assert result == 1
|
|
|
|
def test_finds_highest_index(self, temp_migrations_dir):
|
|
"""Should find highest existing index and increment."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
# Create some migration files
|
|
(temp_migrations_dir / "0001-first.sql").write_text("-- migration 1")
|
|
(temp_migrations_dir / "0002-second.sql").write_text("-- migration 2")
|
|
|
|
result = get_next_migration_index(temp_migrations_dir)
|
|
assert result == 3
|
|
|
|
def test_handles_gaps_in_sequence(self, temp_migrations_dir):
|
|
"""Should handle non-contiguous indexes."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
# Create migrations with gaps
|
|
(temp_migrations_dir / "0001-first.sql").write_text("-- migration 1")
|
|
(temp_migrations_dir / "0003-third.sql").write_text("-- migration 3")
|
|
(temp_migrations_dir / "0005-fifth.sql").write_text("-- migration 5")
|
|
|
|
result = get_next_migration_index(temp_migrations_dir)
|
|
assert result == 6
|
|
|
|
def test_handles_python_migrations(self, temp_migrations_dir):
|
|
"""Should handle .py migration files."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
(temp_migrations_dir / "0001-first.sql").write_text("-- sql")
|
|
(temp_migrations_dir / "0002-second.py").write_text("# python")
|
|
|
|
result = get_next_migration_index(temp_migrations_dir)
|
|
assert result == 3
|
|
|
|
def test_handles_shell_migrations(self, temp_migrations_dir):
|
|
"""Should handle .sh migration files."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
(temp_migrations_dir / "0001-first.sql").write_text("-- sql")
|
|
(temp_migrations_dir / "0002-second.sh").write_text("#!/bin/bash")
|
|
|
|
result = get_next_migration_index(temp_migrations_dir)
|
|
assert result == 3
|
|
|
|
def test_ignores_non_migration_files(self, temp_migrations_dir):
|
|
"""Should ignore files that don't match migration pattern."""
|
|
from animaltrack.migrations import get_next_migration_index
|
|
|
|
# Create non-migration files
|
|
(temp_migrations_dir / "0001-first.sql").write_text("-- migration")
|
|
(temp_migrations_dir / "README.md").write_text("# readme")
|
|
(temp_migrations_dir / ".gitkeep").write_text("")
|
|
(temp_migrations_dir / "random.txt").write_text("random")
|
|
|
|
result = get_next_migration_index(temp_migrations_dir)
|
|
assert result == 2
|
|
|
|
|
|
class TestSanitizeDescription:
|
|
"""Test description sanitization for filenames."""
|
|
|
|
def test_lowercases_text(self):
|
|
"""Should convert to lowercase."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description("Add USERS Table")
|
|
assert result == "add-users-table"
|
|
|
|
def test_replaces_spaces_with_hyphens(self):
|
|
"""Should replace spaces with hyphens."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description("add users table")
|
|
assert result == "add-users-table"
|
|
|
|
def test_removes_special_characters(self):
|
|
"""Should remove special characters."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description("Add Users Table!")
|
|
assert result == "add-users-table"
|
|
|
|
def test_collapses_multiple_hyphens(self):
|
|
"""Should collapse multiple hyphens into one."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description("add users table")
|
|
assert result == "add-users-table"
|
|
|
|
def test_strips_leading_trailing_hyphens(self):
|
|
"""Should strip leading and trailing hyphens."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description(" add users table ")
|
|
assert result == "add-users-table"
|
|
|
|
def test_handles_numbers(self):
|
|
"""Should keep numbers."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description("Fix Bug 123")
|
|
assert result == "fix-bug-123"
|
|
|
|
def test_handles_empty_after_sanitization(self):
|
|
"""Should handle strings that become empty after sanitization."""
|
|
from animaltrack.migrations import sanitize_description
|
|
|
|
result = sanitize_description("!!!")
|
|
assert result == ""
|
|
|
|
|
|
class TestCreateMigration:
|
|
"""Test migration file creation."""
|
|
|
|
def test_creates_file_with_correct_name(self, temp_migrations_dir):
|
|
"""Should create migration with 4-digit index and description."""
|
|
from animaltrack.migrations import create_migration
|
|
|
|
path = create_migration(temp_migrations_dir, "add users table")
|
|
assert path.name == "0001-add-users-table.sql"
|
|
assert path.exists()
|
|
|
|
def test_includes_aboutme_header(self, temp_migrations_dir):
|
|
"""Created migration should have ABOUTME comments."""
|
|
from animaltrack.migrations import create_migration
|
|
|
|
path = create_migration(temp_migrations_dir, "add users table")
|
|
content = path.read_text()
|
|
assert "-- ABOUTME:" in content
|
|
assert "add users table" in content
|
|
|
|
def test_increments_index(self, temp_migrations_dir):
|
|
"""Second migration should have index 2."""
|
|
from animaltrack.migrations import create_migration
|
|
|
|
path1 = create_migration(temp_migrations_dir, "first migration")
|
|
path2 = create_migration(temp_migrations_dir, "second migration")
|
|
|
|
assert path1.name == "0001-first-migration.sql"
|
|
assert path2.name == "0002-second-migration.sql"
|
|
|
|
def test_returns_pathlib_path(self, temp_migrations_dir):
|
|
"""Should return pathlib.Path object."""
|
|
from pathlib import Path
|
|
|
|
from animaltrack.migrations import create_migration
|
|
|
|
result = create_migration(temp_migrations_dir, "test")
|
|
assert isinstance(result, Path)
|
|
|
|
def test_creates_directory_if_not_exists(self, tmp_path):
|
|
"""Should create migrations directory if it doesn't exist."""
|
|
from animaltrack.migrations import create_migration
|
|
|
|
nonexistent = tmp_path / "new_migrations"
|
|
path = create_migration(nonexistent, "test")
|
|
|
|
assert nonexistent.exists()
|
|
assert path.exists()
|
|
|
|
|
|
class TestGetDbVersion:
|
|
"""Test database version retrieval."""
|
|
|
|
def test_returns_zero_for_new_versioned_db(self, fresh_db_path):
|
|
"""New versioned database should be version 0."""
|
|
import fastmigrate
|
|
|
|
from animaltrack.migrations import get_db_version
|
|
|
|
# Initialize db with fastmigrate
|
|
fastmigrate.create_db(fresh_db_path)
|
|
|
|
result = get_db_version(fresh_db_path)
|
|
assert result == 0
|
|
|
|
def test_raises_for_nonexistent_db(self, tmp_path):
|
|
"""Should raise FileNotFoundError for non-existent database."""
|
|
from animaltrack.migrations import get_db_version
|
|
|
|
nonexistent = tmp_path / "does_not_exist.db"
|
|
with pytest.raises(FileNotFoundError):
|
|
get_db_version(nonexistent)
|
|
|
|
|
|
class TestRunMigrations:
|
|
"""Test migration execution."""
|
|
|
|
def test_initializes_unversioned_db(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should initialize database if not versioned."""
|
|
from animaltrack.migrations import get_db_version, run_migrations
|
|
|
|
# Fresh db - run migrations should initialize it
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is True
|
|
|
|
# Should now have version 0 (no migrations to run)
|
|
version = get_db_version(fresh_db_path)
|
|
assert version == 0
|
|
|
|
def test_runs_pending_migrations(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should execute all pending migrations in order."""
|
|
from animaltrack.db import get_db
|
|
from animaltrack.migrations import get_db_version, run_migrations
|
|
|
|
# Create migrations
|
|
(temp_migrations_dir / "0001-create-table.sql").write_text(
|
|
"CREATE TABLE test1 (id INTEGER PRIMARY KEY);"
|
|
)
|
|
(temp_migrations_dir / "0002-add-column.sql").write_text(
|
|
"CREATE TABLE test2 (id INTEGER PRIMARY KEY);"
|
|
)
|
|
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is True
|
|
|
|
# Version should be 2
|
|
assert get_db_version(fresh_db_path) == 2
|
|
|
|
# Tables should exist
|
|
db = get_db(fresh_db_path)
|
|
tables = db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'test%'"
|
|
).fetchall()
|
|
table_names = [t[0] for t in tables]
|
|
assert "test1" in table_names
|
|
assert "test2" in table_names
|
|
|
|
def test_skips_applied_migrations(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should not re-run already applied migrations."""
|
|
from animaltrack.migrations import get_db_version, run_migrations
|
|
|
|
# Create migration and run
|
|
(temp_migrations_dir / "0001-create-table.sql").write_text(
|
|
"CREATE TABLE test1 (id INTEGER PRIMARY KEY);"
|
|
)
|
|
run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert get_db_version(fresh_db_path) == 1
|
|
|
|
# Add another migration and run again
|
|
(temp_migrations_dir / "0002-another-table.sql").write_text(
|
|
"CREATE TABLE test2 (id INTEGER PRIMARY KEY);"
|
|
)
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is True
|
|
assert get_db_version(fresh_db_path) == 2
|
|
|
|
def test_stops_on_migration_failure(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should stop at first failing migration."""
|
|
from animaltrack.migrations import get_db_version, run_migrations
|
|
|
|
# Create migrations - second one will fail
|
|
(temp_migrations_dir / "0001-good.sql").write_text(
|
|
"CREATE TABLE test1 (id INTEGER PRIMARY KEY);"
|
|
)
|
|
(temp_migrations_dir / "0002-bad.sql").write_text("THIS IS INVALID SQL;")
|
|
(temp_migrations_dir / "0003-never-runs.sql").write_text(
|
|
"CREATE TABLE test3 (id INTEGER PRIMARY KEY);"
|
|
)
|
|
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is False
|
|
|
|
# Version should be 1 (stopped at failure)
|
|
assert get_db_version(fresh_db_path) == 1
|
|
|
|
def test_returns_true_when_no_migrations(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should return True when no migrations to run."""
|
|
from animaltrack.migrations import run_migrations
|
|
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is True
|
|
|
|
|
|
class TestMigrationIntegration:
|
|
"""Integration tests for full migration lifecycle."""
|
|
|
|
def test_create_and_run_migration(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should create migration file and successfully apply it."""
|
|
from animaltrack.db import get_db
|
|
from animaltrack.migrations import (
|
|
create_migration,
|
|
get_db_version,
|
|
run_migrations,
|
|
)
|
|
|
|
# Create migration
|
|
path = create_migration(temp_migrations_dir, "create users")
|
|
|
|
# Edit it to have actual SQL
|
|
path.write_text(
|
|
"""-- ABOUTME: Migration 0001 - create users
|
|
-- ABOUTME: Creates the users table.
|
|
|
|
CREATE TABLE users (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
|
|
# Run migrations
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is True
|
|
assert get_db_version(fresh_db_path) == 1
|
|
|
|
# Verify table exists
|
|
db = get_db(fresh_db_path)
|
|
result = db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='users'"
|
|
).fetchone()
|
|
assert result is not None
|
|
|
|
def test_multiple_migrations_in_sequence(self, fresh_db_path, temp_migrations_dir):
|
|
"""Should handle creating and running multiple migrations."""
|
|
from animaltrack.db import get_db
|
|
from animaltrack.migrations import (
|
|
create_migration,
|
|
get_db_version,
|
|
run_migrations,
|
|
)
|
|
|
|
# Create and populate multiple migrations
|
|
path1 = create_migration(temp_migrations_dir, "create users")
|
|
path1.write_text("CREATE TABLE users (id TEXT PRIMARY KEY);")
|
|
|
|
path2 = create_migration(temp_migrations_dir, "create products")
|
|
path2.write_text("CREATE TABLE products (id TEXT PRIMARY KEY);")
|
|
|
|
path3 = create_migration(temp_migrations_dir, "create orders")
|
|
path3.write_text("CREATE TABLE orders (id TEXT PRIMARY KEY);")
|
|
|
|
# Run all
|
|
success = run_migrations(fresh_db_path, temp_migrations_dir, verbose=False)
|
|
assert success is True
|
|
assert get_db_version(fresh_db_path) == 3
|
|
|
|
# Verify all tables exist
|
|
db = get_db(fresh_db_path)
|
|
for table_name in ["users", "products", "orders"]:
|
|
result = db.execute(
|
|
f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}'"
|
|
).fetchone()
|
|
assert result is not None, f"Table {table_name} should exist"
|