From 62b8596616ea46849e29ca77a77e1196417dc1f9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pave=C5=82=20Ty=C5=9Blacki?= <pavel.tyslacki@gmail.com>
Date: Wed, 16 Jan 2019 00:14:20 +0300
Subject: [PATCH] Made test table cleanup in OperationTestBase more robust.
Some non-unique constraint names were added in
b69f8eb04cc8762d3dfd5af5ea1fc58e3f2ebcc3 which resulted in failures
depending on the order in which tests were run.
---
tests/migrations/test_multidb.py | 18 +++++++++-----
tests/migrations/test_operations.py | 38 ++++++++++++++---------------
2 files changed, 30 insertions(+), 26 deletions(-)
diff --git a/tests/migrations/test_multidb.py b/tests/migrations/test_multidb.py
index b2c4320ad3..e0c5a4d3c5 100644
--- a/tests/migrations/test_multidb.py
+++ b/tests/migrations/test_multidb.py
@@ -122,13 +122,16 @@ class MultiDBOperationTests(OperationTestBase):
self.assertEqual(Pony.objects.count(), 0)
@override_settings(DATABASE_ROUTERS=[MigrateNothingRouter()])
- def test_run_sql(self):
+ def test_run_sql_migrate_nothing_router(self):
self._test_run_sql("test_mltdb_runsql", should_run=False)
@override_settings(DATABASE_ROUTERS=[MigrateWhenFooRouter()])
- def test_run_sql2(self):
+ def test_run_sql_migrate_foo_router_without_hints(self):
self._test_run_sql("test_mltdb_runsql2", should_run=False)
- self._test_run_sql("test_mltdb_runsql2", should_run=True, hints={'foo': True})
+
+ @override_settings(DATABASE_ROUTERS=[MigrateWhenFooRouter()])
+ def test_run_sql_migrate_foo_router_with_hints(self):
+ self._test_run_sql('test_mltdb_runsql3', should_run=True, hints={'foo': True})
def _test_run_python(self, app_label, should_run, hints=None):
with override_settings(DATABASE_ROUTERS=[MigrateEverythingRouter()]):
@@ -156,10 +159,13 @@ class MultiDBOperationTests(OperationTestBase):
self.assertEqual(Pony.objects.count(), 0)
@override_settings(DATABASE_ROUTERS=[MigrateNothingRouter()])
- def test_run_python(self):
+ def test_run_python_migrate_nothing_router(self):
self._test_run_python("test_mltdb_runpython", should_run=False)
@override_settings(DATABASE_ROUTERS=[MigrateWhenFooRouter()])
- def test_run_python2(self):
+ def test_run_python_migrate_foo_router_without_hints(self):
self._test_run_python("test_mltdb_runpython2", should_run=False)
- self._test_run_python("test_mltdb_runpython2", should_run=True, hints={'foo': True})
+
+ @override_settings(DATABASE_ROUTERS=[MigrateWhenFooRouter()])
+ def test_run_python_migrate_foo_router_with_hints(self):
+ self._test_run_python('test_mltdb_runpython3', should_run=True, hints={'foo': True})
diff --git a/tests/migrations/test_operations.py b/tests/migrations/test_operations.py
index e6d0c4bce4..d5c8396596 100644
--- a/tests/migrations/test_operations.py
+++ b/tests/migrations/test_operations.py
@@ -22,6 +22,24 @@ class OperationTestBase(MigrationTestBase):
Common functions to help test operations.
"""
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls._initial_table_names = frozenset(connection.introspection.table_names())
+
+ def tearDown(self):
+ self.cleanup_test_tables()
+ super().tearDown()
+
+ def cleanup_test_tables(self):
+ table_names = frozenset(connection.introspection.table_names()) - self._initial_table_names
+ with connection.schema_editor() as editor:
+ with connection.constraint_checks_disabled():
+ for table_name in table_names:
+ editor.execute(editor.sql_delete_table % {
+ 'table': editor.quote_name(table_name),
+ })
+
def apply_operations(self, app_label, project_state, operations, atomic=True):
migration = Migration('name', app_label)
migration.operations = operations
@@ -51,26 +69,6 @@ class OperationTestBase(MigrationTestBase):
"""
Creates a test model state and database table.
"""
- # Delete the tables if they already exist
- table_names = [
- # Start with ManyToMany tables
- '_pony_stables', '_pony_vans',
- # Then standard model tables
- '_pony', '_stable', '_van',
- ]
- tables = [(app_label + table_name) for table_name in table_names]
- with connection.cursor() as cursor:
- table_names = connection.introspection.table_names(cursor)
- connection.disable_constraint_checking()
- sql_delete_table = connection.schema_editor().sql_delete_table
- with transaction.atomic():
- for table in tables:
- if table in table_names:
- cursor.execute(sql_delete_table % {
- "table": connection.ops.quote_name(table),
- })
- connection.enable_constraint_checking()
-
# Make the "current" state
model_options = {
"swappable": "TEST_SWAP_MODEL",