python requests_toolbelt模块传输csv文件

import os
import requests
from urllib import parse
from requests_toolbelt import MultipartEncoder
from utils.handle_yaml import test_info
from utils.log_record import logger

env1 = test_info.env_info("env1")

file = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "test_data", "upload_cameraInfo.csv")


class TestUploadCameraInfo(object):
    """相机信息信息"""

    def test_upload_camera_info_succs(self, login):
        # 导入相机
        env1["headers"]["accessToken"] = login
        res_project = requests.get(parse.urljoin(env1["base_url"], "/api/project/query"),
                                   headers=env1["headers"]).json()
        site_id = res_project.get("data").get("siteId")

        url = parse.urljoin(env1["base_url"], "/api/camera/upload_camera_info/%s" % site_id)
        logger(url=url, method="POST")
        # 数据组装
        data = MultipartEncoder(fields={"file": ("upload_cameraInfo.csv", open(file, "rb"), "text/csv")})
        # 重写Content-Type
        env1["headers"]["Content-Type"] = data.content_type
        logger(headers=env1["headers"]["Content-Type"], data=data)

        res = requests.post(url, headers=env1["headers"], data=data).json()

        logger(response=str(res))

        exp_data = {
            "code": 0,
            "data": {},
            "message": "success"
        }

        assert res == exp_data