From 9136ceb6fb8225625631671147ff70c1fcfbbcdc Mon Sep 17 00:00:00 2001 From: wrwrwr Date: Sat, 29 Nov 2014 18:09:40 +0100 Subject: [PATCH] Replaced router.routers usage with override_settings(DATABASE_ROUTERS); refs #23933. --- django/contrib/gis/tests/layermap/tests.py | 12 +- tests/cache/tests.py | 42 ++-- tests/commands_sql/tests.py | 11 +- tests/contenttypes_tests/tests.py | 13 +- tests/migrations/test_operations.py | 14 +- tests/multiple_database/tests.py | 226 ++++++++------------- tests/select_for_update/tests.py | 17 +- 7 files changed, 121 insertions(+), 214 deletions(-) diff --git a/django/contrib/gis/tests/layermap/tests.py b/django/contrib/gis/tests/layermap/tests.py index b9b1404b3c..45709e30ce 100644 --- a/django/contrib/gis/tests/layermap/tests.py +++ b/django/contrib/gis/tests/layermap/tests.py @@ -8,9 +8,9 @@ import unittest from unittest import skipUnless from django.contrib.gis.gdal import HAS_GDAL -from django.db import connection, router +from django.db import connection from django.conf import settings -from django.test import TestCase, skipUnlessDBFeature +from django.test import TestCase, override_settings, skipUnlessDBFeature from django.utils._os import upath if HAS_GDAL: @@ -328,15 +328,9 @@ class OtherRouter(object): @skipUnless(HAS_GDAL, "LayerMapRouterTest needs GDAL support") @skipUnlessDBFeature("gis_enabled") +@override_settings(DATABASE_ROUTERS=[OtherRouter()]) class LayerMapRouterTest(TestCase): - def setUp(self): - self.old_routers = router.routers - router.routers = [OtherRouter()] - - def tearDown(self): - router.routers = self.old_routers - @unittest.skipUnless(len(settings.DATABASES) > 1, 'multiple databases required') def test_layermapping_default_db(self): lm = LayerMapping(City, city_shp, city_mapping) diff --git a/tests/cache/tests.py b/tests/cache/tests.py index 9dfd7af15d..e3ad724a86 100644 --- a/tests/cache/tests.py +++ b/tests/cache/tests.py @@ -19,7 +19,7 @@ from django.core import management from django.core.cache import (cache, caches, CacheKeyWarning, InvalidCacheBackendError, DEFAULT_CACHE_ALIAS) from django.core.context_processors import csrf -from django.db import connection, connections, router, transaction +from django.db import connection, connections, transaction from django.core.cache.utils import make_template_fragment_key from django.http import HttpResponse, StreamingHttpResponse from django.middleware.cache import (FetchFromCacheMiddleware, @@ -974,29 +974,25 @@ class DBCacheRouter(object): class CreateCacheTableForDBCacheTests(TestCase): multi_db = True + @override_settings(DATABASE_ROUTERS=[DBCacheRouter()]) def test_createcachetable_observes_database_router(self): - old_routers = router.routers - try: - router.routers = [DBCacheRouter()] - # cache table should not be created on 'default' - with self.assertNumQueries(0, using='default'): - management.call_command('createcachetable', - database='default', - verbosity=0, interactive=False) - # cache table should be created on 'other' - # Queries: - # 1: check table doesn't already exist - # 2: create savepoint (if transactional DDL is supported) - # 3: create the table - # 4: create the index - # 5: release savepoint (if transactional DDL is supported) - num = 5 if connections['other'].features.can_rollback_ddl else 3 - with self.assertNumQueries(num, using='other'): - management.call_command('createcachetable', - database='other', - verbosity=0, interactive=False) - finally: - router.routers = old_routers + # cache table should not be created on 'default' + with self.assertNumQueries(0, using='default'): + management.call_command('createcachetable', + database='default', + verbosity=0, interactive=False) + # cache table should be created on 'other' + # Queries: + # 1: check table doesn't already exist + # 2: create savepoint (if transactional DDL is supported) + # 3: create the table + # 4: create the index + # 5: release savepoint (if transactional DDL is supported) + num = 5 if connections['other'].features.can_rollback_ddl else 3 + with self.assertNumQueries(num, using='other'): + management.call_command('createcachetable', + database='other', + verbosity=0, interactive=False) class PicklingSideEffect(object): diff --git a/tests/commands_sql/tests.py b/tests/commands_sql/tests.py index e9499d6b05..e381c031a5 100644 --- a/tests/commands_sql/tests.py +++ b/tests/commands_sql/tests.py @@ -7,8 +7,8 @@ from django.apps import apps from django.core.management.color import no_style from django.core.management.sql import (sql_create, sql_delete, sql_indexes, sql_destroy_indexes, sql_all) -from django.db import connections, DEFAULT_DB_ALIAS, router -from django.test import TestCase +from django.db import connections, DEFAULT_DB_ALIAS +from django.test import TestCase, override_settings from django.utils import six # See also initial_sql_regress for 'custom_sql_for_model' tests @@ -91,13 +91,8 @@ class TestRouter(object): return False +@override_settings(DATABASE_ROUTERS=[TestRouter()]) class SQLCommandsRouterTestCase(TestCase): - def setUp(self): - self._old_routers = router.routers - router.routers = [TestRouter()] - - def tearDown(self): - router.routers = self._old_routers def test_router_honored(self): app_config = apps.get_app_config('commands_sql') diff --git a/tests/contenttypes_tests/tests.py b/tests/contenttypes_tests/tests.py index a21d783008..0e33f40c5c 100644 --- a/tests/contenttypes_tests/tests.py +++ b/tests/contenttypes_tests/tests.py @@ -8,9 +8,9 @@ from django.contrib.contenttypes.fields import ( from django.contrib.contenttypes import management from django.contrib.contenttypes.models import ContentType from django.core import checks -from django.db import connections, models, router -from django.test import TestCase -from django.test.utils import captured_stdout, override_settings +from django.db import connections, models +from django.test import TestCase, override_settings +from django.test.utils import captured_stdout from django.utils.encoding import force_str from .models import Author, Article, SchemeIncludedURL @@ -397,12 +397,10 @@ class TestRouter(object): return 'default' +@override_settings(DATABASE_ROUTERS=[TestRouter()]) class ContentTypesMultidbTestCase(TestCase): def setUp(self): - self.old_routers = router.routers - router.routers = [TestRouter()] - # Whenever a test starts executing, only the "default" database is # connected. We explicitly connect to the "other" database here. If we # don't do it, then it will be implicitly connected later when we query @@ -411,9 +409,6 @@ class ContentTypesMultidbTestCase(TestCase): # "SET SQL_AUTO_IS_NULL = 0"), which will affect assertNumQueries(). connections['other'].ensure_connection() - def tearDown(self): - router.routers = self.old_routers - def test_multidb(self): """ Test that, when using multiple databases, we use the db_for_read (see diff --git a/tests/migrations/test_operations.py b/tests/migrations/test_operations.py index ef6cbfc688..b08bdbc489 100644 --- a/tests/migrations/test_operations.py +++ b/tests/migrations/test_operations.py @@ -8,13 +8,13 @@ except ImportError: sqlparse = None from django import test -from django.test import override_settings -from django.db import connection, migrations, models, router +from django.db import connection, migrations, models from django.db.migrations.migration import Migration from django.db.migrations.state import ProjectState from django.db.models.fields import NOT_PROVIDED from django.db.transaction import atomic from django.db.utils import IntegrityError, DatabaseError +from django.test import override_settings from django.utils import six from .test_base import MigrationTestBase @@ -1566,18 +1566,10 @@ class MigrateNothingRouter(object): return False +@override_settings(DATABASE_ROUTERS=[MigrateNothingRouter()]) class MultiDBOperationTests(MigrationTestBase): multi_db = True - def setUp(self): - # Make the 'other' database appear to be a replica of the 'default' - self.old_routers = router.routers - router.routers = [MigrateNothingRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - def test_create_model(self): """ Tests that CreateModel honours multi-db settings. diff --git a/tests/multiple_database/tests.py b/tests/multiple_database/tests.py index 17a1ef90b1..90e418ba15 100644 --- a/tests/multiple_database/tests.py +++ b/tests/multiple_database/tests.py @@ -883,18 +883,11 @@ class ConnectionRouterTestCase(TestCase): ['TestRouter', 'WriteRouter']) +# Make the 'other' database appear to be a replica of the 'default' +@override_settings(DATABASE_ROUTERS=[TestRouter()]) class RouterTestCase(TestCase): multi_db = True - def setUp(self): - # Make the 'other' database appear to be a replica of the 'default' - self.old_routers = router.routers - router.routers = [TestRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - def test_db_selection(self): "Check that querysets obey the router for db suggestions" self.assertEqual(Book.objects.db, 'other') @@ -914,24 +907,22 @@ class RouterTestCase(TestCase): self.assertTrue(router.allow_migrate('other', User)) self.assertTrue(router.allow_migrate('other', Book)) - # Add the auth router to the chain. - # TestRouter is a universal synchronizer, so it should have no effect. - router.routers = [TestRouter(), AuthRouter()] + with override_settings(DATABASE_ROUTERS=[TestRouter(), AuthRouter()]): + # Add the auth router to the chain. TestRouter is a universal + # synchronizer, so it should have no effect. + self.assertTrue(router.allow_migrate('default', User)) + self.assertTrue(router.allow_migrate('default', Book)) - self.assertTrue(router.allow_migrate('default', User)) - self.assertTrue(router.allow_migrate('default', Book)) + self.assertTrue(router.allow_migrate('other', User)) + self.assertTrue(router.allow_migrate('other', Book)) - self.assertTrue(router.allow_migrate('other', User)) - self.assertTrue(router.allow_migrate('other', Book)) + with override_settings(DATABASE_ROUTERS=[AuthRouter(), TestRouter()]): + # Now check what happens if the router order is reversed. + self.assertFalse(router.allow_migrate('default', User)) + self.assertTrue(router.allow_migrate('default', Book)) - # Now check what happens if the router order is the other way around - router.routers = [AuthRouter(), TestRouter()] - - self.assertFalse(router.allow_migrate('default', User)) - self.assertTrue(router.allow_migrate('default', Book)) - - self.assertTrue(router.allow_migrate('other', User)) - self.assertFalse(router.allow_migrate('other', Book)) + self.assertTrue(router.allow_migrate('other', User)) + self.assertFalse(router.allow_migrate('other', Book)) def test_partial_router(self): "A router can choose to implement a subset of methods" @@ -951,18 +942,17 @@ class RouterTestCase(TestCase): self.assertTrue(router.allow_migrate('default', User)) self.assertTrue(router.allow_migrate('default', Book)) - router.routers = [WriteRouter(), AuthRouter(), TestRouter()] + with override_settings(DATABASE_ROUTERS=[WriteRouter(), AuthRouter(), TestRouter()]): + self.assertEqual(router.db_for_read(User), 'default') + self.assertEqual(router.db_for_read(Book), 'other') - self.assertEqual(router.db_for_read(User), 'default') - self.assertEqual(router.db_for_read(Book), 'other') + self.assertEqual(router.db_for_write(User), 'writer') + self.assertEqual(router.db_for_write(Book), 'writer') - self.assertEqual(router.db_for_write(User), 'writer') - self.assertEqual(router.db_for_write(Book), 'writer') + self.assertTrue(router.allow_relation(dive, dive)) - self.assertTrue(router.allow_relation(dive, dive)) - - self.assertFalse(router.allow_migrate('default', User)) - self.assertTrue(router.allow_migrate('default', Book)) + self.assertFalse(router.allow_migrate('default', User)) + self.assertTrue(router.allow_migrate('default', Book)) def test_database_routing(self): marty = Person.objects.using('default').create(name="Marty Alchin") @@ -1447,18 +1437,10 @@ class RouterTestCase(TestCase): datetime.date(2009, 5, 4)) +@override_settings(DATABASE_ROUTERS=[AuthRouter()]) class AuthTestCase(TestCase): multi_db = True - def setUp(self): - # Make the 'other' database appear to be a replica of the 'default' - self.old_routers = router.routers - router.routers = [AuthRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - def test_auth_manager(self): "The methods on the auth manager obey database hints" # Create one user using default allocation policy @@ -1522,15 +1504,7 @@ class FixtureTestCase(TestCase): multi_db = True fixtures = ['multidb-common', 'multidb'] - def setUp(self): - # Install the anti-pet router - self.old_routers = router.routers - router.routers = [AntiPetRouter()] - - def tearDown(self): - # Restore the 'other' database as an independent database - router.routers = self.old_routers - + @override_settings(DATABASE_ROUTERS=[AntiPetRouter()]) def test_fixture_loading(self): "Multi-db fixtures are loaded correctly" # Check that "Pro Django" exists on the default database, but not on other database @@ -1571,6 +1545,7 @@ class FixtureTestCase(TestCase): except Book.DoesNotExist: self.fail('"The Definitive Guide to Django" should exist on both databases') + @override_settings(DATABASE_ROUTERS=[AntiPetRouter()]) def test_pseudo_empty_fixtures(self): "A fixture can contain entries, but lead to nothing in the database; this shouldn't raise an error (ref #14068)" new_io = StringIO() @@ -1609,19 +1584,8 @@ class WriteToOtherRouter(object): class SignalTests(TestCase): multi_db = True - def setUp(self): - self.old_routers = router.routers - - def tearDown(self): - router.routers = self.old_routers - - def _write_to_other(self): - "Sends all writes to 'other'." - router.routers = [WriteToOtherRouter()] - - def _write_to_default(self): - "Sends all writes to the default DB" - router.routers = self.old_routers + def override_router(self): + return override_settings(DATABASE_ROUTERS=[WriteToOtherRouter()]) def test_database_arg_save_and_delete(self): """ @@ -1684,33 +1648,29 @@ class SignalTests(TestCase): # Test addition b.authors.add(p) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - b.authors.add(p) - self._write_to_default() + with self.override_router(): + b.authors.add(p) self.assertEqual(receiver._database, "other") # Test removal b.authors.remove(p) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - b.authors.remove(p) - self._write_to_default() + with self.override_router(): + b.authors.remove(p) self.assertEqual(receiver._database, "other") # Test addition in reverse p.book_set.add(b) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - p.book_set.add(b) - self._write_to_default() + with self.override_router(): + p.book_set.add(b) self.assertEqual(receiver._database, "other") # Test clearing b.authors.clear() self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) - self._write_to_other() - b.authors.clear() - self._write_to_default() + with self.override_router(): + b.authors.clear() self.assertEqual(receiver._database, "other") @@ -1726,47 +1686,41 @@ class AttributeErrorRouter(object): class RouterAttributeErrorTestCase(TestCase): multi_db = True - def setUp(self): - self.old_routers = router.routers - router.routers = [AttributeErrorRouter()] - - def tearDown(self): - router.routers = self.old_routers + def override_router(self): + return override_settings(DATABASE_ROUTERS=[AttributeErrorRouter()]) def test_attribute_error_read(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" - router.routers = [] # Reset routers so we can save a Book instance b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) - router.routers = [AttributeErrorRouter()] # Install our router - self.assertRaises(AttributeError, Book.objects.get, pk=b.pk) + with self.override_router(): + self.assertRaises(AttributeError, Book.objects.get, pk=b.pk) def test_attribute_error_save(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" dive = Book() dive.title = "Dive into Python" dive.published = datetime.date(2009, 5, 4) - self.assertRaises(AttributeError, dive.save) + with self.override_router(): + self.assertRaises(AttributeError, dive.save) def test_attribute_error_delete(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" - router.routers = [] # Reset routers so we can save our Book, Person instances b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) p = Person.objects.create(name="Marty Alchin") b.authors = [p] b.editor = p - router.routers = [AttributeErrorRouter()] # Install our router - self.assertRaises(AttributeError, b.delete) + with self.override_router(): + self.assertRaises(AttributeError, b.delete) def test_attribute_error_m2m(self): "Check that the AttributeError from AttributeErrorRouter bubbles up" - router.routers = [] # Reset routers so we can save our Book, Person instances b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) p = Person.objects.create(name="Marty Alchin") - router.routers = [AttributeErrorRouter()] # Install our router - self.assertRaises(AttributeError, setattr, b, 'authors', [p]) + with self.override_router(): + self.assertRaises(AttributeError, setattr, b, 'authors', [p]) class ModelMetaRouter(object): @@ -1776,16 +1730,10 @@ class ModelMetaRouter(object): raise ValueError +@override_settings(DATABASE_ROUTERS=[ModelMetaRouter()]) class RouterModelArgumentTestCase(TestCase): multi_db = True - def setUp(self): - self.old_routers = router.routers - router.routers = [ModelMetaRouter()] - - def tearDown(self): - router.routers = self.old_routers - def test_m2m_collection(self): b = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) @@ -1840,13 +1788,9 @@ class MigrateTestCase(TestCase): cts = ContentType.objects.using('other').filter(app_label='multiple_database') cts.delete() - try: - old_routers = router.routers - router.routers = [SyncOnlyDefaultDatabaseRouter()] + with override_settings(DATABASE_ROUTERS=[SyncOnlyDefaultDatabaseRouter()]): management.call_command('migrate', verbosity=0, interactive=False, load_initial_data=False, database='other') - finally: - router.routers = old_routers self.assertEqual(cts.count(), 0) @@ -1868,21 +1812,15 @@ class RouteForWriteTestCase(TestCase): def db_for_write(self, model, **hints): raise RouterUsed(mode=RouterUsed.WRITE, model=model, hints=hints) - def setUp(self): - self._old_rtrs = router.routers - - def tearDown(self): - router.routers = self._old_rtrs - - def enable_router(self): - router.routers = [RouteForWriteTestCase.WriteCheckRouter()] + def override_router(self): + return override_settings(DATABASE_ROUTERS=[RouteForWriteTestCase.WriteCheckRouter()]) def test_fk_delete(self): owner = Person.objects.create(name='Someone') pet = Pet.objects.create(name='fido', owner=owner) - self.enable_router() try: - pet.owner.delete() + with self.override_router(): + pet.owner.delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1892,9 +1830,9 @@ class RouteForWriteTestCase(TestCase): def test_reverse_fk_delete(self): owner = Person.objects.create(name='Someone') to_del_qs = owner.pet_set.all() - self.enable_router() try: - to_del_qs.delete() + with self.override_router(): + to_del_qs.delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1903,9 +1841,9 @@ class RouteForWriteTestCase(TestCase): def test_reverse_fk_get_or_create(self): owner = Person.objects.create(name='Someone') - self.enable_router() try: - owner.pet_set.get_or_create(name='fido') + with self.override_router(): + owner.pet_set.get_or_create(name='fido') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1915,9 +1853,9 @@ class RouteForWriteTestCase(TestCase): def test_reverse_fk_update(self): owner = Person.objects.create(name='Someone') Pet.objects.create(name='fido', owner=owner) - self.enable_router() try: - owner.pet_set.update(name='max') + with self.override_router(): + owner.pet_set.update(name='max') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1928,9 +1866,9 @@ class RouteForWriteTestCase(TestCase): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) - self.enable_router() try: - book.authors.add(auth) + with self.override_router(): + book.authors.add(auth) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1942,9 +1880,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - book.authors.clear() + with self.override_router(): + book.authors.clear() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1956,9 +1894,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - book.authors.all().delete() + with self.override_router(): + book.authors.all().delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1969,9 +1907,9 @@ class RouteForWriteTestCase(TestCase): Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) - self.enable_router() try: - book.authors.get_or_create(name='Someone else') + with self.override_router(): + book.authors.get_or_create(name='Someone else') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1983,10 +1921,10 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() self.assertRaisesMessage(AttributeError, self.RAISE_MSG, ) try: - book.authors.remove(auth) + with self.override_router(): + book.authors.remove(auth) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -1998,9 +1936,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - book.authors.all().update(name='Different') + with self.override_router(): + book.authors.all().update(name='Different') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -2011,9 +1949,9 @@ class RouteForWriteTestCase(TestCase): auth = Person.objects.create(name='Someone') book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) - self.enable_router() try: - auth.book_set.add(book) + with self.override_router(): + auth.book_set.add(book) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -2025,9 +1963,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - auth.book_set.clear() + with self.override_router(): + auth.book_set.clear() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -2039,9 +1977,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - auth.book_set.all().delete() + with self.override_router(): + auth.book_set.all().delete() self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -2052,9 +1990,9 @@ class RouteForWriteTestCase(TestCase): auth = Person.objects.create(name='Someone') Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) - self.enable_router() try: - auth.book_set.get_or_create(title="New Book", published=datetime.datetime.now()) + with self.override_router(): + auth.book_set.get_or_create(title="New Book", published=datetime.datetime.now()) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -2066,9 +2004,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - auth.book_set.remove(book) + with self.override_router(): + auth.book_set.remove(book) self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) @@ -2080,9 +2018,9 @@ class RouteForWriteTestCase(TestCase): book = Book.objects.create(title="Pro Django", published=datetime.date(2008, 12, 16)) book.authors.add(auth) - self.enable_router() try: - auth.book_set.all().update(title='Different') + with self.override_router(): + auth.book_set.all().update(title='Different') self.fail('db_for_write() not invoked on router') except RouterUsed as e: self.assertEqual(e.mode, RouterUsed.WRITE) diff --git a/tests/select_for_update/tests.py b/tests/select_for_update/tests.py index a0bd206923..d3f71ab65c 100644 --- a/tests/select_for_update/tests.py +++ b/tests/select_for_update/tests.py @@ -6,9 +6,10 @@ import unittest from django.conf import settings from django.db import transaction, connection, router from django.db.utils import ConnectionHandler, DEFAULT_DB_ALIAS, DatabaseError -from django.test import (TransactionTestCase, skipIfDBFeature, - skipUnlessDBFeature) -from django.test import override_settings +from django.test import ( + TransactionTestCase, override_settings, skipIfDBFeature, + skipUnlessDBFeature, +) from multiple_database.routers import TestRouter @@ -257,14 +258,10 @@ class SelectForUpdateTests(TransactionTestCase): self.assertIsInstance(status[-1], DatabaseError) @skipUnlessDBFeature('has_select_for_update') + @override_settings(DATABASE_ROUTERS=[TestRouter()]) def test_select_for_update_on_multidb(self): - old_routers = router.routers - try: - router.routers = [TestRouter()] - query = Person.objects.select_for_update() - self.assertEqual(router.db_for_write(Person), query.db) - finally: - router.routers = old_routers + query = Person.objects.select_for_update() + self.assertEqual(router.db_for_write(Person), query.db) @skipUnlessDBFeature('has_select_for_update') def test_select_for_update_with_get(self):