Adjust Thread for using in-memory sqlite in tests

If you are using in memory sqlite database for running tests, as Django does, you will find trouble covering Python code that uses threading and reads or writes into the database in the thread routine.

Pysqlite’s default behavior is to prohibit the usage of a single connection in more than one thread. This is originally intended to work with older versions of SQLite that did not support multithreaded operation under various circumstances. In particular, older SQLite versions did not allow a:memory: database to be used in multiple threads under any circumstances.

as stated in SQLAlchemy docs

Basically, when running code in a separate thread, it will connect to a new empty instance of an sqlite in-memory database. Your tests will fail with errors like DatabaseError: no such table: app_model

The simplest trivial solution to this is replacing threading.Thread class with one that runs your code using the same thread (and thus, synchronously, but at least you get to run your tests using sqlite’s in-memory speed).

stubs.py:

import mock
 
 
class FakeThread(object):
    """
    Partial stub of threading.Thread
    `start` blocks running the callable, `join` does nothing
 
    """  
 
    def __init__(self, group=None, target=None, name=None,
            args=(), kwargs={}, verbose=None):
        self.target = target
        self.args = args
        self.kwargs = kwargs
 
    def start(self):
        self.target(*self.args, **self.kwargs)
 
    def join(self):
        pass

You use it like this (considering your “app.some.module” does “from threading import Thread”):

import unittest
import mock
 
from stubs import FakeThread
 
 
class SomeTestCase(unittest.TestCase):
 
    @mock.patch('app.some.module.Thread', new=FakeThread)
    @mock.patch('app.some.module.close_connection')
    def test_that_calls_code_using_appsomemodule(self, close_con):
        """
        do your thing, execution cursor reaches
        app.some.module, Thread will be replaced by FakeThread
        and your code will simply execute synchronously,
        e.g. .start() will block, .join() is no-op
 
        """

What I didn’t mention here is that you will probably have to mock other things which made sense when running in separate thread – e.g. using transactions in your threads. Failing to do so will lead to transaction related exceptions, e.g. TransactionManagementError: This code isn’t under transaction management or Transaction managed block ended with pending COMMIT/ROLLBACK. In my example I mocked “from django.db import close_connection”.