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:
self.assertEqual(1, RCountry.objects.count())failed whereasself.assertEqual(1, ListRandomization.objects.count())succeed.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 andself.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()}'