Fixed #12118 -- Added shared cache support to SQLite in-memory testing.

This commit is contained in:
Andriy Sokolovskiy 2014-12-04 01:17:59 +02:00 committed by Tim Graham
parent fca866763a
commit 8c99b7920e
6 changed files with 62 additions and 10 deletions

View File

@ -9,6 +9,7 @@ from __future__ import unicode_literals
import datetime
import decimal
import re
import sys
import uuid
import warnings
@ -123,6 +124,14 @@ class DatabaseFeatures(BaseDatabaseFeatures):
def can_release_savepoints(self):
return self.uses_savepoints
@cached_property
def can_share_in_memory_db(self):
return (
sys.version_info[:2] >= (3, 4) and
Database.__name__ == 'sqlite3.dbapi2' and
Database.sqlite_version_info >= (3, 7, 13)
)
@cached_property
def supports_stddev(self):
"""Confirm support for STDDEV and related stats functions
@ -405,6 +414,8 @@ class DatabaseWrapper(BaseDatabaseWrapper):
RuntimeWarning
)
kwargs.update({'check_same_thread': False})
if self.features.can_share_in_memory_db:
kwargs.update({'uri': True})
return kwargs
def get_new_connection(self, conn_params):
@ -429,7 +440,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
# If database is in memory, closing the connection destroys the
# database. To prevent accidental data loss, ignore close requests on
# an in-memory db.
if self.settings_dict['NAME'] != ":memory:":
if not self.is_in_memory_db(self.settings_dict['NAME']):
BaseDatabaseWrapper.close(self)
def _savepoint_allowed(self):
@ -505,6 +516,9 @@ class DatabaseWrapper(BaseDatabaseWrapper):
"""
self.cursor().execute("BEGIN")
def is_in_memory_db(self, name):
return name == ":memory:" or "mode=memory" in name
FORMAT_QMARK_REGEX = re.compile(r'(?<!%)%s')

View File

@ -1,6 +1,7 @@
import os
import sys
from django.core.exceptions import ImproperlyConfigured
from django.db.backends.creation import BaseDatabaseCreation
from django.utils.six.moves import input
@ -51,14 +52,22 @@ class DatabaseCreation(BaseDatabaseCreation):
def _get_test_db_name(self):
test_database_name = self.connection.settings_dict['TEST']['NAME']
if test_database_name and test_database_name != ':memory:':
if 'mode=memory' in test_database_name:
raise ImproperlyConfigured(
"Using `mode=memory` parameter in the database name is not allowed, "
"use `:memory:` instead."
)
return test_database_name
if self.connection.features.can_share_in_memory_db:
return 'file:memorydb_%s?mode=memory&cache=shared' % self.connection.alias
return ':memory:'
def _create_test_db(self, verbosity, autoclobber, keepdb=False):
test_database_name = self._get_test_db_name()
if keepdb:
return test_database_name
if test_database_name != ':memory:':
if not self.connection.is_in_memory_db(test_database_name):
# Erase the old test database
if verbosity >= 1:
print("Destroying old test database '%s'..." % self.connection.alias)
@ -80,7 +89,7 @@ class DatabaseCreation(BaseDatabaseCreation):
return test_database_name
def _destroy_test_db(self, test_database_name, verbosity):
if test_database_name and test_database_name != ":memory:":
if test_database_name and not self.connection.is_in_memory_db(test_database_name):
# Remove the SQLite database file
os.remove(test_database_name)
@ -92,8 +101,8 @@ class DatabaseCreation(BaseDatabaseCreation):
SQLite since the databases will be distinct despite having the same
TEST NAME. See http://www.sqlite.org/inmemorydb.html
"""
test_dbname = self._get_test_db_name()
test_database_name = self._get_test_db_name()
sig = [self.connection.settings_dict['NAME']]
if test_dbname == ':memory:':
if self.connection.is_in_memory_db(test_database_name):
sig.append(self.connection.alias)
return tuple(sig)

View File

@ -1212,8 +1212,7 @@ class LiveServerTestCase(TransactionTestCase):
for conn in connections.all():
# If using in-memory sqlite databases, pass the connections to
# the server thread.
if (conn.vendor == 'sqlite'
and conn.settings_dict['NAME'] == ':memory:'):
if conn.vendor == 'sqlite' and conn.is_in_memory_db(conn.settings_dict['NAME']):
# Explicitly enable thread-shareability for this connection
conn.allow_thread_sharing = True
connections_override[conn.alias] = conn
@ -1267,10 +1266,9 @@ class LiveServerTestCase(TransactionTestCase):
cls.server_thread.terminate()
cls.server_thread.join()
# Restore sqlite connections' non-shareability
# Restore sqlite in-memory database connections' non-shareability
for conn in connections.all():
if (conn.vendor == 'sqlite'
and conn.settings_dict['NAME'] == ':memory:'):
if conn.vendor == 'sqlite' and conn.is_in_memory_db(conn.settings_dict['NAME']):
conn.allow_thread_sharing = False
@classmethod

View File

@ -573,6 +573,10 @@ Tests
* Added test client support for file uploads with file-like objects.
* A shared cache is now used when testing with a SQLite in-memory database when
using Python 3.4+ and SQLite 3.7.13+. This allows sharing the database
between threads.
Validators
^^^^^^^^^^

View File

@ -185,12 +185,20 @@ control the particular collation used by the test database. See the
:doc:`settings documentation </ref/settings>` for details of these
and other advanced settings.
If using a SQLite in-memory database with Python 3.4+ and SQLite 3.7.13+,
`shared cache <https://www.sqlite.org/sharedcache.html>`_ will be enabled, so
you can write tests with ability to share the database between threads.
.. versionchanged:: 1.7
The different options in the :setting:`TEST <DATABASE-TEST>` database
setting used to be separate options in the database settings dictionary,
prefixed with ``TEST_``.
.. versionadded:: 1.8
The ability to use SQLite with a shared cache as described above was added.
.. admonition:: Finding data from your production database when running tests?
If your code attempts to access the database when its modules are compiled,

View File

@ -6,6 +6,7 @@ import copy
import datetime
from decimal import Decimal, Rounded
import re
import sys
import threading
import unittest
import warnings
@ -1201,3 +1202,21 @@ class DBTestSettingsRenamedTests(IgnoreAllDeprecationWarningsMixin, TestCase):
def test_empty_settings(self):
with override_settings(DATABASES=self.db_settings):
self.handler.prepare_test_settings('default')
@unittest.skipUnless(connection.vendor == 'sqlite', 'SQLite specific test.')
@skipUnlessDBFeature('can_share_in_memory_db')
class TestSqliteThreadSharing(TransactionTestCase):
available_apps = ['backends']
def test_database_sharing_in_threads(self):
def create_object():
models.Object.objects.create()
create_object()
thread = threading.Thread(target=create_object)
thread.start()
thread.join()
self.assertEqual(models.Object.objects.count(), 2)