1
0
mirror of https://github.com/django/django.git synced 2024-12-22 17:16:24 +00:00

Replaced router.routers usage with override_settings(DATABASE_ROUTERS); refs #23933.

This commit is contained in:
wrwrwr 2014-11-29 18:09:40 +01:00 committed by Tim Graham
parent e6f19ec322
commit 9136ceb6fb
7 changed files with 121 additions and 214 deletions

View File

@ -8,9 +8,9 @@ import unittest
from unittest import skipUnless from unittest import skipUnless
from django.contrib.gis.gdal import HAS_GDAL 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.conf import settings
from django.test import TestCase, skipUnlessDBFeature from django.test import TestCase, override_settings, skipUnlessDBFeature
from django.utils._os import upath from django.utils._os import upath
if HAS_GDAL: if HAS_GDAL:
@ -328,15 +328,9 @@ class OtherRouter(object):
@skipUnless(HAS_GDAL, "LayerMapRouterTest needs GDAL support") @skipUnless(HAS_GDAL, "LayerMapRouterTest needs GDAL support")
@skipUnlessDBFeature("gis_enabled") @skipUnlessDBFeature("gis_enabled")
@override_settings(DATABASE_ROUTERS=[OtherRouter()])
class LayerMapRouterTest(TestCase): 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') @unittest.skipUnless(len(settings.DATABASES) > 1, 'multiple databases required')
def test_layermapping_default_db(self): def test_layermapping_default_db(self):
lm = LayerMapping(City, city_shp, city_mapping) lm = LayerMapping(City, city_shp, city_mapping)

42
tests/cache/tests.py vendored
View File

@ -19,7 +19,7 @@ from django.core import management
from django.core.cache import (cache, caches, CacheKeyWarning, from django.core.cache import (cache, caches, CacheKeyWarning,
InvalidCacheBackendError, DEFAULT_CACHE_ALIAS) InvalidCacheBackendError, DEFAULT_CACHE_ALIAS)
from django.core.context_processors import csrf 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.core.cache.utils import make_template_fragment_key
from django.http import HttpResponse, StreamingHttpResponse from django.http import HttpResponse, StreamingHttpResponse
from django.middleware.cache import (FetchFromCacheMiddleware, from django.middleware.cache import (FetchFromCacheMiddleware,
@ -974,29 +974,25 @@ class DBCacheRouter(object):
class CreateCacheTableForDBCacheTests(TestCase): class CreateCacheTableForDBCacheTests(TestCase):
multi_db = True multi_db = True
@override_settings(DATABASE_ROUTERS=[DBCacheRouter()])
def test_createcachetable_observes_database_router(self): def test_createcachetable_observes_database_router(self):
old_routers = router.routers # cache table should not be created on 'default'
try: with self.assertNumQueries(0, using='default'):
router.routers = [DBCacheRouter()] management.call_command('createcachetable',
# cache table should not be created on 'default' database='default',
with self.assertNumQueries(0, using='default'): verbosity=0, interactive=False)
management.call_command('createcachetable', # cache table should be created on 'other'
database='default', # Queries:
verbosity=0, interactive=False) # 1: check table doesn't already exist
# cache table should be created on 'other' # 2: create savepoint (if transactional DDL is supported)
# Queries: # 3: create the table
# 1: check table doesn't already exist # 4: create the index
# 2: create savepoint (if transactional DDL is supported) # 5: release savepoint (if transactional DDL is supported)
# 3: create the table num = 5 if connections['other'].features.can_rollback_ddl else 3
# 4: create the index with self.assertNumQueries(num, using='other'):
# 5: release savepoint (if transactional DDL is supported) management.call_command('createcachetable',
num = 5 if connections['other'].features.can_rollback_ddl else 3 database='other',
with self.assertNumQueries(num, using='other'): verbosity=0, interactive=False)
management.call_command('createcachetable',
database='other',
verbosity=0, interactive=False)
finally:
router.routers = old_routers
class PicklingSideEffect(object): class PicklingSideEffect(object):

View File

@ -7,8 +7,8 @@ from django.apps import apps
from django.core.management.color import no_style from django.core.management.color import no_style
from django.core.management.sql import (sql_create, sql_delete, sql_indexes, from django.core.management.sql import (sql_create, sql_delete, sql_indexes,
sql_destroy_indexes, sql_all) sql_destroy_indexes, sql_all)
from django.db import connections, DEFAULT_DB_ALIAS, router from django.db import connections, DEFAULT_DB_ALIAS
from django.test import TestCase from django.test import TestCase, override_settings
from django.utils import six from django.utils import six
# See also initial_sql_regress for 'custom_sql_for_model' tests # See also initial_sql_regress for 'custom_sql_for_model' tests
@ -91,13 +91,8 @@ class TestRouter(object):
return False return False
@override_settings(DATABASE_ROUTERS=[TestRouter()])
class SQLCommandsRouterTestCase(TestCase): 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): def test_router_honored(self):
app_config = apps.get_app_config('commands_sql') app_config = apps.get_app_config('commands_sql')

