Created
October 3, 2016 23:49
-
-
Save johnmee/e0c74c92c2290afd976f08ff848193ff to your computer and use it in GitHub Desktop.
Django Test Runner with timings of each individual test
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from time import sleep | |
from time import time | |
from psycopg2 import OperationalError | |
from django.test.runner import DiscoverRunner | |
import django.db.backends.postgresql_psycopg2.base | |
from unittest.runner import TextTestRunner | |
from unittest.runner import TextTestResult | |
class TimedTextTestResult(TextTestResult): | |
def __init__(self, *args, **kwargs): | |
super(TimedTextTestResult, self).__init__(*args, **kwargs) | |
self.clocks = dict() | |
def startTest(self, test): | |
super(TextTestResult, self).startTest(test) | |
if self.showAll: | |
self.stream.write(self.getDescription(test)) | |
self.stream.write(" ... ") | |
self.stream.flush() | |
self.clocks[test] = time() | |
def addSuccess(self, test): | |
super(TextTestResult, self).addSuccess(test) | |
if self.showAll: | |
self.stream.writeln("ok-dokey (%.6fs)" % (time() - self.clocks[test])) | |
elif self.dots: | |
self.stream.write('.') | |
self.stream.flush() | |
class TimedTextTestRunner(TextTestRunner): | |
resultclass = TimedTextTestResult | |
class MqdTestRunner(DiscoverRunner): | |
def __init__(self, *args, **kwargs): | |
super(MqdTestRunner, self).__init__(*args, **kwargs) | |
self.original_postgres_conn_func = None | |
def setup_test_environment(self, **kwargs): | |
super(MqdTestRunner, self).setup_test_environment(**kwargs) | |
# The Postgres running on chromium.cmcrc.com is used to run MQD unit tests and | |
# integration tests. We get random 'LDAP authentication' errors when running the | |
# tests. Those errors are hard to reproduce and usually will go away when we restart | |
# the tests. | |
# | |
# The following function: patched_postgres_conn_func() automates the "re-connecting on | |
# LDAP authentication errors" by monkey patching the get_new_connection() method of | |
# Django postgresql backend. | |
if self.original_postgres_conn_func is not None: | |
# patch already applied, skip | |
return | |
else: | |
# save the original function before patching | |
self.original_postgres_conn_func = \ | |
django.db.backends.postgresql_psycopg2.base.DatabaseWrapper.get_new_connection | |
def patched_postgres_conn_func(database_wrapper_obj, conn_params): | |
max_retry_times = 10 | |
retry_interval = 0.5 # wait for 0.5 second before re-connecting | |
for _ in range(max_retry_times): | |
try: | |
return self.original_postgres_conn_func(database_wrapper_obj, conn_params) | |
except OperationalError as e: | |
if 'LDAP authentication failed' in e.message: | |
sleep(retry_interval) | |
print 'Re-connect database on error: %s' % e.message | |
continue | |
else: | |
raise | |
else: | |
raise | |
django.db.backends.postgresql_psycopg2.base.DatabaseWrapper.get_new_connection = \ | |
patched_postgres_conn_func | |
def teardown_test_environment(self, **kwargs): | |
if self.original_postgres_conn_func is not None: | |
django.db.backends.postgresql_psycopg2.base.DatabaseWrapper.get_new_connection = \ | |
self.original_postgres_conn_func | |
self.original_postgres_conn_func = None | |
super(MqdTestRunner, self).teardown_test_environment(**kwargs) | |
test_runner = TimedTextTestRunner |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment