diff options
Diffstat (limited to 'tests/unit/auth/test_migrations_add_remove_columns.py')
| -rw-r--r-- | tests/unit/auth/test_migrations_add_remove_columns.py | 42 |
1 files changed, 22 insertions, 20 deletions
diff --git a/tests/unit/auth/test_migrations_add_remove_columns.py b/tests/unit/auth/test_migrations_add_remove_columns.py index b43e1cf..89948b3 100644 --- a/tests/unit/auth/test_migrations_add_remove_columns.py +++ b/tests/unit/auth/test_migrations_add_remove_columns.py @@ -1,35 +1,37 @@ """Test migrations that alter tables adding/removing columns.""" - -from contextlib import closing - import pytest -import sqlite3 +from gn3.auth import db from gn3.migrations import get_migration, apply_migrations, rollback_migrations from tests.unit.auth.conftest import ( apply_single_migration, rollback_single_migration, migrations_up_to) -query = "SELECT sql FROM sqlite_schema WHERE name=?" +QUERY = "SELECT sql FROM sqlite_schema WHERE name=?" -test_params = ( +TEST_PARAMS = ( ("20221109_01_HbD5F-add-resource-meta-field-to-resource-categories-field.py", "resource_categories", "resource_meta TEXT", True), - ("20221110_08_23psB-add-privilege-category-and-privilege-description-columns-to-privileges-table.py", + (("20221110_08_23psB-add-privilege-category-and-privilege-description-" + "columns-to-privileges-table.py"), "privileges", "privilege_category TEXT", True), - ("20221110_08_23psB-add-privilege-category-and-privilege-description-columns-to-privileges-table.py", + (("20221110_08_23psB-add-privilege-category-and-privilege-description-" + "columns-to-privileges-table.py"), "privileges", "privilege_description TEXT", True)) def found(haystack: str, needle: str) -> bool: - return any([ - (line.strip().find(needle) >= 0) for line in haystack.split("\n")]) + """Check whether `needle` is found in `haystack`""" + return any( + (line.strip().find(needle) >= 0) for line in haystack.split("\n")) def pristine_before_migration(adding: bool, result_str: str, column: str) -> bool: + """Check that database is pristine before running the migration""" col_was_found = found(result_str, column) if adding: return not col_was_found return col_was_found def applied_successfully(adding: bool, result_str: str, column: str) -> bool: + """Check that the migration ran successfully""" col_was_found = found(result_str, column) if adding: return col_was_found @@ -37,8 +39,8 @@ def applied_successfully(adding: bool, result_str: str, column: str) -> bool: @pytest.mark.unit_test @pytest.mark.parametrize( - "migration_file,the_table,the_column,adding", test_params) -def test_apply_add_remove_column( + "migration_file,the_table,the_column,adding", TEST_PARAMS) +def test_apply_add_remove_column(# pylint: disable=[too-many-arguments] auth_migrations_dir, auth_testdb_path, backend, migration_file, the_table, the_column, adding): """ @@ -51,11 +53,11 @@ def test_apply_add_remove_column( older_migrations = migrations_up_to(migration_path, auth_migrations_dir) the_migration = get_migration(migration_path) apply_migrations(backend, older_migrations) - with closing(sqlite3.connect(auth_testdb_path)) as conn, closing(conn.cursor()) as cursor: - cursor.execute(query, (the_table,)) + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: + cursor.execute(QUERY, (the_table,)) results_before_migration = cursor.fetchone() apply_single_migration(backend, the_migration) - cursor.execute(query, (the_table,)) + cursor.execute(QUERY, (the_table,)) results_after_migration = cursor.fetchone() rollback_migrations(backend, older_migrations + [the_migration]) @@ -68,8 +70,8 @@ def test_apply_add_remove_column( @pytest.mark.unit_test @pytest.mark.parametrize( - "migration_file,the_table,the_column,adding", test_params) -def test_rollback_add_remove_column( + "migration_file,the_table,the_column,adding", TEST_PARAMS) +def test_rollback_add_remove_column(# pylint: disable=[too-many-arguments] auth_migrations_dir, auth_testdb_path, backend, migration_file, the_table, the_column, adding): """ @@ -83,11 +85,11 @@ def test_rollback_add_remove_column( the_migration = get_migration(migration_path) apply_migrations(backend, older_migrations) apply_single_migration(backend, the_migration) - with closing(sqlite3.connect(auth_testdb_path)) as conn, closing(conn.cursor()) as cursor: - cursor.execute(query, (the_table,)) + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: + cursor.execute(QUERY, (the_table,)) results_before_migration = cursor.fetchone() rollback_single_migration(backend, the_migration) - cursor.execute(query, (the_table,)) + cursor.execute(QUERY, (the_table,)) results_after_migration = cursor.fetchone() rollback_migrations(backend, older_migrations + [the_migration]) |