TransactionManagementError "You can't execute queries until the end of the 'atomic' block" while using signals, but only during Unit Testing

transactiontestcase
django tests transactionmanagementerror
django transaction atomic
hive atomic' block
this is forbidden when an 'atomic' block is active.
django on_commit example
the rollback flag doesn t work outside of an atomic' block
an error occurred in the current transaction

I am getting TransactionManagementError when trying to save a Django User model instance and in its post_save signal, I'm saving some models that have the user as the foreign key.

The context and error is pretty similar to this question django TransactionManagementError when using signals

However, in this case, the error occurs only while unit testing.

It works well in manual testing, but unit tests fails.

Is there anything that I'm missing?

Here are the code snippets:

views.py

@csrf_exempt
def mobileRegister(request):
    if request.method == 'GET':
        response = {"error": "GET request not accepted!!"}
        return HttpResponse(json.dumps(response), content_type="application/json",status=500)
    elif request.method == 'POST':
        postdata = json.loads(request.body)
        try:
            # Get POST data which is to be used to save the user
            username = postdata.get('phone')
            password = postdata.get('password')
            email = postdata.get('email',"")
            first_name = postdata.get('first_name',"")
            last_name = postdata.get('last_name',"")
            user = User(username=username, email=email,
                        first_name=first_name, last_name=last_name)
            user._company = postdata.get('company',None)
            user._country_code = postdata.get('country_code',"+91")
            user.is_verified=True
            user._gcm_reg_id = postdata.get('reg_id',None)
            user._gcm_device_id = postdata.get('device_id',None)
            # Set Password for the user
            user.set_password(password)
            # Save the user
            user.save()

signal.py

def create_user_profile(sender, instance, created, **kwargs):
    if created:
        company = None
        companycontact = None
        try:   # Try to make userprofile with company and country code provided
            user = User.objects.get(id=instance.id)
            rand_pass = random.randint(1000, 9999)
            company = Company.objects.get_or_create(name=instance._company,user=user)
            companycontact = CompanyContact.objects.get_or_create(contact_type="Owner",company=company,contact_number=instance.username)
            profile = UserProfile.objects.get_or_create(user=instance,phone=instance.username,verification_code=rand_pass,company=company,country_code=instance._country_code)
            gcmDevice = GCMDevice.objects.create(registration_id=instance._gcm_reg_id,device_id=instance._gcm_reg_id,user=instance)
        except Exception, e:
            pass

tests.py

class AuthTestCase(TestCase):
    fixtures = ['nextgencatalogs/fixtures.json']
    def setUp(self):
        self.user_data={
            "phone":"0000000000",
            "password":"123",
            "first_name":"Gaurav",
            "last_name":"Toshniwal"
            }

    def test_registration_api_get(self):
        response = self.client.get("/mobileRegister/")
        self.assertEqual(response.status_code,500)

    def test_registration_api_post(self):
        response = self.client.post(path="/mobileRegister/",
                                    data=json.dumps(self.user_data),
                                    content_type="application/json")
        self.assertEqual(response.status_code,201)
        self.user_data['username']=self.user_data['phone']
        user = User.objects.get(username=self.user_data['username'])
        # Check if the company was created
        company = Company.objects.get(user__username=self.user_data['phone'])
        self.assertIsInstance(company,Company)
        # Check if the owner's contact is the same as the user's phone number
        company_contact = CompanyContact.objects.get(company=company,contact_type="owner")
        self.assertEqual(user.username,company_contact[0].contact_number)

Traceback:

