| # -*- coding: utf-8 -*- |
| # Unit and doctests for specific database backends. |
| from __future__ import with_statement, absolute_import |
| |
| import datetime |
| import threading |
| |
| from django.conf import settings |
| from django.core.management.color import no_style |
| from django.core.exceptions import ImproperlyConfigured |
| from django.db import (backend, connection, connections, DEFAULT_DB_ALIAS, |
| IntegrityError, transaction) |
| from django.db.backends.signals import connection_created |
| from django.db.backends.postgresql_psycopg2 import version as pg_version |
| from django.db.utils import ConnectionHandler, DatabaseError, load_backend |
| from django.test import TestCase, skipUnlessDBFeature, TransactionTestCase |
| from django.test.utils import override_settings |
| from django.utils import unittest |
| |
| from . import models |
| |
| |
| class OracleChecks(unittest.TestCase): |
| |
| @unittest.skipUnless(connection.vendor == 'oracle', |
| "No need to check Oracle cursor semantics") |
| def test_dbms_session(self): |
| # If the backend is Oracle, test that we can call a standard |
| # stored procedure through our cursor wrapper. |
| convert_unicode = backend.convert_unicode |
| cursor = connection.cursor() |
| cursor.callproc(convert_unicode('DBMS_SESSION.SET_IDENTIFIER'), |
| [convert_unicode('_django_testing!'),]) |
| |
| @unittest.skipUnless(connection.vendor == 'oracle', |
| "No need to check Oracle cursor semantics") |
| def test_cursor_var(self): |
| # If the backend is Oracle, test that we can pass cursor variables |
| # as query parameters. |
| cursor = connection.cursor() |
| var = cursor.var(backend.Database.STRING) |
| cursor.execute("BEGIN %s := 'X'; END; ", [var]) |
| self.assertEqual(var.getvalue(), 'X') |
| |
| @unittest.skipUnless(connection.vendor == 'oracle', |
| "No need to check Oracle cursor semantics") |
| def test_long_string(self): |
| # If the backend is Oracle, test that we can save a text longer |
| # than 4000 chars and read it properly |
| c = connection.cursor() |
| c.execute('CREATE TABLE ltext ("TEXT" NCLOB)') |
| long_str = ''.join([unicode(x) for x in xrange(4000)]) |
| c.execute('INSERT INTO ltext VALUES (%s)',[long_str]) |
| c.execute('SELECT text FROM ltext') |
| row = c.fetchone() |
| self.assertEqual(long_str, row[0].read()) |
| c.execute('DROP TABLE ltext') |
| |
| @unittest.skipUnless(connection.vendor == 'oracle', |
| "No need to check Oracle connection semantics") |
| def test_client_encoding(self): |
| # If the backend is Oracle, test that the client encoding is set |
| # correctly. This was broken under Cygwin prior to r14781. |
| c = connection.cursor() # Ensure the connection is initialized. |
| self.assertEqual(connection.connection.encoding, "UTF-8") |
| self.assertEqual(connection.connection.nencoding, "UTF-8") |
| |
| class MySQLTests(TestCase): |
| @unittest.skipUnless(connection.vendor == 'mysql', |
| "Test valid only for MySQL") |
| def test_server_version_connections(self): |
| connection.close() |
| connection.get_server_version() |
| self.assertTrue(connection.connection is None) |
| |
| class DateQuotingTest(TestCase): |
| |
| def test_django_date_trunc(self): |
| """ |
| Test the custom ``django_date_trunc method``, in particular against |
| fields which clash with strings passed to it (e.g. 'year') - see |
| #12818__. |
| |
| __: http://code.djangoproject.com/ticket/12818 |
| |
| """ |
| updated = datetime.datetime(2010, 2, 20) |
| models.SchoolClass.objects.create(year=2009, last_updated=updated) |
| years = models.SchoolClass.objects.dates('last_updated', 'year') |
| self.assertEqual(list(years), [datetime.datetime(2010, 1, 1, 0, 0)]) |
| |
| def test_django_extract(self): |
| """ |
| Test the custom ``django_extract method``, in particular against fields |
| which clash with strings passed to it (e.g. 'day') - see #12818__. |
| |
| __: http://code.djangoproject.com/ticket/12818 |
| |
| """ |
| updated = datetime.datetime(2010, 2, 20) |
| models.SchoolClass.objects.create(year=2009, last_updated=updated) |
| classes = models.SchoolClass.objects.filter(last_updated__day=20) |
| self.assertEqual(len(classes), 1) |
| |
| class LastExecutedQueryTest(TestCase): |
| |
| def setUp(self): |
| # connection.queries will not be filled in without this |
| settings.DEBUG = True |
| |
| def tearDown(self): |
| settings.DEBUG = False |
| |
| # There are no tests for the sqlite backend because it does not |
| # implement paramater escaping. See #14091. |
| |
| @unittest.skipUnless(connection.vendor in ('oracle', 'postgresql'), |
| "These backends use the standard parameter escaping rules") |
| def test_parameter_escaping(self): |
| # check that both numbers and string are properly quoted |
| list(models.Tag.objects.filter(name="special:\\\"':", object_id=12)) |
| sql = connection.queries[-1]['sql'] |
| self.assertTrue("= 'special:\\\"'':' " in sql) |
| self.assertTrue("= 12 " in sql) |
| |
| @unittest.skipUnless(connection.vendor == 'mysql', |
| "MySQL uses backslashes to escape parameters.") |
| def test_parameter_escaping(self): |
| list(models.Tag.objects.filter(name="special:\\\"':", object_id=12)) |
| sql = connection.queries[-1]['sql'] |
| # only this line is different from the test above |
| self.assertTrue("= 'special:\\\\\\\"\\':' " in sql) |
| self.assertTrue("= 12 " in sql) |
| |
| class ParameterHandlingTest(TestCase): |
| def test_bad_parameter_count(self): |
| "An executemany call with too many/not enough parameters will raise an exception (Refs #12612)" |
| cursor = connection.cursor() |
| query = ('INSERT INTO %s (%s, %s) VALUES (%%s, %%s)' % ( |
| connection.introspection.table_name_converter('backends_square'), |
| connection.ops.quote_name('root'), |
| connection.ops.quote_name('square') |
| )) |
| self.assertRaises(Exception, cursor.executemany, query, [(1,2,3),]) |
| self.assertRaises(Exception, cursor.executemany, query, [(1,),]) |
| |
| # Unfortunately, the following tests would be a good test to run on all |
| # backends, but it breaks MySQL hard. Until #13711 is fixed, it can't be run |
| # everywhere (although it would be an effective test of #13711). |
| class LongNameTest(TestCase): |
| """Long primary keys and model names can result in a sequence name |
| that exceeds the database limits, which will result in truncation |
| on certain databases (e.g., Postgres). The backend needs to use |
| the correct sequence name in last_insert_id and other places, so |
| check it is. Refs #8901. |
| """ |
| |
| @skipUnlessDBFeature('supports_long_model_names') |
| def test_sequence_name_length_limits_create(self): |
| """Test creation of model with long name and long pk name doesn't error. Ref #8901""" |
| models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create() |
| |
| @skipUnlessDBFeature('supports_long_model_names') |
| def test_sequence_name_length_limits_m2m(self): |
| """Test an m2m save of a model with a long name and a long m2m field name doesn't error as on Django >=1.2 this now uses object saves. Ref #8901""" |
| obj = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create() |
| rel_obj = models.Person.objects.create(first_name='Django', last_name='Reinhardt') |
| obj.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.add(rel_obj) |
| |
| @skipUnlessDBFeature('supports_long_model_names') |
| def test_sequence_name_length_limits_flush(self): |
| """Test that sequence resetting as part of a flush with model with long name and long pk name doesn't error. Ref #8901""" |
| # A full flush is expensive to the full test, so we dig into the |
| # internals to generate the likely offending SQL and run it manually |
| |
| # Some convenience aliases |
| VLM = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ |
| VLM_m2m = VLM.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.through |
| tables = [ |
| VLM._meta.db_table, |
| VLM_m2m._meta.db_table, |
| ] |
| sequences = [ |
| { |
| 'column': VLM._meta.pk.column, |
| 'table': VLM._meta.db_table |
| }, |
| ] |
| cursor = connection.cursor() |
| for statement in connection.ops.sql_flush(no_style(), tables, sequences): |
| cursor.execute(statement) |
| |
| class SequenceResetTest(TestCase): |
| def test_generic_relation(self): |
| "Sequence names are correct when resetting generic relations (Ref #13941)" |
| # Create an object with a manually specified PK |
| models.Post.objects.create(id=10, name='1st post', text='hello world') |
| |
| # Reset the sequences for the database |
| cursor = connection.cursor() |
| commands = connections[DEFAULT_DB_ALIAS].ops.sequence_reset_sql(no_style(), [models.Post]) |
| for sql in commands: |
| cursor.execute(sql) |
| |
| # If we create a new object now, it should have a PK greater |
| # than the PK we specified manually. |
| obj = models.Post.objects.create(name='New post', text='goodbye world') |
| self.assertTrue(obj.pk > 10) |
| |
| class PostgresVersionTest(TestCase): |
| def assert_parses(self, version_string, version): |
| self.assertEqual(pg_version._parse_version(version_string), version) |
| |
| def test_parsing(self): |
| """Test PostgreSQL version parsing from `SELECT version()` output""" |
| self.assert_parses("PostgreSQL 8.3 beta4", 80300) |
| self.assert_parses("PostgreSQL 8.3", 80300) |
| self.assert_parses("EnterpriseDB 8.3", 80300) |
| self.assert_parses("PostgreSQL 8.3.6", 80306) |
| self.assert_parses("PostgreSQL 8.4beta1", 80400) |
| self.assert_parses("PostgreSQL 8.3.1 on i386-apple-darwin9.2.2, compiled by GCC i686-apple-darwin9-gcc-4.0.1 (GCC) 4.0.1 (Apple Inc. build 5478)", 80301) |
| |
| def test_version_detection(self): |
| """Test PostgreSQL version detection""" |
| |
| # Helper mocks |
| class CursorMock(object): |
| "Very simple mock of DB-API cursor" |
| def execute(self, arg): |
| pass |
| |
| def fetchone(self): |
| return ["PostgreSQL 8.3"] |
| |
| class OlderConnectionMock(object): |
| "Mock of psycopg2 (< 2.0.12) connection" |
| def cursor(self): |
| return CursorMock() |
| |
| # psycopg2 < 2.0.12 code path |
| conn = OlderConnectionMock() |
| self.assertEqual(pg_version.get_version(conn), 80300) |
| |
| class PostgresNewConnectionTest(TestCase): |
| """ |
| #17062: PostgreSQL shouldn't roll back SET TIME ZONE, even if the first |
| transaction is rolled back. |
| """ |
| @unittest.skipUnless( |
| connection.vendor == 'postgresql' and connection.isolation_level > 0, |
| "This test applies only to PostgreSQL without autocommit") |
| def test_connect_and_rollback(self): |
| new_connections = ConnectionHandler(settings.DATABASES) |
| new_connection = new_connections[DEFAULT_DB_ALIAS] |
| try: |
| # Ensure the database default time zone is different than |
| # the time zone in new_connection.settings_dict. We can |
| # get the default time zone by reset & show. |
| cursor = new_connection.cursor() |
| cursor.execute("RESET TIMEZONE") |
| cursor.execute("SHOW TIMEZONE") |
| db_default_tz = cursor.fetchone()[0] |
| new_tz = 'Europe/Paris' if db_default_tz == 'UTC' else 'UTC' |
| new_connection.close() |
| |
| # Fetch a new connection with the new_tz as default |
| # time zone, run a query and rollback. |
| new_connection.settings_dict['TIME_ZONE'] = new_tz |
| new_connection.enter_transaction_management() |
| cursor = new_connection.cursor() |
| new_connection.rollback() |
| |
| # Now let's see if the rollback rolled back the SET TIME ZONE. |
| cursor.execute("SHOW TIMEZONE") |
| tz = cursor.fetchone()[0] |
| self.assertEqual(new_tz, tz) |
| finally: |
| try: |
| new_connection.close() |
| except DatabaseError: |
| pass |
| |
| |
| # Unfortunately with sqlite3 the in-memory test database cannot be |
| # closed, and so it cannot be re-opened during testing, and so we |
| # sadly disable this test for now. |
| class ConnectionCreatedSignalTest(TestCase): |
| @skipUnlessDBFeature('test_db_allows_multiple_connections') |
| def test_signal(self): |
| data = {} |
| def receiver(sender, connection, **kwargs): |
| data["connection"] = connection |
| |
| connection_created.connect(receiver) |
| connection.close() |
| cursor = connection.cursor() |
| self.assertTrue(data["connection"].connection is connection.connection) |
| |
| connection_created.disconnect(receiver) |
| data.clear() |
| cursor = connection.cursor() |
| self.assertTrue(data == {}) |
| |
| |
| class EscapingChecks(TestCase): |
| |
| @unittest.skipUnless(connection.vendor == 'sqlite', |
| "This is a sqlite-specific issue") |
| def test_parameter_escaping(self): |
| #13648: '%s' escaping support for sqlite3 |
| cursor = connection.cursor() |
| response = cursor.execute( |
| "select strftime('%%s', date('now'))").fetchall()[0][0] |
| self.assertNotEqual(response, None) |
| # response should be an non-zero integer |
| self.assertTrue(int(response)) |
| |
| |
| class BackendTestCase(TestCase): |
| |
| def create_squares_with_executemany(self, args): |
| cursor = connection.cursor() |
| opts = models.Square._meta |
| tbl = connection.introspection.table_name_converter(opts.db_table) |
| f1 = connection.ops.quote_name(opts.get_field('root').column) |
| f2 = connection.ops.quote_name(opts.get_field('square').column) |
| query = 'INSERT INTO %s (%s, %s) VALUES (%%s, %%s)' % (tbl, f1, f2) |
| cursor.executemany(query, args) |
| |
| def test_cursor_executemany(self): |
| #4896: Test cursor.executemany |
| args = [(i, i**2) for i in range(-5, 6)] |
| self.create_squares_with_executemany(args) |
| self.assertEqual(models.Square.objects.count(), 11) |
| for i in range(-5, 6): |
| square = models.Square.objects.get(root=i) |
| self.assertEqual(square.square, i**2) |
| |
| def test_cursor_executemany_with_empty_params_list(self): |
| #4765: executemany with params=[] does nothing |
| args = [] |
| self.create_squares_with_executemany(args) |
| self.assertEqual(models.Square.objects.count(), 0) |
| |
| def test_cursor_executemany_with_iterator(self): |
| #10320: executemany accepts iterators |
| args = iter((i, i**2) for i in range(-3, 2)) |
| self.create_squares_with_executemany(args) |
| self.assertEqual(models.Square.objects.count(), 5) |
| |
| args = iter((i, i**2) for i in range(3, 7)) |
| with override_settings(DEBUG=True): |
| # same test for DebugCursorWrapper |
| self.create_squares_with_executemany(args) |
| self.assertEqual(models.Square.objects.count(), 9) |
| |
| |
| def test_unicode_fetches(self): |
| #6254: fetchone, fetchmany, fetchall return strings as unicode objects |
| qn = connection.ops.quote_name |
| models.Person(first_name="John", last_name="Doe").save() |
| models.Person(first_name="Jane", last_name="Doe").save() |
| models.Person(first_name="Mary", last_name="Agnelline").save() |
| models.Person(first_name="Peter", last_name="Parker").save() |
| models.Person(first_name="Clark", last_name="Kent").save() |
| opts2 = models.Person._meta |
| f3, f4 = opts2.get_field('first_name'), opts2.get_field('last_name') |
| query2 = ('SELECT %s, %s FROM %s ORDER BY %s' |
| % (qn(f3.column), qn(f4.column), connection.introspection.table_name_converter(opts2.db_table), |
| qn(f3.column))) |
| cursor = connection.cursor() |
| cursor.execute(query2) |
| self.assertEqual(cursor.fetchone(), (u'Clark', u'Kent')) |
| self.assertEqual(list(cursor.fetchmany(2)), [(u'Jane', u'Doe'), (u'John', u'Doe')]) |
| self.assertEqual(list(cursor.fetchall()), [(u'Mary', u'Agnelline'), (u'Peter', u'Parker')]) |
| |
| def test_database_operations_helper_class(self): |
| # Ticket #13630 |
| self.assertTrue(hasattr(connection, 'ops')) |
| self.assertTrue(hasattr(connection.ops, 'connection')) |
| self.assertEqual(connection, connection.ops.connection) |
| |
| def test_duplicate_table_error(self): |
| """ Test that creating an existing table returns a DatabaseError """ |
| cursor = connection.cursor() |
| query = 'CREATE TABLE %s (id INTEGER);' % models.Article._meta.db_table |
| with self.assertRaises(DatabaseError): |
| cursor.execute(query) |
| |
| # We don't make these tests conditional because that means we would need to |
| # check and differentiate between: |
| # * MySQL+InnoDB, MySQL+MYISAM (something we currently can't do). |
| # * if sqlite3 (if/once we get #14204 fixed) has referential integrity turned |
| # on or not, something that would be controlled by runtime support and user |
| # preference. |
| # verify if its type is django.database.db.IntegrityError. |
| |
| class FkConstraintsTests(TransactionTestCase): |
| |
| def setUp(self): |
| # Create a Reporter. |
| self.r = models.Reporter.objects.create(first_name='John', last_name='Smith') |
| |
| def test_integrity_checks_on_creation(self): |
| """ |
| Try to create a model instance that violates a FK constraint. If it |
| fails it should fail with IntegrityError. |
| """ |
| a = models.Article(headline="This is a test", pub_date=datetime.datetime(2005, 7, 27), reporter_id=30) |
| try: |
| a.save() |
| except IntegrityError: |
| return |
| self.skipTest("This backend does not support integrity checks.") |
| |
| def test_integrity_checks_on_update(self): |
| """ |
| Try to update a model instance introducing a FK constraint violation. |
| If it fails it should fail with IntegrityError. |
| """ |
| # Create an Article. |
| models.Article.objects.create(headline="Test article", pub_date=datetime.datetime(2010, 9, 4), reporter=self.r) |
| # Retrive it from the DB |
| a = models.Article.objects.get(headline="Test article") |
| a.reporter_id = 30 |
| try: |
| a.save() |
| except IntegrityError: |
| return |
| self.skipTest("This backend does not support integrity checks.") |
| |
| def test_disable_constraint_checks_manually(self): |
| """ |
| When constraint checks are disabled, should be able to write bad data without IntegrityErrors. |
| """ |
| with transaction.commit_manually(): |
| # Create an Article. |
| models.Article.objects.create(headline="Test article", pub_date=datetime.datetime(2010, 9, 4), reporter=self.r) |
| # Retrive it from the DB |
| a = models.Article.objects.get(headline="Test article") |
| a.reporter_id = 30 |
| try: |
| connection.disable_constraint_checking() |
| a.save() |
| connection.enable_constraint_checking() |
| except IntegrityError: |
| self.fail("IntegrityError should not have occurred.") |
| finally: |
| transaction.rollback() |
| |
| def test_disable_constraint_checks_context_manager(self): |
| """ |
| When constraint checks are disabled (using context manager), should be able to write bad data without IntegrityErrors. |
| """ |
| with transaction.commit_manually(): |
| # Create an Article. |
| models.Article.objects.create(headline="Test article", pub_date=datetime.datetime(2010, 9, 4), reporter=self.r) |
| # Retrive it from the DB |
| a = models.Article.objects.get(headline="Test article") |
| a.reporter_id = 30 |
| try: |
| with connection.constraint_checks_disabled(): |
| a.save() |
| except IntegrityError: |
| self.fail("IntegrityError should not have occurred.") |
| finally: |
| transaction.rollback() |
| |
| def test_check_constraints(self): |
| """ |
| Constraint checks should raise an IntegrityError when bad data is in the DB. |
| """ |
| with transaction.commit_manually(): |
| # Create an Article. |
| models.Article.objects.create(headline="Test article", pub_date=datetime.datetime(2010, 9, 4), reporter=self.r) |
| # Retrive it from the DB |
| a = models.Article.objects.get(headline="Test article") |
| a.reporter_id = 30 |
| try: |
| with connection.constraint_checks_disabled(): |
| a.save() |
| with self.assertRaises(IntegrityError): |
| connection.check_constraints() |
| finally: |
| transaction.rollback() |
| |
| |
| class ThreadTests(TestCase): |
| |
| def test_default_connection_thread_local(self): |
| """ |
| Ensure that the default connection (i.e. django.db.connection) is |
| different for each thread. |
| Refs #17258. |
| """ |
| connections_set = set() |
| connection.cursor() |
| connections_set.add(connection.connection) |
| def runner(): |
| from django.db import connection |
| connection.cursor() |
| connections_set.add(connection.connection) |
| for x in xrange(2): |
| t = threading.Thread(target=runner) |
| t.start() |
| t.join() |
| self.assertEquals(len(connections_set), 3) |
| # Finish by closing the connections opened by the other threads (the |
| # connection opened in the main thread will automatically be closed on |
| # teardown). |
| for conn in connections_set: |
| if conn != connection.connection: |
| conn.close() |
| |
| def test_connections_thread_local(self): |
| """ |
| Ensure that the connections are different for each thread. |
| Refs #17258. |
| """ |
| connections_set = set() |
| for conn in connections.all(): |
| connections_set.add(conn) |
| def runner(): |
| from django.db import connections |
| for conn in connections.all(): |
| # Allow thread sharing so the connection can be closed by the |
| # main thread. |
| conn.allow_thread_sharing = True |
| connections_set.add(conn) |
| for x in xrange(2): |
| t = threading.Thread(target=runner) |
| t.start() |
| t.join() |
| self.assertEquals(len(connections_set), 6) |
| # Finish by closing the connections opened by the other threads (the |
| # connection opened in the main thread will automatically be closed on |
| # teardown). |
| for conn in connections_set: |
| if conn != connection: |
| conn.close() |
| |
| def test_pass_connection_between_threads(self): |
| """ |
| Ensure that a connection can be passed from one thread to the other. |
| Refs #17258. |
| """ |
| models.Person.objects.create(first_name="John", last_name="Doe") |
| |
| def do_thread(): |
| def runner(main_thread_connection): |
| from django.db import connections |
| connections['default'] = main_thread_connection |
| try: |
| models.Person.objects.get(first_name="John", last_name="Doe") |
| except DatabaseError, e: |
| exceptions.append(e) |
| t = threading.Thread(target=runner, args=[connections['default']]) |
| t.start() |
| t.join() |
| |
| # Without touching allow_thread_sharing, which should be False by default. |
| exceptions = [] |
| do_thread() |
| # Forbidden! |
| self.assertTrue(isinstance(exceptions[0], DatabaseError)) |
| |
| # If explicitly setting allow_thread_sharing to False |
| connections['default'].allow_thread_sharing = False |
| exceptions = [] |
| do_thread() |
| # Forbidden! |
| self.assertTrue(isinstance(exceptions[0], DatabaseError)) |
| |
| # If explicitly setting allow_thread_sharing to True |
| connections['default'].allow_thread_sharing = True |
| exceptions = [] |
| do_thread() |
| # All good |
| self.assertEqual(len(exceptions), 0) |
| |
| def test_closing_non_shared_connections(self): |
| """ |
| Ensure that a connection that is not explicitly shareable cannot be |
| closed by another thread. |
| Refs #17258. |
| """ |
| # First, without explicitly enabling the connection for sharing. |
| exceptions = set() |
| def runner1(): |
| def runner2(other_thread_connection): |
| try: |
| other_thread_connection.close() |
| except DatabaseError, e: |
| exceptions.add(e) |
| t2 = threading.Thread(target=runner2, args=[connections['default']]) |
| t2.start() |
| t2.join() |
| t1 = threading.Thread(target=runner1) |
| t1.start() |
| t1.join() |
| # The exception was raised |
| self.assertEqual(len(exceptions), 1) |
| |
| # Then, with explicitly enabling the connection for sharing. |
| exceptions = set() |
| def runner1(): |
| def runner2(other_thread_connection): |
| try: |
| other_thread_connection.close() |
| except DatabaseError, e: |
| exceptions.add(e) |
| # Enable thread sharing |
| connections['default'].allow_thread_sharing = True |
| t2 = threading.Thread(target=runner2, args=[connections['default']]) |
| t2.start() |
| t2.join() |
| t1 = threading.Thread(target=runner1) |
| t1.start() |
| t1.join() |
| # No exception was raised |
| self.assertEqual(len(exceptions), 0) |
| |
| |
| class BackendLoadingTests(TestCase): |
| def test_old_style_backends_raise_useful_exception(self): |
| self.assertRaisesRegexp(ImproperlyConfigured, |
| "Try using django.db.backends.sqlite3 instead", |
| load_backend, 'sqlite3') |
