diff options
| -rw-r--r-- | migrations/auth/20221114_02_DKKjn-drop-generic-role-tables.py | 39 | ||||
| -rw-r--r-- | tests/unit/auth/test_migrations_drop_tables.py | 64 |
2 files changed, 103 insertions, 0 deletions
diff --git a/migrations/auth/20221114_02_DKKjn-drop-generic-role-tables.py b/migrations/auth/20221114_02_DKKjn-drop-generic-role-tables.py new file mode 100644 index 0000000..e767aeb --- /dev/null +++ b/migrations/auth/20221114_02_DKKjn-drop-generic-role-tables.py @@ -0,0 +1,39 @@ +""" +Drop 'generic_role*' tables +""" + +from yoyo import step + +__depends__ = {'20221114_01_n8gsF-create-generic-role-privileges-table'} + +steps = [ + step( + """ + DROP INDEX IF EXISTS + idx_tbl_generic_role_privileges_cols_generic_role_id + """, + """ + CREATE INDEX IF NOT EXISTS + idx_tbl_generic_role_privileges_cols_generic_role_id + ON generic_role_privileges(generic_role_id) + """), + step( + "DROP TABLE IF EXISTS generic_role_privileges", + """ + CREATE TABLE IF NOT EXISTS generic_role_privileges( + generic_role_id TEXT NOT NULL, + privilege_id TEXT NOT NULL, + PRIMARY KEY(generic_role_id, privilege_id), + FOREIGN KEY(generic_role_id) REFERENCES generic_roles(role_id), + FOREIGN KEY(privilege_id) REFERENCES privileges(privilege_id) + ) WITHOUT ROWID + """), + step( + "DROP TABLE IF EXISTS generic_roles", + """ + CREATE TABLE IF NOT EXISTS generic_roles( + role_id TEXT PRIMARY KEY, + role_name TEXT NOT NULL + ) WITHOUT ROWID + """) +] diff --git a/tests/unit/auth/test_migrations_drop_tables.py b/tests/unit/auth/test_migrations_drop_tables.py new file mode 100644 index 0000000..dfab3f6 --- /dev/null +++ b/tests/unit/auth/test_migrations_drop_tables.py @@ -0,0 +1,64 @@ +"""Test migrations that create tables""" +from contextlib import closing + +import pytest +import sqlite3 + +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) + +test_params = ( + ("20221114_02_DKKjn-drop-generic-role-tables.py", "generic_roles"), + ("20221114_02_DKKjn-drop-generic-role-tables.py", "generic_role_privileges")) + +@pytest.mark.unit_test +@pytest.mark.parametrize("migration_file,the_table", test_params) +def test_drop_table( + auth_testdb_path, auth_migrations_dir, backend, all_migrations, + migration_file, the_table): + """ + GIVEN: A database migration script to create table, `the_table` + WHEN: The migration is applied + THEN: Ensure that the table `the_table` is created + """ + migration_path=f"{auth_migrations_dir}/{migration_file}" + 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("SELECT name FROM sqlite_schema WHERE type='table'") + result_before_migration = cursor.fetchall() + apply_single_migration(backend, the_migration) + cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") + result_after_migration = cursor.fetchall() + + rollback_migrations(backend, older_migrations + [the_migration]) + assert the_table in [row[0] for row in result_before_migration] + assert the_table not in [row[0] for row in result_after_migration] + +@pytest.mark.unit_test +@pytest.mark.parametrize("migration_file,the_table", test_params) +def test_rollback_drop_table( + auth_testdb_path, auth_migrations_dir, backend, migration_file, + the_table): + """ + GIVEN: A database migration script to create the table `the_table` + WHEN: The migration is rolled back + THEN: Ensure that the table `the_table` no longer exists + """ + migration_path=f"{auth_migrations_dir}/{migration_file}" + 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: + apply_single_migration(backend, the_migration) + cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") + result_after_migration = cursor.fetchall() + rollback_single_migration(backend, the_migration) + cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") + result_after_rollback = cursor.fetchall() + + rollback_migrations(backend, older_migrations) + assert the_table not in [row[0] for row in result_after_migration] + assert the_table in [row[0] for row in result_after_rollback] |