======================================================================
ERROR: test_registration_api_post (nextgencatalogs.apps.catalogsapp.tests.AuthTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/nextgencatalogs/apps/catalogsapp/tests.py", line 29, in test_registration_api_post
    user = User.objects.get(username=self.user_data['username'])
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/manager.py", line 151, in get
    return self.get_queryset().get(*args, **kwargs)
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 301, in get
    num = len(clone)
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 77, in __len__
    self._fetch_all()
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 854, in _fetch_all
    self._result_cache = list(self.iterator())
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 220, in iterator
    for row in compiler.results_iter():
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 710, in results_iter
    for rows in self.execute_sql(MULTI):
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 781, in execute_sql
    cursor.execute(sql, params)
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/backends/util.py", line 47, in execute
    self.db.validate_no_broken_transaction()
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/backends/__init__.py", line 365, in validate_no_broken_transaction
    "An error occurred in the current transaction. You can't "
TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block.

----------------------------------------------------------------------

I ran into this same problem myself. This is caused by a quirk in how transactions are handled in the newer versions of Django coupled with a unittest that intentionally triggers an exception.

I had a unittest that checked to make sure a unique column constraint was enforced by purposefully triggering an IntegrityError exception:

def test_constraint(self):
    try:
        # Duplicates should be prevented.
        models.Question.objects.create(domain=self.domain, slug='barks')
        self.fail('Duplicate question allowed.')
    except IntegrityError:
        pass

    do_more_model_stuff()

In Django 1.4, this works fine. However, in Django 1.5/1.6, each test is wrapped in a transaction, so if an exception occurs, it breaks the transaction until you explicitly roll it back. Therefore, any further ORM operations in that transaction, such as my do_more_model_stuff(), will fail with that django.db.transaction.TransactionManagementError exception.

Like caio mentioned in the comments, the solution is to capture your exception with transaction.atomic like:

from django.db import transaction
def test_constraint(self):
    try:
        # Duplicates should be prevented.
        with transaction.atomic():
            models.Question.objects.create(domain=self.domain, slug='barks')
        self.fail('Duplicate question allowed.')
    except IntegrityError:
        pass

That will prevent the purposefully-thrown exception from breaking the entire unittest's transaction.

TransactionManagementError during testing with Django 1.10, TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block. The following are code examples for showing how to use django.db.transaction.TransactionManagementError().They are from open source Python projects. You can vote up the examples you like or vote down the ones you don't like.

Since @mkoistinen never made his comment, an answer, I'll post his suggestion so people won't have to dig through comments.

consider just declaring your test class as a TransactionTestCase rather than just TestCase.

From the docs: A TransactionTestCase may call commit and rollback and observe the effects of these calls on the database.

django.db.transaction.TransactionManagementError Python Example, TransactionManagementError(). They are from open source Python projects. You can vote up the examples you like or vote down the ones you don't like. (Commits are on connection level.) """ if self.transaction_state: del self.transaction_state[-1] else: raise TransactionManagementError( "This code isn't under transaction management") # We will pass the next status (after leaving the previous state # behind) to subclass hook.

If using pytest-django you can pass transaction=True to the django_db decorator to avoid this error.

See https://pytest-django.readthedocs.io/en/latest/database.html#testing-transactions

Django itself has the TransactionTestCase which allows you to test transactions and will flush the database between tests to isolate them. The downside of this is that these tests are much slower to set up due to the required flushing of the database. pytest-django also supports this style of tests, which you can select using an argument to the django_db mark:

@pytest.mark.django_db(transaction=True)
def test_spam():
    pass  # test relying on transactions

django.db.transaction.TransactionManagementError: An error , TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block. #190. You can't " TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block.

For me, the proposed fixes did not work. In my tests, I open some subprocesses with Popen to analyze/lint migrations (e.g. one test checks if there are no model changes).

For me, subclassing from SimpleTestCase instead of TestCase did do the trick.

Note that SimpleTestCase doesn't allow to use the database.

While this does not answer the original question, I hope this helps some people anyway.

django.db.transaction.TransactionManagementError when adding , TransactionManagementError TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the  TransactionManagementError: select_for_update cannot be used outside of a transaction. What are some of the solutions for solving this? django transactions django-1.5 django-1.6

Here is another way to do it, based on the answer to this question:

with transaction.atomic():
    self.assertRaises(IntegrityError, models.Question.objects.create, **{'domain':self.domain, 'slug':'barks'})

Django Exceptions | Django documentation, TransactionManagementError is raised for any and all problems related to database transactions. Testing Framework Exceptions¶. Exceptions provided by the  A transaction is a unit of work performed against the database. It is a set of work (T-SQL statements) that are executed together such as a single unit in a specific logical order as a single unit.

#21540 (TestCase with multiple assertRaises fails with , You can't " TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block. I assume the  Introduction to Transactions. A transaction is a logical unit of work that contains one or more SQL statements. A transaction is an atomic unit. The effects of all the SQL statements in a transaction can be either all committed (applied to the database) or all rolled back (undone from the database).

TransactionManagementError | Форум, Подскажите, плз, стала при добавлении объекта выпадать ошибка TransactionManagementError. Раньше этот же функционал работал. И сейчас на  What is a Database Transaction? A transaction is a logical unit of processing in a DBMS which entails one or more database access operation. In a nutshell, database transactions represent real-world e

TransactionManagementError from inside atomic block, Google Groups allows you to create and participate in online forums and email-​based groups with a rich experience for community conversations. Global transactions enable you to work with multiple transactional resources, typically relational databases and message queues. The application server manages global transactions through the JTA, which is a cumbersome API to use (partly due to its exception model).

Comments
  • From the docs: "A TestCase, on the other hand, does not truncate tables after a test. Instead, it encloses the test code in a database transaction that is rolled back at the end of the test. Both explicit commits like transaction.commit() and implicit ones that may be caused by transaction.atomic() are replaced with a nop operation. This guarantees that the rollback at the end of the test restores the database to its initial state."
  • I found my problem. There was an IntegrityError exception like this "try: ... except IntegrityError: ..." what I had to do is to use the transaction.atomic inside the try-block: "try: with transaction.atomic(): ... except IntegrityError: ..." now everything works fine.
  • docs.djangoproject.com/en/dev/topics/db/transactions and then search for "Wrapping atomic in a try/except block allows for natural handling of integrity errors:"
  • Also consider just declaring your test class as a TransactionTestCase rather than just TestCase.
  • Oh, I found the related document from another question. The document is here.
  • For me, I already had a transaction.atomic() block, but I got this error and I had no idea why. I took this answer's advice and put a nested atomic block inside of my atomic block around the trouble-area. After that, it gave a detailed error of the integrity error I hit, allowing me to fix my code and do what I was trying to do.
  • @mkoistinen TestCase is inheriting from TransactionTestCase so no need to change that. If you don't operate on DB in test use SimpleTestCase.
  • @bns you're missing the point of the comment. Yes TestCase inherits from TransactionTestCase but its behavior is quite different: it wraps each test method in a transaction. TransactionTestCase, on the other hand, is perhaps misleadingly named: it truncates tables to reset the db -- the naming seems to reflect that you can test transactions within a test, not that the test is wrapped as a transaction!
  • +1 for this, but, as the docs say, "Django’s TestCase class is a more commonly used subclass of TransactionTestCase". To answer the original question, shouldn't we use SimpleTestCase instead of TestCase? SimpleTestCase doesn't have the atomic database features.
  • @daigorocub When inheriting from SimpleTestCase, allow_database_queries = True must be added inside the test class, so it doesn't spit an AssertionError("Database queries aren't allowed in SimpleTestCase...",).
  • This is the answer that works best for me as I was trying to test for integrityerror will be raised and then subsequently I needed to run more database save queries
  • I had an issue with this solution, I had initial data in my DB (added by migrations). This solution flush the database, so other tests dependent on this initial data started to fail.
  • get_or_create() works as well, it seems it's the .save() it doesnt like inside a transaction.atomic() decorated function (mine failed with just 1 call in there).