diff --git a/tests/test_cli.py b/tests/test_cli.py index 0cdd004..23c2e0f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2,7 +2,6 @@ import os import sqlite3 -import tempfile from pathlib import Path import io import sys @@ -24,347 +23,332 @@ def test_cli_help_backup_db(): assert result.returncode == 0 assert "usage: fastmigrate_backup_db [-h] [--db DB]" in result.stdout -def test_cli_explicit_paths(): +def test_cli_explicit_paths(tmp_path): """Test CLI with explicit path arguments.""" - with tempfile.TemporaryDirectory() as temp_dir: - # Create custom directories - temp_dir_path = Path(temp_dir) - migrations_dir = temp_dir_path / "custom_migrations" - db_dir = temp_dir_path / "custom_data" - migrations_dir.mkdir() - db_dir.mkdir() - - db_path = db_dir / "custom.db" - - # Create empty database file - conn = sqlite3.connect(db_path) - conn.close() - - # Initialize the database with _meta table - _ensure_meta_table(db_path) - - # Create a migration - with open(migrations_dir / "0001-test.sql", "w") as f: - f.write("CREATE TABLE custom (id INTEGER PRIMARY KEY);") - - # Run with explicit paths - result = subprocess.run([ - "fastmigrate_run_migrations", - "--db", db_path, - "--migrations", migrations_dir - ]) - - assert result.returncode == 0 - - # Verify migration was applied - conn = sqlite3.connect(db_path) - cursor = conn.execute("SELECT version FROM _meta") - assert cursor.fetchone()[0] == 1 - - # Check the migration was applied - cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='custom'") - assert cursor.fetchone() is not None - - conn.close() + # Create custom directories + migrations_dir = tmp_path / "custom_migrations" + db_dir = tmp_path / "custom_data" + migrations_dir.mkdir() + db_dir.mkdir() + + db_path = db_dir / "custom.db" + + # Create empty database file + conn = sqlite3.connect(db_path) + conn.close() + + # Initialize the database with _meta table + _ensure_meta_table(db_path) + + # Create a migration + with open(migrations_dir / "0001-test.sql", "w") as f: + f.write("CREATE TABLE custom (id INTEGER PRIMARY KEY);") + + # Run with explicit paths + result = subprocess.run([ + "fastmigrate_run_migrations", + "--db", db_path, + "--migrations", migrations_dir + ]) + + assert result.returncode == 0 + + # Verify migration was applied + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT version FROM _meta") + assert cursor.fetchone()[0] == 1 + + # Check the migration was applied + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='custom'") + assert cursor.fetchone() is not None + + conn.close() -def test_cli_backup_option(): +def test_cli_backup_option(tmp_path): """Test CLI with the --backup option.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir_path = Path(temp_dir) - db_path = temp_dir_path / "test.db" - migrations_path = temp_dir_path / "migrations" - migrations_path.mkdir() - - # Create a database with initial data - conn = sqlite3.connect(db_path) - conn.execute("CREATE TABLE initial (id INTEGER PRIMARY KEY, value TEXT)") - conn.execute("INSERT INTO initial (value) VALUES ('initial data')") - conn.commit() - conn.close() - - # Initialize the database with _meta table - _ensure_meta_table(db_path) - - # Create a test migration - with open(migrations_path / "0001-test.sql", "w") as f: - f.write("CREATE TABLE test (id INTEGER PRIMARY KEY);") - - # Run the backup - result = subprocess.run([ - "fastmigrate_backup_db", - "--db", db_path - ]) + db_path = tmp_path / "test.db" + migrations_path = tmp_path / "migrations" + migrations_path.mkdir() + + # Create a database with initial data + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE initial (id INTEGER PRIMARY KEY, value TEXT)") + conn.execute("INSERT INTO initial (value) VALUES ('initial data')") + conn.commit() + conn.close() + + # Initialize the database with _meta table + _ensure_meta_table(db_path) + + # Create a test migration + with open(migrations_path / "0001-test.sql", "w") as f: + f.write("CREATE TABLE test (id INTEGER PRIMARY KEY);") + + # Run the backup + result = subprocess.run([ + "fastmigrate_backup_db", + "--db", db_path + ]) - # Run the migration - result = subprocess.run([ - "fastmigrate_run_migrations", - "--db", db_path, - "--migrations", migrations_path, - ]) - - assert result.returncode == 0 - - # Check that a backup file was created - backup_files = list(temp_dir_path.glob("*.backup")) - assert len(backup_files) == 1 - backup_path = backup_files[0] - - # Verify the backup has the initial data but not the migration - conn_backup = sqlite3.connect(backup_path) - - # Should have the initial table with data - cursor = conn_backup.execute("SELECT value FROM initial") - assert cursor.fetchone()[0] == "initial data" - - # Should NOT have the test table from the migration - cursor = conn_backup.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test'") - assert cursor.fetchone() is None - - # But the original DB should have both tables - conn = sqlite3.connect(db_path) - - # Original should have the initial table - cursor = conn.execute("SELECT value FROM initial") - assert cursor.fetchone()[0] == "initial data" - - # Original should have the test table from the migration - cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test'") - assert cursor.fetchone() is not None - - conn_backup.close() - conn.close() + # Run the migration + result = subprocess.run([ + "fastmigrate_run_migrations", + "--db", db_path, + "--migrations", migrations_path, + ]) + + assert result.returncode == 0 + + # Check that a backup file was created + backup_files = list(tmp_path.glob("*.backup")) + assert len(backup_files) == 1 + backup_path = backup_files[0] + + # Verify the backup has the initial data but not the migration + conn_backup = sqlite3.connect(backup_path) + + # Should have the initial table with data + cursor = conn_backup.execute("SELECT value FROM initial") + assert cursor.fetchone()[0] == "initial data" + + # Should NOT have the test table from the migration + cursor = conn_backup.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test'") + assert cursor.fetchone() is None + + # But the original DB should have both tables + conn = sqlite3.connect(db_path) + + # Original should have the initial table + cursor = conn.execute("SELECT value FROM initial") + assert cursor.fetchone()[0] == "initial data" + + # Original should have the test table from the migration + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test'") + assert cursor.fetchone() is not None + + conn_backup.close() + conn.close() -def test_cli_config_file(): +def test_cli_config_file(tmp_path): """Test CLI with configuration from file.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir = Path(temp_dir) - # Create custom directories - migrations_dir = temp_dir / "custom_migrations" - db_dir = temp_dir / "custom_data" - migrations_dir.mkdir() - db_dir.mkdir() - - db_path = db_dir / "custom.db" - config_path = temp_dir / "custom.ini" - - # Create empty database file - conn = sqlite3.connect(db_path) - conn.close() - - # Initialize the database with _meta table - _ensure_meta_table(db_path) - - # Create a migration - (migrations_dir / "0001-test.sql").write_text("CREATE TABLE custom_config (id INTEGER PRIMARY KEY);") - - # Create a config file - config_path.write_text(f"[paths]\ndb = {db_path}\nmigrations = {migrations_dir}") + # Create custom directories + migrations_dir = tmp_path / "custom_migrations" + db_dir = tmp_path / "custom_data" + migrations_dir.mkdir() + db_dir.mkdir() + + db_path = db_dir / "custom.db" + config_path = tmp_path / "custom.ini" + + # Create empty database file + conn = sqlite3.connect(db_path) + conn.close() + + # Initialize the database with _meta table + _ensure_meta_table(db_path) + + # Create a migration + (migrations_dir / "0001-test.sql").write_text("CREATE TABLE custom_config (id INTEGER PRIMARY KEY);") + + # Create a config file + config_path.write_text(f"[paths]\ndb = {db_path}\nmigrations = {migrations_dir}") - # Run with config file - result = subprocess.run(["fastmigrate_run_migrations", "--config", config_path]) + # Run with config file + result = subprocess.run(["fastmigrate_run_migrations", "--config", config_path]) - assert result.returncode == 0 - - # Verify migration was applied - conn = sqlite3.connect(db_path) - cursor = conn.execute("SELECT version FROM _meta") - assert cursor.fetchone()[0] == 1 - - # Check the migration was applied - cursor = conn.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='custom_config'" - ) - assert cursor.fetchone() is not None - - conn.close() + assert result.returncode == 0 + + # Verify migration was applied + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT version FROM _meta") + assert cursor.fetchone()[0] == 1 + + # Check the migration was applied + cursor = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='custom_config'" + ) + assert cursor.fetchone() is not None + + conn.close() -def test_cli_precedence(): +def test_cli_precedence(tmp_path): """Test that CLI arguments take precedence over config file.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir_path = Path(temp_dir) - - # Create multiple directories to test precedence - migrations_config = temp_dir_path / "config_migrations" - migrations_cli = temp_dir_path / "cli_migrations" - db_dir_config = temp_dir_path / "config_db_dir" - db_dir_cli = temp_dir_path / "cli_db_dir" - - migrations_config.mkdir() - migrations_cli.mkdir() - db_dir_config.mkdir() - db_dir_cli.mkdir() - - db_path_config = db_dir_config / "config.db" - db_path_cli = db_dir_cli / "cli.db" - config_path = temp_dir_path / "precedence.ini" - - # Create empty database files - for db in [db_path_config, db_path_cli]: - conn = sqlite3.connect(db) - conn.close() - # Initialize the database with _meta table - _ensure_meta_table(db) - - # Create different migrations in each directory - with open(migrations_config / "0001-config.sql", "w") as f: - f.write("CREATE TABLE config_table (id INTEGER PRIMARY KEY);") - - with open(migrations_cli / "0001-cli.sql", "w") as f: - f.write("CREATE TABLE cli_table (id INTEGER PRIMARY KEY);") - - # Create a config file with specific paths - with open(config_path, "w") as f: - f.write(f"[paths]\ndb = {db_path_config}\nmigrations = {migrations_config}") - - # Run with BOTH config file AND explicit CLI args - # CLI args should take precedence - result = subprocess.run([ - "fastmigrate_run_migrations", - "--config", config_path, - "--db", db_path_cli, - "--migrations", migrations_cli - ]) - - assert result.returncode == 0 - - # Verify migration was applied to the CLI database, not the config one - # Config DB should be untouched - conn_config = sqlite3.connect(db_path_config) - cursor = conn_config.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='config_table'") - assert cursor.fetchone() is None, "Config DB should not have config_table" - conn_config.close() - - # CLI DB should have the CLI migration applied - conn_cli = sqlite3.connect(db_path_cli) - cursor = conn_cli.execute("SELECT version FROM _meta") - assert cursor.fetchone()[0] == 1, "CLI DB should have version 1" - - cursor = conn_cli.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='cli_table'") - assert cursor.fetchone() is not None, "CLI DB should have cli_table" - - cursor = conn_cli.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='config_table'") - assert cursor.fetchone() is None, "CLI DB should not have config_table" - - conn_cli.close() - + # Create multiple directories to test precedence + migrations_config = tmp_path / "config_migrations" + migrations_cli = tmp_path / "cli_migrations" + db_dir_config = tmp_path / "config_db_dir" + db_dir_cli = tmp_path / "cli_db_dir" + + migrations_config.mkdir() + migrations_cli.mkdir() + db_dir_config.mkdir() + db_dir_cli.mkdir() + + db_path_config = db_dir_config / "config.db" + db_path_cli = db_dir_cli / "cli.db" + config_path = tmp_path / "precedence.ini" + + # Create empty database files + for db in [db_path_config, db_path_cli]: + conn = sqlite3.connect(db) + conn.close() + # Initialize the database with _meta table + _ensure_meta_table(db) + + # Create different migrations in each directory + with open(migrations_config / "0001-config.sql", "w") as f: + f.write("CREATE TABLE config_table (id INTEGER PRIMARY KEY);") + + with open(migrations_cli / "0001-cli.sql", "w") as f: + f.write("CREATE TABLE cli_table (id INTEGER PRIMARY KEY);") + + # Create a config file with specific paths + with open(config_path, "w") as f: + f.write(f"[paths]\ndb = {db_path_config}\nmigrations = {migrations_config}") + + # Run with BOTH config file AND explicit CLI args + # CLI args should take precedence + result = subprocess.run([ + "fastmigrate_run_migrations", + "--config", config_path, + "--db", db_path_cli, + "--migrations", migrations_cli + ]) + + assert result.returncode == 0 + + # Verify migration was applied to the CLI database, not the config one + # Config DB should be untouched + conn_config = sqlite3.connect(db_path_config) + cursor = conn_config.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='config_table'") + assert cursor.fetchone() is None, "Config DB should not have config_table" + conn_config.close() + + # CLI DB should have the CLI migration applied + conn_cli = sqlite3.connect(db_path_cli) + cursor = conn_cli.execute("SELECT version FROM _meta") + assert cursor.fetchone()[0] == 1, "CLI DB should have version 1" + + cursor = conn_cli.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='cli_table'") + assert cursor.fetchone() is not None, "CLI DB should have cli_table" + + cursor = conn_cli.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='config_table'") + assert cursor.fetchone() is None, "CLI DB should not have config_table" + + conn_cli.close() + -def test_cli_createdb_flag(): +def test_cli_createdb_flag(tmp_path): """Test the --create_db flag properly initializes a database with _meta table.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir_path = Path(temp_dir) - db_path = temp_dir_path / "new_db.db" - - # Verify the database doesn't exist yet - assert not db_path.exists() - - # Run the CLI with just the --create_db flag - result = subprocess.run([ - "fastmigrate_create_db", - "--db", db_path, - ]) - - assert result.returncode == 0 - - # Verify database was created - assert db_path.exists() - - # Verify the _meta table exists with version 0 - conn = sqlite3.connect(db_path) - cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") - assert cursor.fetchone() is not None - - cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") - assert cursor.fetchone()[0] == 0 - - conn.close() + db_path = tmp_path / "new_db.db" + + # Verify the database doesn't exist yet + assert not db_path.exists() + + # Run the CLI with just the --create_db flag + result = subprocess.run([ + "fastmigrate_create_db", + "--db", db_path, + ]) + + assert result.returncode == 0 + + # Verify database was created + assert db_path.exists() + + # Verify the _meta table exists with version 0 + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") + assert cursor.fetchone() is not None + + cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") + assert cursor.fetchone()[0] == 0 + + conn.close() -def test_check_db_version_option(): +def test_check_db_version_option(tmp_path): """Test the --check_db_version option correctly reports the database version.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir_path = Path(temp_dir) - db_path = temp_dir_path / "test.db" - - # Create database file with version 42 - conn = sqlite3.connect(db_path) - conn.close() - _ensure_meta_table(db_path) - _set_db_version(db_path, 42) - - # Test with versioned database - result = subprocess.run([ - "fastmigrate_check_version", - "--db", db_path - ], capture_output=True, text=True) - - assert result.returncode == 0 - assert "Database version: 42" in result.stdout - - # Create unversioned database - unversioned_db = temp_dir_path / "unversioned.db" - conn = sqlite3.connect(unversioned_db) - conn.close() - - # Test with unversioned database - result = subprocess.run([ - "fastmigrate_check_version", - "--db", unversioned_db, - ], capture_output=True, text=True) - - assert result.returncode == 0 - assert "unversioned" in result.stdout.lower() - - # Test with non-existent database - nonexistent_db = temp_dir_path / "nonexistent.db" - result = subprocess.run([ - "fastmigrate_check_version", - "--db", nonexistent_db, - ], capture_output=True, text=True) - - assert result.returncode == 1 - assert "does not exist" in result.stdout + db_path = tmp_path / "test.db" + + # Create database file with version 42 + conn = sqlite3.connect(db_path) + conn.close() + _ensure_meta_table(db_path) + _set_db_version(db_path, 42) + + # Test with versioned database + result = subprocess.run([ + "fastmigrate_check_version", + "--db", db_path + ], capture_output=True, text=True) + + assert result.returncode == 0 + assert "Database version: 42" in result.stdout + + # Create unversioned database + unversioned_db = tmp_path / "unversioned.db" + conn = sqlite3.connect(unversioned_db) + conn.close() + + # Test with unversioned database + result = subprocess.run([ + "fastmigrate_check_version", + "--db", unversioned_db, + ], capture_output=True, text=True) + + assert result.returncode == 0 + assert "unversioned" in result.stdout.lower() + + # Test with non-existent database + nonexistent_db = tmp_path / "nonexistent.db" + result = subprocess.run([ + "fastmigrate_check_version", + "--db", nonexistent_db, + ], capture_output=True, text=True) + + assert result.returncode == 1 + assert "does not exist" in result.stdout -def test_cli_with_testsuite_a(): +def test_cli_with_testsuite_a(tmp_path): """Test CLI using testsuite_a.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir_path = Path(temp_dir) - db_path = temp_dir_path / "test.db" - - # Create empty database file - conn = sqlite3.connect(db_path) - conn.close() - - # Initialize the database with _meta table - _ensure_meta_table(db_path) - - # Run the CLI with explicit paths to the test suite - result = subprocess.run([ - "fastmigrate_run_migrations", - "--db", db_path, - "--migrations", CLI_MIGRATIONS_DIR / "migrations" - ], capture_output=True, text=True) - - assert result.returncode == 0 - - # Verify migrations applied - conn = sqlite3.connect(db_path) - - # Version should be 4 (all migrations applied) - cursor = conn.execute("SELECT version FROM _meta") - assert cursor.fetchone()[0] == 4 - - # Verify tables exist - tables = ["users", "posts", "tags", "post_tags"] - for table in tables: - cursor = conn.execute( - f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}'" - ) - assert cursor.fetchone() is not None - - conn.close() + db_path = tmp_path / "test.db" + + # Create empty database file + conn = sqlite3.connect(db_path) + conn.close() + + # Initialize the database with _meta table + _ensure_meta_table(db_path) + + # Run the CLI with explicit paths to the test suite + result = subprocess.run([ + "fastmigrate_run_migrations", + "--db", db_path, + "--migrations", CLI_MIGRATIONS_DIR / "migrations" + ], capture_output=True, text=True) + + assert result.returncode == 0 + + # Verify migrations applied + conn = sqlite3.connect(db_path) + + # Version should be 4 (all migrations applied) + cursor = conn.execute("SELECT version FROM _meta") + assert cursor.fetchone()[0] == 4 + + # Verify tables exist + tables = ["users", "posts", "tags", "post_tags"] + for table in tables: + cursor = conn.execute( + f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}'" + ) + assert cursor.fetchone() is not None + + conn.close() diff --git a/tests/test_core.py b/tests/test_core.py index 0589b44..c3d0ea6 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -2,7 +2,6 @@ import os import sqlite3 -import tempfile from pathlib import Path from unittest.mock import patch @@ -20,73 +19,79 @@ ) -def test_ensure_meta_table(): +def test_ensure_meta_table(tmp_path): """Test ensuring the _meta table exists.""" # Create a temp file database for testing - with tempfile.NamedTemporaryFile(suffix='.db') as temp_file: - db_path = Path(temp_file.name) - - # Call _ensure_meta_table on the path - _ensure_meta_table(db_path) - - # Connect and check results - conn = sqlite3.connect(db_path) - - # Check the table exists - cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") - assert cursor.fetchone() is not None - - # Check there's one row - cursor = conn.execute("SELECT COUNT(*) FROM _meta") - assert cursor.fetchone()[0] == 1 - - # Check the version is 0 - cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") - assert cursor.fetchone()[0] == 0 - - # Test updating the version - conn.execute("UPDATE _meta SET version = 42 WHERE id = 1") - conn.commit() - cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") - assert cursor.fetchone()[0] == 42 - - # Verify we can't insert duplicate rows due to constraint - try: - conn.execute("INSERT INTO _meta (id, version) VALUES (2, 50)") - assert False, "Should not be able to insert a row with id != 1" - except sqlite3.IntegrityError: - # This is expected - constraint should prevent any id != 1 - pass - - conn.close() + db_path = tmp_path / "test.db" + + # Create the empty database file first + conn = sqlite3.connect(db_path) + conn.close() + # Call _ensure_meta_table on the path + _ensure_meta_table(db_path) + + # Connect and check results + conn = sqlite3.connect(db_path) + + # Check the table exists + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") + assert cursor.fetchone() is not None + + # Check there's one row + cursor = conn.execute("SELECT COUNT(*) FROM _meta") + assert cursor.fetchone()[0] == 1 + + # Check the version is 0 + cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") + assert cursor.fetchone()[0] == 0 + + # Test updating the version + conn.execute("UPDATE _meta SET version = 42 WHERE id = 1") + conn.commit() + cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") + assert cursor.fetchone()[0] == 42 + + # Verify we can't insert duplicate rows due to constraint + try: + conn.execute("INSERT INTO _meta (id, version) VALUES (2, 50)") + assert False, "Should not be able to insert a row with id != 1" + except sqlite3.IntegrityError: + # This is expected - constraint should prevent any id != 1 + pass + + conn.close() + # Test with invalid path to verify exception is raised with pytest.raises(FileNotFoundError): _ensure_meta_table(Path("/nonexistent/path/to/db.db")) -def test_get_set_db_version(): # Tests the internal _set_db_version function +def test_get_set_db_version(tmp_path): # Tests the internal _set_db_version function """Test getting and setting the database version.""" # Create a temp file database for testing - with tempfile.NamedTemporaryFile(suffix='.db') as temp_file: - db_path = Path(temp_file.name) - - # Initialize the database first - _ensure_meta_table(db_path) - - # Initial version should be 0 - assert get_db_version(db_path) == 0 - - # Set and get version - _set_db_version(db_path, 42) - assert get_db_version(db_path) == 42 - - # Check that id=1 is enforced in the database - conn = sqlite3.connect(db_path) - cursor = conn.execute("SELECT id FROM _meta") - assert cursor.fetchone()[0] == 1 - conn.close() + db_path = tmp_path / "test.db" + + # Create the empty database file first + conn = sqlite3.connect(db_path) + conn.close() + # Initialize the database first + _ensure_meta_table(db_path) + + # Initial version should be 0 + assert get_db_version(db_path) == 0 + + # Set and get version + _set_db_version(db_path, 42) + assert get_db_version(db_path) == 42 + + # Check that id=1 is enforced in the database + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT id FROM _meta") + assert cursor.fetchone()[0] == 1 + conn.close() + # Test with nonexistent database to verify exceptions with pytest.raises(FileNotFoundError): get_db_version(Path("/nonexistent/path/to/db.db")) @@ -95,66 +100,63 @@ def test_get_set_db_version(): # Tests the internal _set_db_version function _set_db_version(Path("/nonexistent/path/to/db.db"), 50) -def test_ensure_versioned_db(): +def test_ensure_versioned_db(tmp_path): """Test ensuring a database is versioned.""" # Test case 1: Non-existent DB should be created and versioned - with tempfile.TemporaryDirectory() as temp_dir: - db_path = Path(temp_dir) / "new.db" - - # Verify the file doesn't exist yet - assert not os.path.exists(db_path) - - # Call create_db - should create the DB - version = create_db(db_path) - - # Check results - assert os.path.exists(db_path), "Database file should have been created" - assert version == 0, "Version should be 0 for a new database" - - # Verify the db structure directly - conn = sqlite3.connect(db_path) - cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") - assert cursor.fetchone()[0] == 0, "Version in database should be 0" - conn.close() + db_path = tmp_path / "new.db" + # Verify the file doesn't exist yet + assert not os.path.exists(db_path) + + # Call create_db - should create the DB + version = create_db(db_path) + + # Check results + assert os.path.exists(db_path), "Database file should have been created" + assert version == 0, "Version should be 0 for a new database" + + # Verify the db structure directly + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT version FROM _meta WHERE id = 1") + assert cursor.fetchone()[0] == 0, "Version in database should be 0" + conn.close() + # Test case 2: Existing versioned DB should return its version - with tempfile.NamedTemporaryFile(suffix='.db') as temp_file: - db_path = Path(temp_file.name) - - # Create a versioned database with a specific version - conn = sqlite3.connect(db_path) - conn.execute(""" - CREATE TABLE _meta ( - id INTEGER PRIMARY KEY CHECK (id = 1), - version INTEGER NOT NULL DEFAULT 0 - ) - """) - conn.execute("INSERT INTO _meta (id, version) VALUES (1, 42)") - conn.commit() - conn.close() - - # Call create_db - should detect existing version - version = create_db(db_path) - - # Check the version was detected correctly - assert version == 42, "Should return existing version (42)" + db_path_versioned = tmp_path / "versioned.db" + + # Create a versioned database with a specific version + conn = sqlite3.connect(db_path_versioned) + conn.execute(""" + CREATE TABLE _meta ( + id INTEGER PRIMARY KEY CHECK (id = 1), + version INTEGER NOT NULL DEFAULT 0 + ) + """) + conn.execute("INSERT INTO _meta (id, version) VALUES (1, 42)") + conn.commit() + conn.close() + + # Call create_db - should detect existing version + version = create_db(db_path_versioned) + # Check the version was detected correctly + assert version == 42, "Should return existing version (42)" + # Test case 3: Existing unversioned DB should raise an error - with tempfile.NamedTemporaryFile(suffix='.db') as temp_file: - db_path = Path(temp_file.name) - - # Create an unversioned database with some random table - conn = sqlite3.connect(db_path) - conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)") - conn.commit() - conn.close() - - # Call create_db - should raise a sqlite3.Error - with pytest.raises(sqlite3.Error) as excinfo: - create_db(db_path) - - # Verify error message indicates missing _meta table - assert "_meta table does not exist" in str(excinfo.value) + db_path_unversioned = tmp_path / "unversioned.db" + + # Create an unversioned database with some random table + conn = sqlite3.connect(db_path_unversioned) + conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)") + conn.commit() + conn.close() + + # Call create_db - should raise a sqlite3.Error + with pytest.raises(sqlite3.Error) as excinfo: + create_db(db_path_unversioned) + + # Verify error message indicates missing _meta table + assert "_meta table does not exist" in str(excinfo.value) def test_extract_version_from_filename(): @@ -171,69 +173,66 @@ def test_extract_version_from_filename(): assert extract_version_from_filename("0001_wrong_separator.sql") is None -def test_get_migration_scripts(): +def test_get_migration_scripts(tmp_path): """Test getting migration scripts from a directory.""" - with tempfile.TemporaryDirectory() as temp_dir: - temp_dir = Path(temp_dir) - # Create test migration files - Path(temp_dir, "0001-first.sql").touch() - Path(temp_dir, "0002-second.py").touch() - Path(temp_dir, "0005-fifth.sh").touch() - Path(temp_dir, "invalid.txt").touch() - - # Get migration scripts - scripts = get_migration_scripts(temp_dir) - - # Check we have the expected scripts - assert len(scripts) == 3 - assert 1 in scripts - assert 2 in scripts - assert 5 in scripts - assert os.path.basename(scripts[1]) == "0001-first.sql" - assert os.path.basename(scripts[2]) == "0002-second.py" - assert os.path.basename(scripts[5]) == "0005-fifth.sh" + # Create test migration files + Path(tmp_path, "0001-first.sql").touch() + Path(tmp_path, "0002-second.py").touch() + Path(tmp_path, "0005-fifth.sh").touch() + Path(tmp_path, "invalid.txt").touch() + + # Get migration scripts + scripts = get_migration_scripts(tmp_path) + + # Check we have the expected scripts + assert len(scripts) == 3 + assert 1 in scripts + assert 2 in scripts + assert 5 in scripts + assert os.path.basename(scripts[1]) == "0001-first.sql" + assert os.path.basename(scripts[2]) == "0002-second.py" + assert os.path.basename(scripts[5]) == "0005-fifth.sh" + -def test_get_migration_scripts_duplicate_version(): +def test_get_migration_scripts_duplicate_version(tmp_path): """Test that duplicate version numbers are detected.""" - with tempfile.TemporaryDirectory() as temp_dir: - # Create test migration files with duplicate version - Path(temp_dir, "0001-first.sql").touch() - Path(temp_dir, "0001-duplicate.py").touch() - - # Get migration scripts - should raise ValueError - with pytest.raises(ValueError) as excinfo: - get_migration_scripts(Path(temp_dir)) - - assert "Duplicate migration version" in str(excinfo.value) + # Create test migration files with duplicate version + Path(tmp_path, "0001-first.sql").touch() + Path(tmp_path, "0001-duplicate.py").touch() + + # Get migration scripts - should raise ValueError + with pytest.raises(ValueError) as excinfo: + get_migration_scripts(tmp_path) + + assert "Duplicate migration version" in str(excinfo.value) -def test_run_migrations_on_unversioned_db(): +def test_run_migrations_on_unversioned_db(tmp_path): """Test that run_migrations fails on an unversioned database.""" - with tempfile.TemporaryDirectory() as temp_dir: - # Create migrations directory - migrations_dir = os.path.join(temp_dir, "migrations") - os.makedirs(migrations_dir) - - # Create a simple migration - with open(os.path.join(migrations_dir, "0001-create-table.sql"), "w") as f: - f.write("CREATE TABLE test (id INTEGER PRIMARY KEY);") - - # Create a database without initializing it (no _meta table) - db_path = os.path.join(temp_dir, "test.db") - conn = sqlite3.connect(db_path) - conn.close() - - # Run migrations - should fail because the database is not versioned - assert run_migrations(db_path, migrations_dir) is False - - # Verify no table was created (migration did not run) - conn = sqlite3.connect(db_path) - cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test'") - assert cursor.fetchone() is None, "Migration should not have run on unversioned database" - - # Also verify there's no _meta table (run_migrations shouldn't create one) - cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") - assert cursor.fetchone() is None, "run_migrations should not have created a _meta table" - - conn.close() + # Create migrations directory + migrations_dir = tmp_path / "migrations" + os.makedirs(migrations_dir) + + # Create a simple migration + with open(migrations_dir / "0001-create-table.sql", "w") as f: + f.write("CREATE TABLE test (id INTEGER PRIMARY KEY);") + + # Create a database without initializing it (no _meta table) + db_path = tmp_path / "test.db" + conn = sqlite3.connect(db_path) + conn.close() + + # Run migrations - should fail because the database is not versioned + assert run_migrations(db_path, migrations_dir) is False + + # Verify no table was created (migration did not run) + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test'") + assert cursor.fetchone() is None, "Migration should not have run on unversioned database" + + # Also verify there's no _meta table (run_migrations shouldn't create one) + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") + assert cursor.fetchone() is None, "run_migrations should not have created a _meta table" + + conn.close() diff --git a/tests/test_db_backup.py b/tests/test_db_backup.py index 2dd06e8..a463a3e 100644 --- a/tests/test_db_backup.py +++ b/tests/test_db_backup.py @@ -1,7 +1,6 @@ import pytest import sqlite3 import os -import tempfile from datetime import datetime from pathlib import Path @@ -12,16 +11,15 @@ @pytest.fixture -def temp_db(): - """Provides a temporary directory and a path to a test database.""" - with tempfile.TemporaryDirectory() as temp_dir: - db_path = os.path.join(temp_dir, "test.db") - conn = sqlite3.connect(db_path) - conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)") - conn.execute("INSERT INTO test (value) VALUES ('original data')") - conn.commit() - conn.close() - yield db_path, temp_dir # Yield both db_path and temp_dir +def temp_db(tmp_path): + """Provides a test database path and temp directory path.""" + db_path = tmp_path / "test.db" + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)") + conn.execute("INSERT INTO test (value) VALUES ('original data')") + conn.commit() + conn.close() + yield db_path, tmp_path # Yield both db_path and tmp_path def test_create_db_backup_success(temp_db): @@ -31,8 +29,8 @@ def test_create_db_backup_success(temp_db): backup_path = create_db_backup(db_path) assert backup_path is not None - assert os.path.exists(backup_path) - assert str(backup_path).startswith(db_path) + assert backup_path.exists() + assert str(backup_path).startswith(str(db_path)) assert ".backup" in os.path.basename(backup_path) # Check basename for .backup # Verify the backup contains the same data @@ -45,7 +43,7 @@ def test_create_db_backup_success(temp_db): def test_create_db_backup_db_not_exists(temp_db): """Test backup attempt when the source database does not exist.""" _, temp_dir = temp_db # We only need the directory from the fixture - non_existent_path = os.path.join(temp_dir, "nonexistent.db") + non_existent_path = temp_dir / "nonexistent.db" result = create_db_backup(non_existent_path)