mirror of
https://github.com/django/django.git
synced 2024-11-18 23:44:22 +00:00
86c20e39eb
In addition, removed a possibly problematic .filter() call from backends.test_query_encoding test. It is possible the .filter could cause collation problems on MySQL, and as it wasn't absolutely needed for the test it seemed better to get rid of the call. Refs #18461.
675 lines
28 KiB
Python
675 lines
28 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Unit and doctests for specific database backends.
|
|
from __future__ import absolute_import, unicode_literals
|
|
|
|
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, skipIfDBFeature,
|
|
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.
|
|
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_autoincrement(self):
|
|
"""
|
|
Check that auto_increment fields are reset correctly by sql_flush().
|
|
Before MySQL version 5.0.13 TRUNCATE did not do auto_increment reset.
|
|
Refs #16961.
|
|
"""
|
|
statements = connection.ops.sql_flush(no_style(),
|
|
tables=['test'],
|
|
sequences=[{
|
|
'table': 'test',
|
|
'col': 'somecol',
|
|
}])
|
|
found_reset = False
|
|
for sql in statements:
|
|
found_reset = found_reset or 'ALTER TABLE' in sql
|
|
if connection.mysql_version < (5,0,13):
|
|
self.assertTrue(found_reset)
|
|
else:
|
|
self.assertFalse(found_reset)
|
|
|
|
@unittest.skipUnless(connection.vendor == 'mysql',
|
|
"Test valid only for MySQL")
|
|
def test_server_version_connections(self):
|
|
connection.close()
|
|
connection.mysql_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):
|
|
# There are no escaping 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")
|
|
@override_settings(DEBUG=True)
|
|
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.")
|
|
@override_settings(DEBUG=True)
|
|
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)
|
|
|
|
def test_query_encoding(self):
|
|
"""
|
|
Test that last_executed_query() returns an Unicode string
|
|
"""
|
|
tags = models.Tag.objects.extra(select={'föö':1})
|
|
sql, params = tags.query.sql_with_params()
|
|
cursor = tags.query.get_compiler('default').execute_sql(None)
|
|
last_sql = cursor.db.ops.last_executed_query(cursor, sql, params)
|
|
self.assertTrue(isinstance(last_sql, unicode))
|
|
|
|
|
|
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(), ('Clark', 'Kent'))
|
|
self.assertEqual(list(cursor.fetchmany(2)), [('Jane', 'Doe'), ('John', 'Doe')])
|
|
self.assertEqual(list(cursor.fetchall()), [('Mary', 'Agnelline'), ('Peter', '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_cached_db_features(self):
|
|
self.assertIn(connection.features.supports_transactions, (True, False))
|
|
self.assertIn(connection.features.supports_stddev, (True, False))
|
|
self.assertIn(connection.features.can_introspect_foreign_keys, (True, False))
|
|
|
|
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.assertEqual(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.assertEqual(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 as 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 as 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 as 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')
|
|
|
|
|
|
class MySQLPKZeroTests(TestCase):
|
|
"""
|
|
Zero as id for AutoField should raise exception in MySQL, because MySQL
|
|
does not allow zero for automatic primary key.
|
|
"""
|
|
|
|
@skipIfDBFeature('allows_primary_key_0')
|
|
def test_zero_as_autoval(self):
|
|
with self.assertRaises(ValueError):
|
|
models.Square.objects.create(id=0, root=0, square=1)
|