Нужен пример загрузки файла теста pytest в graphql

Я пытаюсь запустить pytest и загрузить файл на конечную точку /graphql. Я прочитал официальную документацию, но она не охватывает мой вопрос. Тогда я попробовал задать вопрос чатботу. Ответ содержит ошибку. Возникает ошибка "Сервер не запущен"

У меня есть опыт тестирования загрузки файла с помощью djangoREST. Мне не нужно запускать реальный сервер во время тестирования. Поэтому я думаю, что gql сделает то же самое.

query

mutation ExcelUploadFile($file: Upload!, $orderType: String!, $saleOrg: String!, $uploadType: UploadType!, $distributionChannel: String!, $division: String!) {
  uploadExcel(
    file: $file
    orderType: $orderType
    saleOrg: $saleOrg
    uploadType: $uploadType
    distributionChannel: $distributionChannel
    division: $division
  ) {
    status
    rows
    errors {
      field
      code
      message
      __typename
    }
    __typename
  }
}

variables

{
  "file": null,
  "orderType": "ZP30",
  "saleOrg": "0254",
  "uploadType": "GROUP",
  "distributionChannel": "10",
  "division": "00"
}

Вот мой код pytest.

from io import BytesIO

import openpyxl
import pytest
import requests
from django.core.files.uploadedfile import InMemoryUploadedFile
from requests_toolbelt.multipart.encoder import MultipartEncoder

from scgp_po_upload.implementations.excel_upload import validate_excel_upload_file
from scgp_po_upload.tests.fixtures import client_query

# URL of your GraphQL endpoint
GRAPHQL_ENDPOINT = "http://localhost:8000/graphql/"

# Mutation query
UPLOAD_MUTATION = """
mutation ExcelUploadFile($file: Upload!, $orderType: String!, $saleOrg: String!, $uploadType: UploadType!, $distributionChannel: String!, $division: String!) {
    uploadExcel(
        file: $file
        orderType: $orderType
        saleOrg: $saleOrg
        uploadType: $uploadType
        distributionChannel: $distributionChannel
        division: $division
    ) {
        status
        rows
        errors {
            field
            code
            message
        }
    }
}
"""

Главная причина, по которой я не использую client_query из здесь. потому что он использует from graphene_django.utils.testing import graphql_query

@pytest.fixture
def client_query(client):
    def func(*args, **kwargs):
        return graphql_query(*args, **kwargs, client=client)

    return func

graphql_query из graphene_django не поддерживает multipart-formdata. Он поддерживает только application/json

import json
import warnings

from django.test import Client, TestCase

DEFAULT_GRAPHQL_URL = "/graphql/"


def graphql_query(
    query,
    op_name=None,
    input_data=None,
    variables=None,
    headers=None,
    client=None,
    graphql_url=None,
):
    """
    Args:
        query (string)              - GraphQL query to run
        op_name (string)            - If the query is a mutation or named query, you must
                                      supply the op_name.  For annon queries ("{ ... }"),
                                      should be None (default).
        input_data (dict)           - If provided, the $input variable in GraphQL will be set
                                      to this value. If both ``input_data`` and ``variables``,
                                      are provided, the ``input`` field in the ``variables``
                                      dict will be overwritten with this value.
        variables (dict)            - If provided, the "variables" field in GraphQL will be
                                      set to this value.
        headers (dict)              - If provided, the headers in POST request to GRAPHQL_URL
                                      will be set to this value.
        client (django.test.Client) - Test client. Defaults to django.test.Client.
        graphql_url (string)        - URL to graphql endpoint. Defaults to "/graphql".

    Returns:
        Response object from client
    """
    if client is None:
        client = Client()
    if not graphql_url:
        graphql_url = DEFAULT_GRAPHQL_URL

    body = {"query": query}
    if op_name:
        body["operationName"] = op_name
    if variables:
        body["variables"] = variables
    if input_data:
        if "variables" in body:
            body["variables"]["input"] = input_data
        else:
            body["variables"] = {"input": input_data}
    if headers:
        resp = client.post(
            graphql_url, json.dumps(body), content_type="application/json", **headers
        )
    else:
        resp = client.post(
            graphql_url, json.dumps(body), content_type="application/json"
        )
    return resp

def test_upload_blank_content_excel_file(get_admin_user_token, client_query, client) -> None:
    """Expect BE return error and asking on each column."""
    file_path = "artifacts/SEO-8609_blank_content.xlsx"
    variables = {
        "orderType": "ZP30",
        "saleOrg": "0254",
        "uploadType": "GROUP",
        "distributionChannel": "10",
        "division": "00",
    }

    # Prepare the multipart request with the file and variables
    multipart_data = MultipartEncoder(
        fields={
            'operations': (
                None,
                '{"query": "%s", "variables": {"file": null, "orderType": "%s", "saleOrg": "%s", "uploadType": "%s", "distributionChannel": "%s", "division": "%s"}}' %
                (UPLOAD_MUTATION, variables['orderType'], variables['saleOrg'], variables['uploadType'],
                 variables['distributionChannel'], variables['division'])
            ),
            'map': (None, '{"0": ["variables.file"]}'),
            '0': (
            'file', open(file_path, 'rb'), 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
        }
    )
    # Send the request
    headers = {
        'Content-Type': multipart_data.content_type,
        'Authorization': f"Bearer {get_admin_user_token}"
    }
    # AttributeError: 'MultipartEncoder' object has no attribute 'items'
    # response = client.post(path=GRAPHQL_ENDPOINT, headers=headers, data=multipart_data)

    # E           requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=8000): Max retries exceeded with url: /graphql/ (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x140f76df0>: Failed to establish a new connection: [Errno 61] Connection refused'))
    # import requests
    # response = requests.post(url=GRAPHQL_ENDPOINT, headers=headers, data=multipart_data)

    # assert 200 == response.status_code

Когда я использую client.post, я получаю MultipartEncoder объект не имеет атрибутов элементов

Когда я использую requests.post, я получаю ConnectionError

Обновление:

def test_imitage_claude(get_admin_user_token, client_query, client) -> None:
    """Expect BE return error and asking on each column."""
    file_path = "artifacts/SEO-8609_blank_content.xlsx"
    from scgp_po_upload.graphql.enums import UploadType
    variables = {
        "orderType": "ZP30",
        "saleOrg": "0254",
        "uploadType": UploadType.GROUP.value,
        "distributionChannel": "10",
        "division": "00",
        "file": open(file_path, 'rb')
    }
    from graphene.test import Client
    from saleor.graphql.api import Mutation
    my_schema = graphene.Schema(mutation=Mutation)
    client = Client(my_schema)
    res = client.execute(UPLOAD_MUTATION, variable_values=variables)

Я попробовал Client из graphene.test и получил другую ошибку. Ошибка:

res
{'errors': [{'message': "'NoneType' object has no attribute 'is_mutation'", 'locations': [{'line': 3, 'column': 5}], 'path': ['uploadExcel']}], 'data': OrderedDict([('uploadExcel', None)])}

Есть идеи?

Вопрос:
Как записать файл выгрузки pytest в конечную точку graphql?

Вернуться на верх