View File

@ -8,9 +8,9 @@ from django.contrib.contenttypes.fields import (
from django.contrib.contenttypes import management from django.contrib.contenttypes import management
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from django.core import checks from django.core import checks
from django.db import connections, models, router from django.db import connections, models
from django.test import TestCase from django.test import TestCase, override_settings
from django.test.utils import captured_stdout, override_settings from django.test.utils import captured_stdout
from django.utils.encoding import force_str from django.utils.encoding import force_str
from .models import Author, Article, SchemeIncludedURL from .models import Author, Article, SchemeIncludedURL
@ -397,12 +397,10 @@ class TestRouter(object):
return 'default' return 'default'
@override_settings(DATABASE_ROUTERS=[TestRouter()])
class ContentTypesMultidbTestCase(TestCase): class ContentTypesMultidbTestCase(TestCase):
def setUp(self): def setUp(self):
self.old_routers = router.routers
router.routers = [TestRouter()]
# Whenever a test starts executing, only the "default" database is # Whenever a test starts executing, only the "default" database is
# connected. We explicitly connect to the "other" database here. If we # 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 # 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(). # "SET SQL_AUTO_IS_NULL = 0"), which will affect assertNumQueries().
connections['other'].ensure_connection() connections['other'].ensure_connection()
def tearDown(self):
router.routers = self.old_routers
def test_multidb(self): def test_multidb(self):
""" """
Test that, when using multiple databases, we use the db_for_read (see Test that, when using multiple databases, we use the db_for_read (see

View File

@ -8,13 +8,13 @@ except ImportError:
sqlparse = None sqlparse = None
from django import test from django import test
from django.test import override_settings from django.db import connection, migrations, models
from django.db import connection, migrations, models, router
from django.db.migrations.migration import Migration from django.db.migrations.migration import Migration
from django.db.migrations.state import ProjectState from django.db.migrations.state import ProjectState
from django.db.models.fields import NOT_PROVIDED from django.db.models.fields import NOT_PROVIDED
from django.db.transaction import atomic from django.db.transaction import atomic
from django.db.utils import IntegrityError, DatabaseError from django.db.utils import IntegrityError, DatabaseError
from django.test import override_settings
from django.utils import six from django.utils import six
from .test_base import MigrationTestBase from .test_base import MigrationTestBase
@ -1566,18 +1566,10 @@ class MigrateNothingRouter(object):
return False return False
@override_settings(DATABASE_ROUTERS=[MigrateNothingRouter()])
class MultiDBOperationTests(MigrationTestBase): class MultiDBOperationTests(MigrationTestBase):
multi_db = True 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): def test_create_model(self):
""" """
Tests that CreateModel honours multi-db settings. Tests that CreateModel honours multi-db settings.

View File

@ -883,18 +883,11 @@ class ConnectionRouterTestCase(TestCase):
['TestRouter', 'WriteRouter']) ['TestRouter', 'WriteRouter'])
# Make the 'other' database appear to be a replica of the 'default'
@override_settings(DATABASE_ROUTERS=[TestRouter()])
class RouterTestCase(TestCase): class RouterTestCase(TestCase):
multi_db = True 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): def test_db_selection(self):
"Check that querysets obey the router for db suggestions" "Check that querysets obey the router for db suggestions"
self.assertEqual(Book.objects.db, 'other') 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', User))
self.assertTrue(router.allow_migrate('other', Book)) self.assertTrue(router.allow_migrate('other', Book))
# Add the auth router to the chain. with override_settings(DATABASE_ROUTERS=[TestRouter(), AuthRouter()]):
# TestRouter is a universal synchronizer, so it should have no effect. # Add the auth router to the chain. TestRouter is a universal
router.routers = [TestRouter(), AuthRouter()] # 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('other', User))
self.assertTrue(router.allow_migrate('default', Book)) self.assertTrue(router.allow_migrate('other', Book))
self.assertTrue(router.allow_migrate('other', User)) with override_settings(DATABASE_ROUTERS=[AuthRouter(), TestRouter()]):
self.assertTrue(router.allow_migrate('other', Book)) # 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 self.assertTrue(router.allow_migrate('other', User))
router.routers = [AuthRouter(), TestRouter()] self.assertFalse(router.allow_migrate('other', Book))
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))
def test_partial_router(self): def test_partial_router(self):
"A router can choose to implement a subset of methods" "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', User))
self.assertTrue(router.allow_migrate('default', Book)) 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_write(User), 'writer')
self.assertEqual(router.db_for_read(Book), 'other') self.assertEqual(router.db_for_write(Book), 'writer')
self.assertEqual(router.db_for_write(User), 'writer') self.assertTrue(router.allow_relation(dive, dive))
self.assertEqual(router.db_for_write(Book), 'writer')
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): def test_database_routing(self):
marty = Person.objects.using('default').create(name="Marty Alchin") marty = Person.objects.using('default').create(name="Marty Alchin")
@ -1447,18 +1437,10 @@ class RouterTestCase(TestCase):
datetime.date(2009, 5, 4)) datetime.date(2009, 5, 4))
@override_settings(DATABASE_ROUTERS=[AuthRouter()])
class AuthTestCase(TestCase): class AuthTestCase(TestCase):
multi_db = True 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): def test_auth_manager(self):
"The methods on the auth manager obey database hints" "The methods on the auth manager obey database hints"
# Create one user using default allocation policy # Create one user using default allocation policy
@ -1522,15 +1504,7 @@ class FixtureTestCase(TestCase):
multi_db = True multi_db = True
fixtures = ['multidb-common', 'multidb'] fixtures = ['multidb-common', 'multidb']
def setUp(self): @override_settings(DATABASE_ROUTERS=[AntiPetRouter()])
# 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
def test_fixture_loading(self): def test_fixture_loading(self):
"Multi-db fixtures are loaded correctly" "Multi-db fixtures are loaded correctly"
# Check that "Pro Django" exists on the default database, but not on other database # Check that "Pro Django" exists on the default database, but not on other database
@ -1571,6 +1545,7 @@ class FixtureTestCase(TestCase):
except Book.DoesNotExist: except Book.DoesNotExist:
self.fail('"The Definitive Guide to Django" should exist on both databases') self.fail('"The Definitive Guide to Django" should exist on both databases')
@override_settings(DATABASE_ROUTERS=[AntiPetRouter()])
def test_pseudo_empty_fixtures(self): 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)" "A fixture can contain entries, but lead to nothing in the database; this shouldn't raise an error (ref #14068)"
new_io = StringIO() new_io = StringIO()
@ -1609,19 +1584,8 @@ class WriteToOtherRouter(object):
class SignalTests(TestCase): class SignalTests(TestCase):
multi_db = True multi_db = True
def setUp(self): def override_router(self):
self.old_routers = router.routers return override_settings(DATABASE_ROUTERS=[WriteToOtherRouter()])
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 test_database_arg_save_and_delete(self): def test_database_arg_save_and_delete(self):
""" """
@ -1684,33 +1648,29 @@ class SignalTests(TestCase):
# Test addition # Test addition
b.authors.add(p) b.authors.add(p)
self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
self._write_to_other() with self.override_router():
b.authors.add(p) b.authors.add(p)
self._write_to_default()
self.assertEqual(receiver._database, "other") self.assertEqual(receiver._database, "other")
# Test removal # Test removal
b.authors.remove(p) b.authors.remove(p)
self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
self._write_to_other() with self.override_router():
b.authors.remove(p) b.authors.remove(p)
self._write_to_default()
self.assertEqual(receiver._database, "other") self.assertEqual(receiver._database, "other")
# Test addition in reverse # Test addition in reverse
p.book_set.add(b) p.book_set.add(b)
self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
self._write_to_other() with self.override_router():
p.book_set.add(b) p.book_set.add(b)
self._write_to_default()
self.assertEqual(receiver._database, "other") self.assertEqual(receiver._database, "other")
# Test clearing # Test clearing
b.authors.clear() b.authors.clear()
self.assertEqual(receiver._database, DEFAULT_DB_ALIAS) self.assertEqual(receiver._database, DEFAULT_DB_ALIAS)
self._write_to_other() with self.override_router():
b.authors.clear() b.authors.clear()
self._write_to_default()
self.assertEqual(receiver._database, "other") self.assertEqual(receiver._database, "other")
@ -1726,47 +1686,41 @@ class AttributeErrorRouter(object):
class RouterAttributeErrorTestCase(TestCase): class RouterAttributeErrorTestCase(TestCase):
multi_db = True multi_db = True
def setUp(self): def override_router(self):
self.old_routers = router.routers return override_settings(DATABASE_ROUTERS=[AttributeErrorRouter()])
router.routers = [AttributeErrorRouter()]
def tearDown(self):
router.routers = self.old_routers
def test_attribute_error_read(self): def test_attribute_error_read(self):
"Check that the AttributeError from AttributeErrorRouter bubbles up" "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", b = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
router.routers = [AttributeErrorRouter()] # Install our router with self.override_router():
self.assertRaises(AttributeError, Book.objects.get, pk=b.pk) self.assertRaises(AttributeError, Book.objects.get, pk=b.pk)
def test_attribute_error_save(self): def test_attribute_error_save(self):
"Check that the AttributeError from AttributeErrorRouter bubbles up" "Check that the AttributeError from AttributeErrorRouter bubbles up"
dive = Book() dive = Book()
dive.title = "Dive into Python" dive.title = "Dive into Python"
dive.published = datetime.date(2009, 5, 4) 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): def test_attribute_error_delete(self):
"Check that the AttributeError from AttributeErrorRouter bubbles up" "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", b = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
p = Person.objects.create(name="Marty Alchin") p = Person.objects.create(name="Marty Alchin")
b.authors = [p] b.authors = [p]
b.editor = p b.editor = p
router.routers = [AttributeErrorRouter()] # Install our router with self.override_router():
self.assertRaises(AttributeError, b.delete) self.assertRaises(AttributeError, b.delete)
def test_attribute_error_m2m(self): def test_attribute_error_m2m(self):
"Check that the AttributeError from AttributeErrorRouter bubbles up" "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", b = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
p = Person.objects.create(name="Marty Alchin") p = Person.objects.create(name="Marty Alchin")
router.routers = [AttributeErrorRouter()] # Install our router with self.override_router():
self.assertRaises(AttributeError, setattr, b, 'authors', [p]) self.assertRaises(AttributeError, setattr, b, 'authors', [p])
class ModelMetaRouter(object): class ModelMetaRouter(object):
@ -1776,16 +1730,10 @@ class ModelMetaRouter(object):
raise ValueError raise ValueError
@override_settings(DATABASE_ROUTERS=[ModelMetaRouter()])
class RouterModelArgumentTestCase(TestCase): class RouterModelArgumentTestCase(TestCase):
multi_db = True 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): def test_m2m_collection(self):
b = Book.objects.create(title="Pro Django", b = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
@ -1840,13 +1788,9 @@ class MigrateTestCase(TestCase):
cts = ContentType.objects.using('other').filter(app_label='multiple_database') cts = ContentType.objects.using('other').filter(app_label='multiple_database')
cts.delete() cts.delete()
try: with override_settings(DATABASE_ROUTERS=[SyncOnlyDefaultDatabaseRouter()]):
old_routers = router.routers
router.routers = [SyncOnlyDefaultDatabaseRouter()]
management.call_command('migrate', verbosity=0, interactive=False, management.call_command('migrate', verbosity=0, interactive=False,
load_initial_data=False, database='other') load_initial_data=False, database='other')
finally:
router.routers = old_routers
self.assertEqual(cts.count(), 0) self.assertEqual(cts.count(), 0)
@ -1868,21 +1812,15 @@ class RouteForWriteTestCase(TestCase):
def db_for_write(self, model, **hints): def db_for_write(self, model, **hints):
raise RouterUsed(mode=RouterUsed.WRITE, model=model, hints=hints) raise RouterUsed(mode=RouterUsed.WRITE, model=model, hints=hints)
def setUp(self): def override_router(self):
self._old_rtrs = router.routers return override_settings(DATABASE_ROUTERS=[RouteForWriteTestCase.WriteCheckRouter()])
def tearDown(self):
router.routers = self._old_rtrs
def enable_router(self):
router.routers = [RouteForWriteTestCase.WriteCheckRouter()]
def test_fk_delete(self): def test_fk_delete(self):
owner = Person.objects.create(name='Someone') owner = Person.objects.create(name='Someone')
pet = Pet.objects.create(name='fido', owner=owner) pet = Pet.objects.create(name='fido', owner=owner)
self.enable_router()
try: try:
pet.owner.delete() with self.override_router():
pet.owner.delete()
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1892,9 +1830,9 @@ class RouteForWriteTestCase(TestCase):
def test_reverse_fk_delete(self): def test_reverse_fk_delete(self):
owner = Person.objects.create(name='Someone') owner = Person.objects.create(name='Someone')
to_del_qs = owner.pet_set.all() to_del_qs = owner.pet_set.all()
self.enable_router()
try: try:
to_del_qs.delete() with self.override_router():
to_del_qs.delete()
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1903,9 +1841,9 @@ class RouteForWriteTestCase(TestCase):
def test_reverse_fk_get_or_create(self): def test_reverse_fk_get_or_create(self):
owner = Person.objects.create(name='Someone') owner = Person.objects.create(name='Someone')
self.enable_router()
try: 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') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1915,9 +1853,9 @@ class RouteForWriteTestCase(TestCase):
def test_reverse_fk_update(self): def test_reverse_fk_update(self):
owner = Person.objects.create(name='Someone') owner = Person.objects.create(name='Someone')
Pet.objects.create(name='fido', owner=owner) Pet.objects.create(name='fido', owner=owner)
self.enable_router()
try: 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') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1928,9 +1866,9 @@ class RouteForWriteTestCase(TestCase):
auth = Person.objects.create(name='Someone') auth = Person.objects.create(name='Someone')
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
self.enable_router()
try: try:
book.authors.add(auth) with self.override_router():
book.authors.add(auth)
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1942,9 +1880,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: try:
book.authors.clear() with self.override_router():
book.authors.clear()
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1956,9 +1894,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: try:
book.authors.all().delete() with self.override_router():
book.authors.all().delete()
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1969,9 +1907,9 @@ class RouteForWriteTestCase(TestCase):
Person.objects.create(name='Someone') Person.objects.create(name='Someone')
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
self.enable_router()
try: 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') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1983,10 +1921,10 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
self.assertRaisesMessage(AttributeError, self.RAISE_MSG, ) self.assertRaisesMessage(AttributeError, self.RAISE_MSG, )
try: try:
book.authors.remove(auth) with self.override_router():
book.authors.remove(auth)
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -1998,9 +1936,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: 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') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -2011,9 +1949,9 @@ class RouteForWriteTestCase(TestCase):
auth = Person.objects.create(name='Someone') auth = Person.objects.create(name='Someone')
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
self.enable_router()
try: try:
auth.book_set.add(book) with self.override_router():
auth.book_set.add(book)
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -2025,9 +1963,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: try:
auth.book_set.clear() with self.override_router():
auth.book_set.clear()
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -2039,9 +1977,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: try:
auth.book_set.all().delete() with self.override_router():
auth.book_set.all().delete()
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -2052,9 +1990,9 @@ class RouteForWriteTestCase(TestCase):
auth = Person.objects.create(name='Someone') auth = Person.objects.create(name='Someone')
Book.objects.create(title="Pro Django", Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
self.enable_router()
try: 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') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -2066,9 +2004,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: try:
auth.book_set.remove(book) with self.override_router():
auth.book_set.remove(book)
self.fail('db_for_write() not invoked on router') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)
@ -2080,9 +2018,9 @@ class RouteForWriteTestCase(TestCase):
book = Book.objects.create(title="Pro Django", book = Book.objects.create(title="Pro Django",
published=datetime.date(2008, 12, 16)) published=datetime.date(2008, 12, 16))
book.authors.add(auth) book.authors.add(auth)
self.enable_router()
try: 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') self.fail('db_for_write() not invoked on router')
except RouterUsed as e: except RouterUsed as e:
self.assertEqual(e.mode, RouterUsed.WRITE) self.assertEqual(e.mode, RouterUsed.WRITE)

View File

@ -6,9 +6,10 @@ import unittest
from django.conf import settings from django.conf import settings
from django.db import transaction, connection, router from django.db import transaction, connection, router
from django.db.utils import ConnectionHandler, DEFAULT_DB_ALIAS, DatabaseError from django.db.utils import ConnectionHandler, DEFAULT_DB_ALIAS, DatabaseError
from django.test import (TransactionTestCase, skipIfDBFeature, from django.test import (
skipUnlessDBFeature) TransactionTestCase, override_settings, skipIfDBFeature,
from django.test import override_settings skipUnlessDBFeature,
)
from multiple_database.routers import TestRouter from multiple_database.routers import TestRouter
@ -257,14 +258,10 @@ class SelectForUpdateTests(TransactionTestCase):
self.assertIsInstance(status[-1], DatabaseError) self.assertIsInstance(status[-1], DatabaseError)
@skipUnlessDBFeature('has_select_for_update') @skipUnlessDBFeature('has_select_for_update')
@override_settings(DATABASE_ROUTERS=[TestRouter()])
def test_select_for_update_on_multidb(self): def test_select_for_update_on_multidb(self):
old_routers = router.routers query = Person.objects.select_for_update()
try: self.assertEqual(router.db_for_write(Person), query.db)
router.routers = [TestRouter()]
query = Person.objects.select_for_update()
self.assertEqual(router.db_for_write(Person), query.db)
finally:
router.routers = old_routers
@skipUnlessDBFeature('has_select_for_update') @skipUnlessDBFeature('has_select_for_update')
def test_select_for_update_with_get(self): def test_select_for_update_with_get(self):