diff options
-rw-r--r-- | tests/unit/auth/conftest.py | 16 | ||||
-rw-r--r-- | tests/unit/auth/test_credentials.py | 35 | ||||
-rw-r--r-- | tests/unit/auth/test_groups.py | 10 | ||||
-rw-r--r-- | tests/unit/auth/test_migrations_add_remove_columns.py | 42 | ||||
-rw-r--r-- | tests/unit/auth/test_migrations_create_tables.py | 12 | ||||
-rw-r--r-- | tests/unit/auth/test_migrations_drop_tables.py | 9 | ||||
-rw-r--r-- | tests/unit/auth/test_migrations_indexes.py | 25 | ||||
-rw-r--r-- | tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py | 28 | ||||
-rw-r--r-- | tests/unit/auth/test_migrations_insert_data_into_empty_table.py | 9 |
9 files changed, 101 insertions, 85 deletions
diff --git a/tests/unit/auth/conftest.py b/tests/unit/auth/conftest.py index a49affe..0c6b7c7 100644 --- a/tests/unit/auth/conftest.py +++ b/tests/unit/auth/conftest.py @@ -17,20 +17,22 @@ def auth_migrations_dir(test_app_config): # pylint: disable=redefined-outer-name """Get the test application's auth database file""" return test_app_config["AUTH_MIGRATIONS"] -def apply_single_migration(backend: DatabaseBackend, migration: Migration): +def apply_single_migration(backend: DatabaseBackend, migration: Migration):# pylint: disable=[redefined-outer-name] """Utility to apply a single migration""" apply_migrations(backend, MigrationList([migration])) -def rollback_single_migration(backend: DatabaseBackend, migration: Migration): +def rollback_single_migration(backend: DatabaseBackend, migration: Migration):# pylint: disable=[redefined-outer-name] """Utility to rollback a single migration""" rollback_migrations(backend, MigrationList([migration])) @pytest.fixture(scope="session") -def backend(auth_testdb_path): # pylint: disable=redefined-outer-name +def backend(auth_testdb_path):# pylint: disable=redefined-outer-name + """Fixture: retrieve yoyo backend for auth database""" return get_backend(f"sqlite:///{auth_testdb_path}") @pytest.fixture(scope="session") def all_migrations(auth_migrations_dir): # pylint: disable=redefined-outer-name + """Retrieve all the migrations""" return read_migrations(auth_migrations_dir) @pytest.fixture(scope="function") @@ -43,15 +45,17 @@ def conn_after_auth_migrations(backend, auth_testdb_path, all_migrations): # pyl rollback_migrations(backend, all_migrations) def migrations_up_to(migration, migrations_dir): + """Run all the migration before `migration`.""" migrations = read_migrations(migrations_dir) index = [mig.path for mig in migrations].index(migration) return MigrationList(migrations[0:index]) @pytest.fixture(scope="function") -def test_users(conn_after_auth_migrations): +def test_users(conn_after_auth_migrations):# pylint: disable=[redefined-outer-name] + """Fixture: setup test users.""" query = "INSERT INTO users(user_id, email, name) VALUES (?, ?, ?)" query_user_roles = "INSERT INTO user_roles(user_id, role_id) VALUES (?, ?)" - test_users = ( + the_users = ( ("ecb52977-3004-469e-9428-2a1856725c7f", "group@lead.er", "Group Leader"), ("21351b66-8aad-475b-84ac-53ce528451e3", "group@mem.ber01", @@ -64,7 +68,7 @@ def test_users(conn_after_auth_migrations): ("ecb52977-3004-469e-9428-2a1856725c7f", "a0e67630-d502-4b9f-b23f-6805d0f30e30"),) with db.cursor(conn_after_auth_migrations) as cursor: - cursor.executemany(query, test_users) + cursor.executemany(query, the_users) cursor.executemany(query_user_roles, test_user_roles) yield conn_after_auth_migrations diff --git a/tests/unit/auth/test_credentials.py b/tests/unit/auth/test_credentials.py index 1425eb2..f2a3d25 100644 --- a/tests/unit/auth/test_credentials.py +++ b/tests/unit/auth/test_credentials.py @@ -1,30 +1,36 @@ """Test the credentials checks""" -import sqlite3 - import pytest -from contextlib import closing from yoyo.migrations import MigrationList from hypothesis import given, settings, strategies, HealthCheck -from tests.unit.auth.conftest import migrations_up_to +from gn3.auth import db from gn3.auth.authentication import credentials_in_database from gn3.migrations import get_migration, apply_migrations, rollback_migrations +from tests.unit.auth.conftest import migrations_up_to + @pytest.fixture def with_credentials_table(backend, auth_testdb_path): + """ + Fixture: Yield a connection object with the 'user_credentials' table + created. + """ migrations_dir = "migrations/auth" migration = f"{migrations_dir}/20221103_02_sGrIs-create-user-credentials-table.py" migrations = (migrations_up_to(migration, migrations_dir) + MigrationList([get_migration(migration)])) apply_migrations(backend, migrations) - with closing(sqlite3.connect(auth_testdb_path)) as conn: + with db.connection(auth_testdb_path) as conn: yield conn rollback_migrations(backend, migrations) @pytest.fixture -def with_credentials(with_credentials_table): - with closing(with_credentials_table.cursor()) as cursor: +def with_credentials(with_credentials_table):# pylint: disable=redefined-outer-name + """ + Fixture: Initialise the database with some user credentials. + """ + with db.cursor(with_credentials_table) as cursor: cursor.executemany( "INSERT INTO users VALUES (:user_id, :email, :name)", ({"user_id": "82552014-21ee-4321-b96a-b8788b97b862", @@ -56,14 +62,13 @@ def with_credentials(with_credentials_table): @pytest.mark.unit_test @given(strategies.emails(), strategies.text()) @settings(suppress_health_check=[HealthCheck.function_scoped_fixture]) -def test_credentials_not_in_database(with_credentials, email, password): +def test_credentials_not_in_database(with_credentials, email, password):# pylint: disable=redefined-outer-name """ GIVEN: credentials that do not exist in the database WHEN: the `credentials_in_database` function is run against the credentials THEN: check that the function returns false in all cases. """ - with closing(with_credentials.cursor()) as cursor: - results = credentials_in_database(cursor, email, password) + with db.cursor(with_credentials) as cursor: assert credentials_in_database(cursor, email, password) is False @pytest.mark.unit_test @@ -71,14 +76,13 @@ def test_credentials_not_in_database(with_credentials, email, password): "email,password", (("first@test.user", "wrongpassword"), ("first@tes.user", "testuser01"))) -def test_partially_wrong_credentials(with_credentials, email, password): +def test_partially_wrong_credentials(with_credentials, email, password):# pylint: disable=redefined-outer-name """ GIVEN: credentials that exist in the database WHEN: the credentials are checked with partially wrong values THEN: the check fails since the credentials are not correct """ - with closing(with_credentials.cursor()) as cursor: - results = credentials_in_database(cursor, email, password) + with db.cursor(with_credentials) as cursor: assert credentials_in_database(cursor, email, password) is False @pytest.mark.unit_test @@ -86,12 +90,11 @@ def test_partially_wrong_credentials(with_credentials, email, password): "email,password", (("first@test.user", "testuser01"), ("second@test.user", "testuser02"))) -def test_partially_correct_credentials(with_credentials, email, password): +def test_partially_correct_credentials(with_credentials, email, password):# pylint: disable=redefined-outer-name """ GIVEN: credentials that exist in the database WHEN: the credentials are checked with correct values THEN: the check passes """ - with closing(with_credentials.cursor()) as cursor: - results = credentials_in_database(cursor, email, password) + with db.cursor(with_credentials) as cursor: assert credentials_in_database(cursor, email, password) is True diff --git a/tests/unit/auth/test_groups.py b/tests/unit/auth/test_groups.py index ac575d0..d83431e 100644 --- a/tests/unit/auth/test_groups.py +++ b/tests/unit/auth/test_groups.py @@ -1,7 +1,6 @@ """Test functions dealing with group management.""" from uuid import UUID -import flask import pytest from gn3.auth import db @@ -26,7 +25,14 @@ group_leader_id = lambda : UUID("d32611e3-07fc-4564-b56c-786c6db6de2b") ("ae9c6245-0966-41a5-9a5e-20885a96bea7", create_group_failure), ("9a0c7ce5-2f40-4e78-979e-bf3527a59579", create_group_failure), ("e614247d-84d2-491d-a048-f80b578216cb", create_group_failure))) -def test_create_group(test_app, auth_testdb_path, mocker, test_users, user_id, expected): +def test_create_group(# pylint: disable=[too-many-arguments] + test_app, auth_testdb_path, mocker, test_users, user_id, expected):# pylint: disable=[unused-argument] + """ + GIVEN: an authenticated user + WHEN: the user attempts to create a group + THEN: verify they are only able to create the group if they have the + appropriate privileges + """ mocker.patch("gn3.auth.authorisation.groups.uuid.uuid4", group_leader_id) with test_app.app_context() as flask_context: flask_context.g.user_id = UUID(user_id) 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]) diff --git a/tests/unit/auth/test_migrations_create_tables.py b/tests/unit/auth/test_migrations_create_tables.py index 2d690da..cf576b6 100644 --- a/tests/unit/auth/test_migrations_create_tables.py +++ b/tests/unit/auth/test_migrations_create_tables.py @@ -1,9 +1,7 @@ """Test migrations that create tables""" -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) @@ -30,8 +28,8 @@ migrations_and_tables = ( @pytest.mark.unit_test @pytest.mark.parametrize("migration_file,the_table", migrations_and_tables) def test_create_table( - auth_testdb_path, auth_migrations_dir, backend, all_migrations, - migration_file, the_table): + auth_testdb_path, auth_migrations_dir, backend, migration_file, + the_table): """ GIVEN: A database migration script to create table, `the_table` WHEN: The migration is applied @@ -40,7 +38,7 @@ def test_create_table( migration_path=f"{auth_migrations_dir}/{migration_file}" older_migrations = migrations_up_to(migration_path, auth_migrations_dir) apply_migrations(backend, older_migrations) - with closing(sqlite3.connect(auth_testdb_path)) as conn, closing(conn.cursor()) as cursor: + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") result_before_migration = cursor.fetchall() apply_single_migration(backend, get_migration(migration_path)) @@ -64,7 +62,7 @@ def test_rollback_create_table( migration_path=f"{auth_migrations_dir}/{migration_file}" older_migrations = migrations_up_to(migration_path, auth_migrations_dir) apply_migrations(backend, older_migrations) - with closing(sqlite3.connect(auth_testdb_path)) as conn, closing(conn.cursor()) as cursor: + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: apply_single_migration(backend, get_migration(migration_path)) cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") result_after_migration = cursor.fetchall() diff --git a/tests/unit/auth/test_migrations_drop_tables.py b/tests/unit/auth/test_migrations_drop_tables.py index dfab3f6..2362c77 100644 --- a/tests/unit/auth/test_migrations_drop_tables.py +++ b/tests/unit/auth/test_migrations_drop_tables.py @@ -1,9 +1,8 @@ """Test migrations that create tables""" -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) @@ -15,7 +14,7 @@ test_params = ( @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, + auth_testdb_path, auth_migrations_dir, backend, migration_file, the_table): """ GIVEN: A database migration script to create table, `the_table` @@ -26,7 +25,7 @@ def test_drop_table( 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: + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") result_before_migration = cursor.fetchall() apply_single_migration(backend, the_migration) @@ -51,7 +50,7 @@ def test_rollback_drop_table( 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: + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: apply_single_migration(backend, the_migration) cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'") result_after_migration = cursor.fetchall() diff --git a/tests/unit/auth/test_migrations_indexes.py b/tests/unit/auth/test_migrations_indexes.py index dba0a98..a346164 100644 --- a/tests/unit/auth/test_migrations_indexes.py +++ b/tests/unit/auth/test_migrations_indexes.py @@ -1,15 +1,12 @@ """Test that indexes are created and removed.""" - -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 = """ +QUERY = """ SELECT name FROM sqlite_master WHERE type='index' AND tbl_name = ? AND name= ? """ @@ -28,7 +25,7 @@ migrations_tables_and_indexes = ( @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,the_table,the_index", migrations_tables_and_indexes) -def test_index_created( +def test_index_created(# pylint: disable=[too-many-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, the_table, the_index): """ @@ -41,11 +38,11 @@ def test_index_created( the_migration = get_migration(migration_path) query_params = (the_table, the_index) apply_migrations(backend, older_migrations) - with closing(sqlite3.connect(auth_testdb_path)) as conn, closing(conn.cursor()) as cursor: - cursor.execute(query, query_params) + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: + cursor.execute(QUERY, query_params) result_before_migration = cursor.fetchall() apply_single_migration(backend, the_migration) - cursor.execute(query, query_params) + cursor.execute(QUERY, query_params) result_after_migration = cursor.fetchall() rollback_migrations(backend, older_migrations + [the_migration]) @@ -59,7 +56,7 @@ def test_index_created( @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,the_table,the_index", migrations_tables_and_indexes) -def test_index_dropped( +def test_index_dropped(# pylint: disable=[too-many-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, the_table, the_index): """ @@ -72,14 +69,14 @@ def test_index_dropped( the_migration = get_migration(migration_path) query_params = (the_table, the_index) apply_migrations(backend, older_migrations) - with closing(sqlite3.connect(auth_testdb_path)) as conn, closing(conn.cursor()) as cursor: - cursor.execute(query, query_params) + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: + cursor.execute(QUERY, query_params) result_before_migration = cursor.fetchall() apply_single_migration(backend, the_migration) - cursor.execute(query, query_params) + cursor.execute(QUERY, query_params) result_after_migration = cursor.fetchall() rollback_single_migration(backend, the_migration) - cursor.execute(query, query_params) + cursor.execute(QUERY, query_params) result_after_rollback = cursor.fetchall() rollback_migrations(backend, older_migrations) diff --git a/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py b/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py index 8c82c53..0e78823 100644 --- a/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py +++ b/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py @@ -1,23 +1,26 @@ """ Test that the `resource_categories` table is initialised with the startup data. """ -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) -migration_path = "migrations/auth/20221108_04_CKcSL-init-data-in-resource-categories-table.py" +MIGRATION_PATH = "migrations/auth/20221108_04_CKcSL-init-data-in-resource-categories-table.py" @pytest.mark.unit_test def test_apply_init_data(auth_testdb_path, auth_migrations_dir, backend): - older_migrations = migrations_up_to(migration_path, auth_migrations_dir) - the_migration = get_migration(migration_path) + """ + GIVEN: A migration script + WHEN: The migration is applied + THEN: Verify that the expected data exists in the table + """ + 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: + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: cursor.execute("SELECT * FROM resource_categories") assert len(cursor.fetchall()) == 0, "Expected empty table." apply_single_migration(backend, the_migration) @@ -35,10 +38,15 @@ def test_apply_init_data(auth_testdb_path, auth_migrations_dir, backend): @pytest.mark.unit_test def test_rollback_init_data(auth_testdb_path, auth_migrations_dir, backend): - older_migrations = migrations_up_to(migration_path, auth_migrations_dir) - the_migration = get_migration(migration_path) + """ + GIVEN: A migration script + WHEN: The migration is rolled back + THEN: Verify that the table is empty + """ + 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: + with db.connection(auth_testdb_path) as conn, db.cursor(conn) as cursor: cursor.execute("SELECT * FROM resource_categories") assert len(cursor.fetchall()) == 0, "Expected empty table." apply_single_migration(backend, the_migration) diff --git a/tests/unit/auth/test_migrations_insert_data_into_empty_table.py b/tests/unit/auth/test_migrations_insert_data_into_empty_table.py index 73d654a..b4845a0 100644 --- a/tests/unit/auth/test_migrations_insert_data_into_empty_table.py +++ b/tests/unit/auth/test_migrations_insert_data_into_empty_table.py @@ -1,9 +1,8 @@ """Test data insertion when migrations are run.""" - +import sqlite3 from contextlib import closing import pytest -import sqlite3 from gn3.migrations import get_migration, apply_migrations, rollback_migrations from tests.unit.auth.conftest import ( @@ -17,7 +16,7 @@ test_params = ( @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,table,row_count", test_params) -def test_apply_insert( +def test_apply_insert(# pylint: disable=[too-many-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, table, row_count): """ @@ -46,7 +45,7 @@ def test_apply_insert( @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,table,row_count", test_params) -def test_rollback_insert( +def test_rollback_insert(# pylint: disable=[too-many-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, table, row_count): """ @@ -75,4 +74,4 @@ def test_rollback_insert( assert result_after_migration[0][0] == row_count, ( f"Expected {row_count} rows") assert result_after_rollback[0][0] == 0, ( - f"Expected empty table after rollback") + "Expected empty table after rollback") |