Jason Moiron committed 326bf30

add a router and some intelligence to johnny's settings file to automatically set up MULTIDB tests if you have multiple databases configured; also added a convenience decorator that shows johnny's hit/miss signals, useful for debugging failing tests or even examining johnny's operation

Comments (0)

Files changed (4)


 """Base test class for Johnny Cache Tests."""
 import sys
+from functools import wraps
 import django
 from django.test import TestCase, TransactionTestCase
 # order matters here;  I guess we aren't deferring foreign key checking :\
 johnny_fixtures = ['authors.json', 'genres.json', 'publishers.json', 'books.json']
+def show_johnny_signals(hit=None, miss=None):
+    """A decorator that can be put on a test function that will show the
+    johnny hit/miss signals by default, or do what is provided by the hit
+    and miss keyword args."""
+    from pprint import pformat
+    def _hit(*args, **kwargs):
+        print "hit:\n\t%s\n\t%s\n" % (pformat(args), pformat(kwargs))
+    def _miss(*args, **kwargs):
+        print "miss:\n\t%s\n\t%s\n" % (pformat(args), pformat(kwargs))
+    hit = _hit if hit is None else hit
+    miss = _miss if miss is None else miss
+    def deco(func):
+        @wraps(func)
+        def wrapped(*args, **kwargs):
+            from johnny.signals import qc_hit, qc_miss
+            qc_hit.connect(hit)
+            qc_miss.connect(miss)
+            ret = func(*args, **kwargs)
+            qc_hit.disconnect(hit)
+            qc_miss.disconnect(miss)
+            return ret
+        return wrapped
+    return deco
 def _pre_setup(self):
     self.saved_INSTALLED_APPS = settings.INSTALLED_APPS
     self.saved_DEBUG = settings.DEBUG
     # load our fake application and syncdb
     call_command('syncdb', verbosity=0, interactive=False)
+    if hasattr(settings, 'DATABASES'):
+        for dbname in settings.DATABASES:
+            if dbname != 'default':
+                call_command('syncdb', verbosity=0, interactive=False, database=dbname)
 def _post_teardown(self):
     settings.INSTALLED_APPS = self.saved_INSTALLED_APPS


             print "\n  Skipping multi database tests"
+        from pprint import pformat
         from testapp.models import Genre, Book, Publisher, Person
         from django.db import connections
         self.failUnless("default" in getattr(settings, "DATABASES"))
         self.failUnless("second" in getattr(settings, "DATABASES"))
         g1 = Genre.objects.using("default").get(pk=1)
         g1.title = "A default database"
         g2 = Genre.objects.using("second").get(pk=1)
         g2.title = "A second database"
         for c in connections:
             connections[c].queries = []
         #fresh from cache since we saved each
         g2 = Genre.objects.using('second').get(pk=1)
         for c in connections:
             self.failUnless(len(connections[c].queries) == 1)
-        # test for a regression on the WhereNode, bitbucket #20
-        g1 = Genre.objects.get(Q(title__iexact="A second database"))
     def test_transactions(self):
     def test_basic_querycaching(self):
         """A basic test that querycaching is functioning properly and is
         being invalidated properly on singular table reads & writes."""
-        from testapp.models import Publisher
+        from testapp.models import Publisher, Genre
+        from django.db.models import Q
         connection.queries = []
         starting_count = Publisher.objects.count()
         starting_count = Publisher.objects.count()
         # this tests the codepath after 'except EmptyResultSet' where
         # result_type == MULTI
         self.failUnless(not list(Publisher.objects.filter(title__in=[])))
+        # test for a regression on the WhereNode, bitbucket #20
+        g1 = Genre.objects.get(pk=1)
+        g1.title = "Survival Horror"
+        g1 = Genre.objects.get(Q(title__iexact="Survival Horror"))
     def test_querycache_return_results(self):
         """Test that the return results from the query cache are what we
+# routers for johnny's tests
+class MultiSyncedRouter(object):
+    def db_for_read(self, *args, **kwargs): return None
+    def db_for_write(self, *args, **kwargs): return None
+    def allow_relation(self, *args, **kwargs): return None
+    def allow_sync_db(self, db, model):
+        return True
 except ImportError:
+# set up a multi-db router if there are multiple databases set
+lcls = locals()
+if 'DATABASES' in lcls and len(lcls['DATABASES']) > 1:
+    DATABASE_ROUTERS = ['routers.MultiSyncedRouter']