path: root/tests/othertests/thread_isolation.py

# tests that db settings can be different in different threads
#
#
#    What's going on here:
#
#    Simulating multiple web requests in a threaded environment, one in
#    which settings are different for each request. So we replace
#    django.conf.settings with a thread local, with different
#    configurations in each thread, and then fire off three
#    simultaneous requests (using a condition to sync them up), and
#    test that each thread sees its own settings and the models in each
#    thread attempt to connect to the correct database as per their
#    settings.
#


import copy
import os
import sys
import threading
from thread import get_ident

from django import conf
from django.core.handlers.wsgi import WSGIHandler
from django.db import models, model_connection_name, _default, connection, \
     connections
from django.http import HttpResponse

try:
    # Only exists in Python 2.4+
    from threading import local
except ImportError:
    # Import copy of _thread_local.py from Python 2.4
    from django.utils._threading_local import local

# state holder
S = {}

# models
class MQ(models.Model):
    val = models.CharField(maxlength=10)
    class Meta:
        app_label = 'ti'


class MX(models.Model):
    val = models.CharField(maxlength=10)
    class Meta:
        app_label = 'ti'

        
class MY(models.Model):
    val = models.CharField(maxlength=10)
    class Meta:
        app_label = 'ti'

# eventused to synchronize threads so we can be sure they are running
# together
ev = threading.Event()


def test_one(path, request):
    """Start out with settings as originally configured"""
    from django.conf import settings
    debug("test_one: %s", settings.OTHER_DATABASES)

    assert model_connection_name(MQ) == _default
    assert model_connection_name(MX) == 'django_test_db_a', \
           "%s != 'django_test_db_a'" % model_connection_name(MX)
    assert MX._default_manager.db.connection.settings.DATABASE_NAME == \
           settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME']
    assert model_connection_name(MY) == 'django_test_db_b'
    assert MY._default_manager.db.connection.settings.DATABASE_NAME == \
           settings.OTHER_DATABASES['django_test_db_b']['DATABASE_NAME'], \
           "%s != %s" % \
           (MY._default_manager.db.connection.settings.DATABASE_NAME,
            settings.OTHER_DATABASES['django_test_db_b']['DATABASE_NAME'])
    assert MQ._default_manager.db.connection is \
           connections[_default].connection        
    assert MQ._default_manager.db.connection.settings.DATABASE_NAME == \
           settings.DATABASE_NAME
    assert connection.settings.DATABASE_NAME ==  settings.DATABASE_NAME


def test_two(path, request):
    """Between the first and second requests, settings change to assign
    model MY to a different connection
    """
    from django.conf import settings
    debug("test_two: %s", settings.OTHER_DATABASES)

    try:
        assert model_connection_name(MQ) == _default
        assert model_connection_name(MX) == 'django_test_db_a'
        assert MX._default_manager.db.connection.settings.DATABASE_NAME == \
               settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME']
        assert model_connection_name(MY) == _default
        assert MY._default_manager.db.connection.settings.DATABASE_NAME == \
               settings.DATABASE_NAME, "%s != %s" % \
               (MY._default_manager.db.connection.settings.DATABASE_NAME,
                settings.DATABASE_NAME)
        assert MQ._default_manager.db.connection is \
           connections[_default].connection        
        assert MQ._default_manager.db.connection.settings.DATABASE_NAME == \
               settings.DATABASE_NAME
        assert connection.settings.DATABASE_NAME == settings.DATABASE_NAME
    except:
        S.setdefault('errors',[]).append(sys.exc_info())


def test_three(path, request):
    """Between the 2nd and 3rd requests, the settings at the names in
    OTHER_DATABASES have changed.
    """
    from django.conf import settings
    debug("3 %s: %s", get_ident(), settings.OTHER_DATABASES)
    debug("3 %s: default: %s", get_ident(), settings.DATABASE_NAME)
    debug("3 %s: conn: %s", get_ident(), connection.settings.DATABASE_NAME)
    
    try:
        assert model_connection_name(MQ) == _default
        assert model_connection_name(MX) == 'django_test_db_b'
        assert MX._default_manager.db.connection.settings.DATABASE_NAME == \
               settings.OTHER_DATABASES['django_test_db_b']['DATABASE_NAME'],\
               "%s != %s" % \
               (MX._default_manager.db.connection.settings.DATABASE_NAME,
                settings.OTHER_DATABASES['django_test_db_b']['DATABASE_NAME'])
        assert model_connection_name(MY) == 'django_test_db_a'
        assert MY._default_manager.db.connection.settings.DATABASE_NAME == \
               settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME'],\
               "%s != %s" % \
               (MY._default_manager.db.connection.settings.DATABASE_NAME,
                settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME'])
        assert MQ._default_manager.db.connection is \
           connections[_default].connection
        assert connection.settings.DATABASE_NAME == \
               settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME'],\
               "%s != %s" % \
               (connection.settings.DATABASE_NAME,
                settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME'])
    except:
        S.setdefault('errors',[]).append(sys.exc_info())
        
    

