'Send csv file in pytest test using restframework ApiClient

I'm trying to test my view which requires csv_file and one parameter in request, but getting error "details": {"csv_file": ["The submitted data was not a file. Check the encoding type on the form."]} I was trying many ways to make csv and send it but always get the same error. How should I modify my test to make it work?

My test.py

import io
import pytest
from freezegun import freeze_time
from pandas import DataFrame

from marking.personal_area.tests.factories import DatasetFactory
from marking.utils.test.testcases import MarkingTestCase


@freeze_time("2022-04-10T09:00:00+03:00")
@pytest.mark.usefixtures("mock_send_lookalike_task")
class TestPostSamplesCSV(MarkingTestCase):
    url = "/api/pa/samples_csv/"
    maxDiff = None

    def setUp(self):
        DatasetFactory.reset_sequence()
        self.dataset = DatasetFactory()

    def test_ok(self):
        file_headers = ['monitoring_object_id','monitoring_object_name','products','attachment_url','source','shop_network_id']
        file_data = [
            [
                73425,
                "name",
                [
                    {'id': 1,'name': 'name','client_id': 'КА-011188'},
                    {'id': 2,'name': 'name','client_id': 'КА-011186'}
                ], 
                "link",
                "bristol",
                30
            ],
            [
                73425,
                "name",
                [
                    {'id': 1,'name': 'name','client_id': 'КА-011188'},
                    {'id': 2,'name': 'name','client_id': 'КА-011186'}
                ],
                "link",
                "kb",
                20
            ],
            [
                76937,
                "name",
                [
                    {'id': 1,'name': 'name': 'КА-012522'},
                    {'id': 2,'name': 'name','client_id': 'КА-012521'}
                ],
                "name",
                "lenta",
                40
            ]
        ]
        expected = {}
        df = DataFrame(columns=file_headers, data=file_data)
        buf = io.StringIO()
        df.to_csv(buf, index=False)

        bytes_data = buf.getvalue().encode('utf-8')

        csv_file = io.BytesIO(bytes_data)

        data = {
            'csv_file': ('data.csv', csv_file, 'text/csv'),
            'dataset_id': self.dataset.id
        }

        r = self.client.post(self.url, data=data)
        self.assertEqual(200, r.status_code, r.content)
        self.assertEqual(expected, r.json())

My validator.py

from rest_framework import serializers
from rest_framework.exceptions import ValidationError

from marking.personal_area.models import Datasets


class PostSamplesCSVValidator(serializers.Serializer):
    dataset_id = serializers.IntegerField(required=True)
    csv_file = serializers.FileField(required=True, allow_empty_file=False)

    def validate_dataset_id(self, dataset_id):
        if not Datasets.objects.filter(id=dataset_id).exists():
            raise ValidationError(
                "Dataset with id %s doesn't exist" % dataset_id
            )
        return dataset_id

A part of my view.py

from pandas import read_csv
from rest_framework.response import Response
from rest_framework.views import APIView

from marking.personal_area.validators.samples_csv import PostSamplesCSVValidator
from marking.utils.exceptions import InvalidInputParameters
from marking.utils.table import create_or_complete_row


class SamplesCSVView(APIView):
    def post(self, request):
        validator = PostSamplesCSVValidator(data=request.data)
        if not validator.is_valid():
            raise InvalidInputParameters(details=validator.errors)

        csv_file = validator.validated_data["csv_file"]
        dataset_id = validator.validated_data["dataset_id"]

        df = read_csv(csv_file)
        data = {}

        for index, row in df.iterrows():
            create_or_complete_row(
                data,
                row["monitoring_object_id"],
                row["monitoring_object_name"],
                row["products"],
                row["attachment_url"],
                row.get("source", None),
                row.get("shop_network_id", None),
            ) ....


Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source