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

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.

From: stackoverflow.com/q/21458387

Back to homepage or read more recommendations: