1
0
mirror of https://github.com/django/django.git synced 2025-10-24 06:06:09 +00:00

Removed legacy transaction management per the deprecation timeline.

This commit is contained in:
Aymeric Augustin
2014-03-21 14:21:43 +01:00
parent 907ac64641
commit 0f9560855e
29 changed files with 78 additions and 1454 deletions

View File

@@ -5,15 +5,12 @@ from datetime import datetime, timedelta
from io import BytesIO
from itertools import chain
import time
from unittest import skipIf
from django.db import connection, connections
from django.core import signals
from django.core.exceptions import SuspiciousOperation
from django.core.handlers.wsgi import WSGIRequest, LimitedStream
from django.http import (HttpRequest, HttpResponse, parse_cookie,
build_request_repr, UnreadablePostError, RawPostDataException)
from django.test import SimpleTestCase, TransactionTestCase, override_settings
from django.test import SimpleTestCase, override_settings
from django.test.client import FakePayload
from django.test.utils import str_prefix
from django.utils import six
@@ -696,60 +693,3 @@ class HostValidationTests(SimpleTestCase):
msg_suggestion2 % "invalid_hostname.com",
request.get_host
)
@skipIf(connection.vendor == 'sqlite'
and connection.settings_dict['TEST']['NAME'] in (None, '', ':memory:'),
"Cannot establish two connections to an in-memory SQLite database.")
class DatabaseConnectionHandlingTests(TransactionTestCase):
available_apps = []
def setUp(self):
# Use a temporary connection to avoid messing with the main one.
self._old_default_connection = connections['default']
del connections['default']
def tearDown(self):
try:
connections['default'].close()
finally:
connections['default'] = self._old_default_connection
def test_request_finished_db_state(self):
# Force closing connection on request end
connection.settings_dict['CONN_MAX_AGE'] = 0
# The GET below will not succeed, but it will give a response with
# defined ._handler_class. That is needed for sending the
# request_finished signal.
response = self.client.get('/')
# Make sure there is an open connection
connection.ensure_connection()
connection.enter_transaction_management()
signals.request_finished.send(sender=response._handler_class)
self.assertEqual(len(connection.transaction_state), 0)
def test_request_finished_failed_connection(self):
# Force closing connection on request end
connection.settings_dict['CONN_MAX_AGE'] = 0
connection.enter_transaction_management()
connection.set_dirty()
# Test that the rollback doesn't succeed (for example network failure
# could cause this).
def fail_horribly():
raise Exception("Horrible failure!")
connection._rollback = fail_horribly
try:
with self.assertRaises(Exception):
signals.request_finished.send(sender=self.__class__)
# The connection's state wasn't cleaned up
self.assertEqual(len(connection.transaction_state), 1)
finally:
del connection._rollback
# The connection will be cleaned on next request where the conn
# works again.
signals.request_finished.send(sender=self.__class__)
self.assertEqual(len(connection.transaction_state), 0)