diff options
Diffstat (limited to 'tests/regressiontests/multiple_database/tests.py')
| -rw-r--r-- | tests/regressiontests/multiple_database/tests.py | 171 |
1 files changed, 155 insertions, 16 deletions
diff --git a/tests/regressiontests/multiple_database/tests.py b/tests/regressiontests/multiple_database/tests.py index 7bde8bf037..7f66ea3bcf 100644 --- a/tests/regressiontests/multiple_database/tests.py +++ b/tests/regressiontests/multiple_database/tests.py @@ -7,6 +7,7 @@ from django.conf import settings from django.contrib.auth.models import User from django.core import management from django.db import connections, router, DEFAULT_DB_ALIAS +from django.db.models import signals from django.db.utils import ConnectionRouter from django.test import TestCase @@ -883,7 +884,13 @@ class QueryTestCase(TestCase): self.assertRaises(ValueError, str, qs.query) # Evaluating the query shouldn't work, either - self.assertRaises(ValueError, list, qs) + try: + for obj in qs: + pass + self.fail('Iterating over query should raise ValueError') + except ValueError: + pass + class TestRouter(object): # A test router. The behaviour is vaguely master/slave, but the @@ -1491,19 +1498,10 @@ class AuthTestCase(TestCase): self.old_routers = router.routers router.routers = [AuthRouter()] - # Redirect stdout to a buffer so we can test - # the output of a management command - self.old_stdout = sys.stdout - self.stdout = StringIO() - sys.stdout = self.stdout - def tearDown(self): # Restore the 'other' database as an independent database router.routers = self.old_routers - # Restore stdout - sys.stdout = self.old_stdout - def test_auth_manager(self): "The methods on the auth manager obey database hints" # Create one user using default allocation policy @@ -1539,14 +1537,16 @@ class AuthTestCase(TestCase): # Check that dumping the default database doesn't try to include auth # because allow_syncdb prohibits auth on default - self.stdout.flush() - management.call_command('dumpdata', 'auth', format='json', database='default') - self.assertEquals(self.stdout.getvalue(), '[]\n') + new_io = StringIO() + management.call_command('dumpdata', 'auth', format='json', database='default', stdout=new_io) + command_output = new_io.getvalue().strip() + self.assertEqual(command_output, '[]') # Check that dumping the other database does include auth - self.stdout.flush() - management.call_command('dumpdata', 'auth', format='json', database='other') - self.assertTrue('alice@example.com' in self.stdout.getvalue()) + new_io = StringIO() + management.call_command('dumpdata', 'auth', format='json', database='other', stdout=new_io) + command_output = new_io.getvalue().strip() + self.assertTrue('"email": "alice@example.com",' in command_output) class UserProfileTestCase(TestCase): def setUp(self): @@ -1570,11 +1570,31 @@ class UserProfileTestCase(TestCase): self.assertEquals(alice.get_profile().flavor, 'chocolate') self.assertEquals(bob.get_profile().flavor, 'crunchy frog') +class AntiPetRouter(object): + # A router that only expresses an opinion on syncdb, + # passing pets to the 'other' database + + def allow_syncdb(self, db, model): + "Make sure the auth app only appears on the 'other' db" + if db == 'other': + return model._meta.object_name == 'Pet' + else: + return model._meta.object_name != 'Pet' + return None 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 + 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 @@ -1612,6 +1632,14 @@ class FixtureTestCase(TestCase): except Book.DoesNotExist: self.fail('"The Definitive Guide to Django" should exist on both databases') + 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() + management.call_command('loaddata', 'pets', stdout=new_io, stderr=new_io) + command_output = new_io.getvalue().strip() + # No objects will actually be loaded + self.assertEqual(command_output, "Installed 0 object(s) (of 2) from 1 fixture(s)") + class PickleQuerySetTestCase(TestCase): multi_db = True @@ -1620,3 +1648,114 @@ class PickleQuerySetTestCase(TestCase): Book.objects.using(db).create(title='Dive into Python', published=datetime.date(2009, 5, 4)) qs = Book.objects.all() self.assertEqual(qs.db, pickle.loads(pickle.dumps(qs)).db) + + +class DatabaseReceiver(object): + """ + Used in the tests for the database argument in signals (#13552) + """ + def __call__(self, signal, sender, **kwargs): + self._database = kwargs['using'] + +class WriteToOtherRouter(object): + """ + A router that sends all writes to the other database. + """ + def db_for_write(self, model, **hints): + return "other" + +class SignalTests(TestCase): + multi_db = True + + def setUp(self): + self.old_routers = router.routers + + def tearDown(self): + router.routser = 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): + """ + Tests that the pre/post_save signal contains the correct database. + (#13552) + """ + # Make some signal receivers + pre_save_receiver = DatabaseReceiver() + post_save_receiver = DatabaseReceiver() + pre_delete_receiver = DatabaseReceiver() + post_delete_receiver = DatabaseReceiver() + # Make model and connect receivers + signals.pre_save.connect(sender=Person, receiver=pre_save_receiver) + signals.post_save.connect(sender=Person, receiver=post_save_receiver) + signals.pre_delete.connect(sender=Person, receiver=pre_delete_receiver) + signals.post_delete.connect(sender=Person, receiver=post_delete_receiver) + p = Person.objects.create(name='Darth Vader') + # Save and test receivers got calls + p.save() + self.assertEqual(pre_save_receiver._database, DEFAULT_DB_ALIAS) + self.assertEqual(post_save_receiver._database, DEFAULT_DB_ALIAS) + # Delete, and test + p.delete() + self.assertEqual(pre_delete_receiver._database, DEFAULT_DB_ALIAS) + self.assertEqual(post_delete_receiver._database, DEFAULT_DB_ALIAS) + # Save again to a different database + p.save(using="other") + self.assertEqual(pre_save_receiver._database, "other") + self.assertEqual(post_save_receiver._database, "other") + # Delete, and test + p.delete(using="other") + self.assertEqual(pre_delete_receiver._database, "other") + self.assertEqual(post_delete_receiver._database, "other") + + def test_database_arg_m2m(self): + """ + Test that the m2m_changed signal has a correct database arg (#13552) + """ + # Make a receiver + receiver = DatabaseReceiver() + # Connect it, and make the models + signals.m2m_changed.connect(receiver=receiver) + + b = Book.objects.create(title="Pro Django", + published=datetime.date(2008, 12, 16)) + + p = Person.objects.create(name="Marty Alchin") + + # 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() + 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() + 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() + 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() + self.assertEqual(receiver._database, "other") |