# helpers
def thread_two(func, *arg):
    def start():
        from django.conf import settings
        for attr in [ a for a in dir(S['base_settings']) if a == a.upper() ]:
            setattr(settings, attr,
                    copy.deepcopy(getattr(S['base_settings'], attr)))
        
        settings.OTHER_DATABASES['django_test_db_b']['MODELS'] = []
        debug("t2 waiting")
        ev.wait(2.0)
        func(*arg)
        debug("t2 complete")
    t2 = threading.Thread(target=start)
    t2.start()
    return t2


def thread_three(func, *arg):
    def start():
        from django.conf import settings
        for attr in [ a for a in dir(S['base_settings']) if a == a.upper() ]:
            setattr(settings, attr,
                    copy.deepcopy(getattr(S['base_settings'], attr)))
            
        settings.OTHER_DATABASES['django_test_db_b']['MODELS'] = ['ti.MY']
        settings.OTHER_DATABASES['django_test_db_b'], \
            settings.OTHER_DATABASES['django_test_db_a'] = \
            settings.OTHER_DATABASES['django_test_db_a'], \
            settings.OTHER_DATABASES['django_test_db_b']

        settings.DATABASE_NAME = \
            settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME']

        # we just manually changed settings, so reset the default
        # connection (normally this isn't needed

        debug("3 %s: start: default: %s", get_ident(), settings.DATABASE_NAME)
        debug("3 %s: start: conn: %s", get_ident(),
              connection.settings.DATABASE_NAME)
        
        debug("t3 waiting")
        ev.wait(2.0)
        func(*arg)
        debug("t3 complete")
    t3 = threading.Thread(target=start)
    t3.start()
    return t3
    

class MockHandler(WSGIHandler):

    def __init__(self, test):
        self.test = test
        super(MockHandler, self).__init__()
        
    def get_response(self, path, request):
        # debug("mock handler answering %s, %s", path, request)
        return HttpResponse(self.test(path, request))
        

def pr(*arg):
    if S['verbosity'] >= 1:
        msg, arg = arg[0], arg[1:]
        print msg % arg


def debug(*arg):
    if S['verbosity'] >= 2:
        msg, arg = arg[0], arg[1:]
        print msg % arg


def setup():
    debug("setup")
    S['settings'] = copy.deepcopy(conf.settings)
    S['base_settings'] = copy.deepcopy(conf.settings)
    S['base_settings'].OTHER_DATABASES['django_test_db_a']['MODELS'] = \
        ['ti.MX']
    S['base_settings'].OTHER_DATABASES['django_test_db_b']['MODELS'] = \
        ['ti.MY']
    
    conf.settings = local()
    for attr in [ a for a in dir(S['base_settings']) if a == a.upper() ]:
        setattr(conf.settings, attr,
                copy.deepcopy(getattr(S['base_settings'], attr)))

    conf.settings.__module__ = S['settings'].__module__
    debug("Setup: %s", conf.settings.OTHER_DATABASES)

    
def teardown():
    debug("teardown")
    conf.settings = S['settings']


def start_response(code, headers):
    debug("start response: %s %s", code, headers)
    pass


def main():
    debug("running tests")

    env = os.environ.copy()
    env['PATH_INFO'] = '/'
    env['REQUEST_METHOD'] = 'GET'
    
    t2 = thread_two(MockHandler(test_two), env, start_response)
    t3 = thread_three(MockHandler(test_three), env, start_response)
    
    try:
        ev.set()
        MockHandler(test_one)(env, start_response)
    finally:
        t2.join()
        t3.join()
        err = S.get('errors', [])
        if err:
            import traceback        
            for e in err:
                traceback.print_exception(*e)
                raise AssertionError("%s thread%s failed" %
                                     (len(err), len(err) > 1 and 's' or ''))
    
def run_tests(verbosity=0):
    S['verbosity'] = verbosity
    
    setup()
    try:
        main()
    finally:
        teardown()

        
if __name__ == '__main__':
    from django.conf import settings
    
    settings.DATABASE_NAME = ':memory:'
    print "MAIN start! ", connection.settings.DATABASE_NAME
    connection.cursor()
    run_tests(2)