Issue whith testing django view that execute asynchrnous task (celery): unexpected tests results

19 Views Asked by At

I try to implement tests in Django (DJango/Docker/django-celery/PostgreSQL) but have 2 issues with tests results when view execute celery asynchronous task.

The upload_file ajax view works fine 'in live'.

But in tests, I face 2 issues I can not figure out:

  1. self.assertEqual(1, RCountry.objects.count()) failed whereas self.assertEqual(1, ListRandomization.objects.count()) succeed.
  2. self.assertEqual(1, ListRandomization.objects.count()) succeed even if file_upload file contain more than one record: if file contain 5 rows with differents ran_ide (FK), should create 5 records in Randomization model and self.assertEqual(1, ListRandomization.objects.count()) should failed

If self.assertEqual(1, ListRandomization.objects.count()) succeed that mean that task succeed and as upadte_or_create of the 2 models are in the same atomic transaction, success of one should imply success of the other.

When I look celery logs container, logs confirm task succeed.

#tests.py

    def test_upload_file_ajax_post_valid(self):
        ...
        with tempfile.NamedTemporaryFile(mode='w+', suffix='.csv', delete=False) as temp_file:
            csv_writer = csv.writer(temp_file)
            ...
        with open(temp_file.name, 'rb') as file_upload:
            response = self.client.post(
                reverse('randomization_settings:upload_file'),  
                data={'file': file_upload},
                HTTP_X_REQUESTED_WITH='XMLHttpRequest',
            )        
        # tests
        self.assertEqual(response.status_code, 200)
        self.assertEqual(1, RCountry.objects.count())
        self.assertEqual(1, ListRandomization.objects.count())
        os.remove(temp_file.name)

views.py

def upload_file(request):

    if request.headers.get('x-requested-with') == 'XMLHttpRequest' and request.method == "POST":
        form = FileUploadForm(request.POST, request.FILES)
        if form.is_valid():
            ...
            
            # Update of Country (adm_pay) Randomization (tbl_ran)
            task = upload_list.delay(request.user.username) # asynchronous task   

            ...
            return JsonResponse({"response": "success"}, status=200)
        else:
            return JsonResponse({"response": "form_invalid"}, status=200)
    else:
        return JsonResponse({"response": "ajax_error"}, status=400)

tasks.py

@shared_task(bind=True)
def upload_list(self,user):
    countries = None   
    try:
        currentListOfRandomization = ListRandomization.objects.latest('created_by')
        file_path = os.path.join(settings.MEDIA_ROOT, str(currentListOfRandomization.file))
        df = pd.read_csv(file_path, delimiter=",")
        with transaction.atomic():  
            countries = df.groupby('PAY_IDE')['PAY_IDE_INI'].first().reset_index()
            for index, row in countries.iterrows():
                country, created = RCountry.objects.update_or_create(...)       
            for index, row in df.iterrows():
                Randomization.objects.update_or_create(...)
    except ListRandomization.DoesNotExist:
        currentListOfRandomization = None
        file_path = None
    except FileNotFoundError:
        print(f"File not found at {file_path}")
    return f'task done {timezone.now()}'

0

There are 0 best solutions below