innov
/
django1
1
0
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

Migrated transactions doctests. Thanks to Eric Florenzano. 

git-svn-id: http://code.djangoproject.com/svn/django/trunk@13826 bcc190cf-cafb-0310-a4f2-bffc1f526a37
This commit is contained in:
Russell Keith-Magee 2010-09-13 05:28:29 +00:00
parent a60948cef2
commit 6948b280c7
2 changed files with 156 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

136
tests/modeltests/transactions/models.py
View File

@ -18,138 +18,4 @@ class Reporter(models.Model):
ordering = ('first_name', 'last_name') ordering = ('first_name', 'last_name')
def __unicode__(self): def __unicode__(self):
return u"%s %s" % (self.first_name, self.last_name) return u"%s %s" % (self.first_name, self.last_name)
__test__ = {'API_TESTS':"""
>>> from django.db import connection, transaction
"""}
from django.conf import settings
building_docs = getattr(settings, 'BUILDING_DOCS', False)
if building_docs or settings.DATABASES[DEFAULT_DB_ALIAS]['ENGINE'] != 'django.db.backends.mysql':
__test__['API_TESTS'] += """
# the default behavior is to autocommit after each save() action
>>> def create_a_reporter_then_fail(first, last):
... a = Reporter(first_name=first, last_name=last)
... a.save()
... raise Exception("I meant to do that")
...
>>> create_a_reporter_then_fail("Alice", "Smith")
Traceback (most recent call last):
...
Exception: I meant to do that
# The object created before the exception still exists
>>> Reporter.objects.all()
[<Reporter: Alice Smith>]
# the autocommit decorator works exactly the same as the default behavior
>>> autocomitted_create_then_fail = transaction.autocommit(create_a_reporter_then_fail)
>>> autocomitted_create_then_fail("Ben", "Jones")
Traceback (most recent call last):
...
Exception: I meant to do that
# Same behavior as before
>>> Reporter.objects.all()
[<Reporter: Alice Smith>, <Reporter: Ben Jones>]
# the autocommit decorator also works with a using argument
>>> using_autocomitted_create_then_fail = transaction.autocommit(using='default')(create_a_reporter_then_fail)
>>> using_autocomitted_create_then_fail("Carol", "Doe")
Traceback (most recent call last):
...
Exception: I meant to do that
# Same behavior as before
>>> Reporter.objects.all()
[<Reporter: Alice Smith>, <Reporter: Ben Jones>, <Reporter: Carol Doe>]
# With the commit_on_success decorator, the transaction is only committed if the
# function doesn't throw an exception
>>> committed_on_success = transaction.commit_on_success(create_a_reporter_then_fail)
>>> committed_on_success("Dirk", "Gently")
Traceback (most recent call last):
...
Exception: I meant to do that
# This time the object never got saved
>>> Reporter.objects.all()
[<Reporter: Alice Smith>, <Reporter: Ben Jones>, <Reporter: Carol Doe>]
# commit_on_success decorator also works with a using argument
>>> using_committed_on_success = transaction.commit_on_success(using='default')(create_a_reporter_then_fail)
>>> using_committed_on_success("Dirk", "Gently")
Traceback (most recent call last):
...
Exception: I meant to do that
# This time the object never got saved
>>> Reporter.objects.all()
[<Reporter: Alice Smith>, <Reporter: Ben Jones>, <Reporter: Carol Doe>]
# If there aren't any exceptions, the data will get saved
>>> def remove_a_reporter():
... r = Reporter.objects.get(first_name="Alice")
... r.delete()
...
>>> remove_comitted_on_success = transaction.commit_on_success(remove_a_reporter)
>>> remove_comitted_on_success()
>>> Reporter.objects.all()
[<Reporter: Ben Jones>, <Reporter: Carol Doe>]
# You can manually manage transactions if you really want to, but you
# have to remember to commit/rollback
>>> def manually_managed():
... r = Reporter(first_name="Dirk", last_name="Gently")
... r.save()
... transaction.commit()
>>> manually_managed = transaction.commit_manually(manually_managed)
>>> manually_managed()
>>> Reporter.objects.all()
[<Reporter: Ben Jones>, <Reporter: Carol Doe>, <Reporter: Dirk Gently>]
# If you forget, you'll get bad errors
>>> def manually_managed_mistake():
... r = Reporter(first_name="Edward", last_name="Woodward")
... r.save()
... # oops, I forgot to commit/rollback!
>>> manually_managed_mistake = transaction.commit_manually(manually_managed_mistake)
>>> manually_managed_mistake()
Traceback (most recent call last):
...
TransactionManagementError: Transaction managed block ended with pending COMMIT/ROLLBACK
# commit_manually also works with a using argument
>>> using_manually_managed_mistake = transaction.commit_manually(using='default')(manually_managed_mistake)
>>> using_manually_managed_mistake()
Traceback (most recent call last):
...
TransactionManagementError: Transaction managed block ended with pending COMMIT/ROLLBACK
"""
# Regression for #11900: If a function wrapped by commit_on_success writes a
# transaction that can't be committed, that transaction should be rolled back.
# The bug is only visible using the psycopg2 backend, though
# the fix is generally a good idea.
pgsql_backends = ('django.db.backends.postgresql_psycopg2', 'postgresql_psycopg2',)
if building_docs or settings.DATABASES[DEFAULT_DB_ALIAS]['ENGINE'] in pgsql_backends:
__test__['API_TESTS'] += """
>>> def execute_bad_sql():
... cursor = connection.cursor()
... cursor.execute("INSERT INTO transactions_reporter (first_name, last_name) VALUES ('Douglas', 'Adams');")
... transaction.set_dirty()
...
>>> execute_bad_sql = transaction.commit_on_success(execute_bad_sql)
>>> execute_bad_sql()
Traceback (most recent call last):
...
IntegrityError: null value in column "email" violates not-null constraint
<BLANKLINE>
>>> transaction.rollback()
"""

155
tests/modeltests/transactions/tests.py Normal file
View File

@ -0,0 +1,155 @@
from django.test import TransactionTestCase
from django.db import connection, transaction, IntegrityError, DEFAULT_DB_ALIAS
from django.conf import settings
from models import Reporter
PGSQL = 'psycopg2' in settings.DATABASES[DEFAULT_DB_ALIAS]['ENGINE']
MYSQL = 'mysql' in settings.DATABASES[DEFAULT_DB_ALIAS]['ENGINE']
class TransactionTests(TransactionTestCase):
if not MYSQL:
def create_a_reporter_then_fail(self, first, last):
a = Reporter(first_name=first, last_name=last)
a.save()
raise Exception("I meant to do that")
def remove_a_reporter(self, first_name):
r = Reporter.objects.get(first_name="Alice")
r.delete()
def manually_managed(self):
r = Reporter(first_name="Dirk", last_name="Gently")
r.save()
transaction.commit()
def manually_managed_mistake(self):
r = Reporter(first_name="Edward", last_name="Woodward")
r.save()
# Oops, I forgot to commit/rollback!
def execute_bad_sql(self):
cursor = connection.cursor()
cursor.execute("INSERT INTO transactions_reporter (first_name, last_name) VALUES ('Douglas', 'Adams');")
transaction.set_dirty()
def test_autocommit(self):
"""
The default behavior is to autocommit after each save() action.
"""
self.assertRaises(Exception,
self.create_a_reporter_then_fail,
"Alice", "Smith"
)
# The object created before the exception still exists
self.assertEqual(Reporter.objects.count(), 1)
def test_autocommit_decorator(self):
"""
The autocommit decorator works exactly the same as the default behavior.
"""
autocomitted_create_then_fail = transaction.autocommit(
self.create_a_reporter_then_fail
)
self.assertRaises(Exception,
autocomitted_create_then_fail,
"Alice", "Smith"
)
# Again, the object created before the exception still exists
self.assertEqual(Reporter.objects.count(), 1)
def test_autocommit_decorator_with_using(self):
"""
The autocommit decorator also works with a using argument.
"""
autocomitted_create_then_fail = transaction.autocommit(using='default')(
self.create_a_reporter_then_fail
)
self.assertRaises(Exception,
autocomitted_create_then_fail,
"Alice", "Smith"
)
# Again, the object created before the exception still exists
self.assertEqual(Reporter.objects.count(), 1)
def test_commit_on_success(self):
"""
With the commit_on_success decorator, the transaction is only committed
if the function doesn't throw an exception.
"""
committed_on_success = transaction.commit_on_success(
self.create_a_reporter_then_fail)
self.assertRaises(Exception, committed_on_success, "Dirk", "Gently")
# This time the object never got saved
self.assertEqual(Reporter.objects.count(), 0)
def test_commit_on_success_with_using(self):
"""
The commit_on_success decorator also works with a using argument.
"""
using_committed_on_success = transaction.commit_on_success(using='default')(
self.create_a_reporter_then_fail
)
self.assertRaises(Exception,
using_committed_on_success,
"Dirk", "Gently"
)
# This time the object never got saved
self.assertEqual(Reporter.objects.count(), 0)
def test_commit_on_success_succeed(self):
"""
If there aren't any exceptions, the data will get saved.
"""
Reporter.objects.create(first_name="Alice", last_name="Smith")
remove_comitted_on_success = transaction.commit_on_success(
self.remove_a_reporter
)
remove_comitted_on_success("Alice")
self.assertEqual(list(Reporter.objects.all()), [])
def test_manually_managed(self):
"""
You can manually manage transactions if you really want to, but you
have to remember to commit/rollback.
"""
manually_managed = transaction.commit_manually(self.manually_managed)
manually_managed()
self.assertEqual(Reporter.objects.count(), 1)
def test_manually_managed_mistake(self):
"""
If you forget, you'll get bad errors.
"""
manually_managed_mistake = transaction.commit_manually(
self.manually_managed_mistake
)
self.assertRaises(transaction.TransactionManagementError,
manually_managed_mistake)
def test_manually_managed_with_using(self):
"""
The commit_manually function also works with a using argument.
"""
using_manually_managed_mistake = transaction.commit_manually(using='default')(
self.manually_managed_mistake
)
self.assertRaises(transaction.TransactionManagementError,
using_manually_managed_mistake
)
if PGSQL:
def test_bad_sql(self):
"""
Regression for #11900: If a function wrapped by commit_on_success
writes a transaction that can't be committed, that transaction should
be rolled back. The bug is only visible using the psycopg2 backend,
though the fix is generally a good idea.
"""
execute_bad_sql = transaction.commit_on_success(self.execute_bad_sql)
self.assertRaises(IntegrityError, execute_bad_sql)
transaction.rollback()