Generalize testing.file_graphql_query

graphene_file_upload/django/testing.py

@@ -1,19 +1,30 @@
 """Use django's Test client to run file based graphql test."""
-import json
-
 from django.test import Client, TestCase
+from .. import testingbase
 
 DEFAULT_GRAPHQL_URL = "/graphql/"
 
 
+def create_client_post(client):
+    """
+    Create client_post function for testingbase.file_graphql_query from django.test.Client
+    """
+    def client_post(graphql_url, data, files, headers=None):
+        data.update(files)
+        if headers:
+            response = client.post(graphql_url, data, headers=headers)
+        else:
+            response = client.post(graphql_url, data)
+        return response
+    return client_post
+
+
 def file_graphql_query(
         query, op_name=None, input_data=None, variables=None,
         headers=None, files=None, client=None, graphql_url=None,
 ):
     """
-    Based on: https://www.sam.today/blog/testing-graphql-with-graphene-django/
-
-    Perform file based mutation.
+    Perform file based mutation, using django.test.Client
 
     :param str query: GraphQL query to run
     :param str op_name: If the query is a mutation or named query, you must
@@ -31,107 +42,12 @@ def file_graphql_query(
     :param str graphql_url: URL to graphql endpoint. Defaults to "/graphql"
     :return: Response object from client
     """
-
-    if not files:
-        raise ValueError('Missing required argument "files".')
-
     client = client or Client()
-    headers = headers or {}
-    variables = variables or {}
-    graphql_url = graphql_url or DEFAULT_GRAPHQL_URL
-    map_ = {}
-
-    for key in files.keys():
-        map_[key] = ['variables.{key}'.format(key=key)]
-        if key not in variables:
-            variables[key] = None
-
-    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}
-
-    data = {
-        'operations': json.dumps(body),
-        'map': json.dumps(map_),
-    }
-
-    data.update(files)
-    if headers:
-        resp = client.post(graphql_url, data, **headers)
-    else:
-        resp = client.post(graphql_url, data)
-
-    return resp
-
-
-class GraphQLFileUploadTestMixin:
-    """GraphQL file upload test mixin."""
-
-    # URL to graphql endpoint
-    GRAPHQL_URL = DEFAULT_GRAPHQL_URL
-
-    def file_query(
-            self, query, op_name=None, input_data=None, files=None,
-            variables=None, headers=None,
-    ):
-        """
-        Perform file based mutation.
-
-        :param str query: GraphQL query to run
-        :param str op_name: If the query is a mutation or named query, you must
-            supply the op_name. For annon queries ("{ ... }"), should be None (default).
-        :param dict input_data: 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. Defaults to None.
-        :param dict variables: If provided, the "variables" field in GraphQL will be
-            set to this value. Defaults to None.
-        :param dict headers: If provided, the headers in POST request to GRAPHQL_URL
-            will be set to this value. Defaults to None
-        :param dict files: Files to be sent via request.FILES. Defaults to None.
-        :param django.test.Client client: Test client. Defaults to django.test.Client.
-        :param str graphql_url: URL to graphql endpoint. Defaults to "/graphql"
-        :return: Response object from client
-        """
-        return file_graphql_query(
-            query,
-            op_name=op_name,
-            input_data=input_data,
-            variables=variables,
-            headers=headers,
-            files=files,
-            client=self.client,
-            graphql_url=self.GRAPHQL_URL,
-        )
-
-    def assertResponseNoErrors(self, resp, msg=None):  # pylint: disable=C0103
-        """
-        Assert that the call went through correctly. 200 means the syntax is ok,
-        if there are no `errors`, the call was fine.
-        :param Response resp: HttpResponse
-        :param str msg: Error message.
-        """
-        content = json.loads(resp.content)
-        self.assertEqual(resp.status_code, 200, msg or content)
-        self.assertNotIn("errors", list(content.keys()), msg or content)
-
-    def assertResponseHasErrors(self, resp, msg=None):  # pylint: disable=C0103
-        """
-        Assert that the call was failing. Take care: Even with errors,
-        GraphQL returns status 200!
-        :param Response resp: HttpResponse
-        :param str msg: Error message.
-        """
-        content = json.loads(resp.content)
-        self.assertIn("errors", list(content.keys()), msg or content)
+    client_post = create_client_post(client)
+    return testingbase.file_graphql_query(
+        query, op_name, input_data, variables, headers, files, client_post, graphql_url)
 
 
-class GraphQLFileUploadTestCase(GraphQLFileUploadTestMixin, TestCase):
-    pass
+class GraphQLFileUploadTestCase(testingbase.GraphQLFileUploadTestMixin, TestCase):
+    def setUp(self) -> None:
+        self.client_post = create_client_post(self.client)

graphene_file_upload/flask/testing.py

@@ -0,0 +1,46 @@
+"""Use flask's Test client to run file based graphql test."""
+from .. import testingbase
+
+DEFAULT_GRAPHQL_URL = "/graphql/"
+
+
+def create_client_post(client):
+    """
+    Create client_post function for testingbase.file_graphql_query from django.test.Client
+    """
+    def client_post(graphql_url, data, files, headers=None):
+        data.update(files)
+        if headers:
+            response = client.post(graphql_url, data=data, headers=headers)
+        else:
+            response = client.post(graphql_url, data=data)
+        return response
+    return client_post
+
+
+def file_graphql_query(
+        query, op_name=None, input_data=None, variables=None,
+        headers=None, files=None, client=None, graphql_url=None,
+):
+    """
+    Perform file based mutation, using django.test.Client
+
+    :param str query: GraphQL query to run
+    :param str op_name: If the query is a mutation or named query, you must
+        supply the op_name. For annon queries ("{ ... }"), should be None (default).
+    :param dict input_data: 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. Defaults to None.
+    :param dict variables: If provided, the "variables" field in GraphQL will be
+        set to this value. Defaults to None.
+    :param dict headers: If provided, the headers in POST request to GRAPHQL_URL
+        will be set to this value. Defaults to None
+    :param dict files: Files to be sent via request.FILES. Defaults to None.
+    :param  flask.testing.FlaskClient: Test client. Defaults to None.
+    :param str graphql_url: URL to graphql endpoint. Defaults to "/graphql"
+    :return: Response object from client
+    """
+    client_post = create_client_post(client)
+    return testingbase.file_graphql_query(
+        query, op_name, input_data, variables, headers, files, client_post, graphql_url)

graphene_file_upload/testingbase.py

@@ -0,0 +1,134 @@
+"""Base of using Test client to run file based graphql test."""
+import json
+DEFAULT_GRAPHQL_URL = "/graphql/"
+
+
+def file_graphql_query(
+        query, op_name=None, input_data=None, variables=None,
+        headers=None, files=None, client_post=None, graphql_url=None,
+):
+    """
+    Based on: https://www.sam.today/blog/testing-graphql-with-graphene-django/
+
+    Perform file based mutation.
+
+    :param str query: GraphQL query to run
+    :param str op_name: If the query is a mutation or named query, you must
+        supply the op_name. For annon queries ("{ ... }"), should be None (default).
+    :param dict input_data: 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. Defaults to None.
+    :param dict variables: If provided, the "variables" field in GraphQL will be
+        set to this value. Defaults to None.
+    :param dict headers: If provided, the headers in POST request to GRAPHQL_URL
+        will be set to this value. Defaults to None
+    :param dict files: Files to be sent via request.FILES. Defaults to None.
+    :param callable client_post: Test client wrapper function.
+                It receives graphql endpoint(str), data(dict), files(dict), headers(dict).
+                When called, it sends POST request using framework specific test client.
+                Defaults to None.
+    :param str graphql_url: URL to graphql endpoint. Defaults to "/graphql"
+    :return: Response object from client
+    """
+
+    if not files:
+        raise ValueError('Missing required argument "files".')
+
+    if not client_post:
+        raise ValueError('Missing required argument "client_post".')
+
+    headers = headers or {}
+    variables = variables or {}
+    graphql_url = graphql_url or DEFAULT_GRAPHQL_URL
+    map_ = {}
+
+    for key in files.keys():
+        map_[key] = [f'variables.{key}']
+        if key not in variables:
+            variables[key] = None
+
+    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}
+
+    data = {
+        'operations': json.dumps(body),
+        'map': json.dumps(map_),
+    }
+
+    # data.update(files)
+    if headers:
+        resp = client_post(graphql_url, data, files ** headers)
+    else:
+        resp = client_post(graphql_url, data, files)
+
+    return resp
+
+
+class GraphQLFileUploadTestMixin:
+    """GraphQL file upload test mixin."""
+
+    # URL to graphql endpoint
+    GRAPHQL_URL = DEFAULT_GRAPHQL_URL
+
+    def file_query(
+            self, query, op_name=None, input_data=None, files=None,
+            variables=None, headers=None,
+    ):
+        """
+        Perform file based mutation.
+        :param str query: GraphQL query to run
+        :param str op_name: If the query is a mutation or named query, you must
+            supply the op_name. For annon queries ("{ ... }"), should be None (default).
+        :param dict input_data: 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. Defaults to None.
+        :param dict variables: If provided, the "variables" field in GraphQL will be
+            set to this value. Defaults to None.
+        :param dict headers: If provided, the headers in POST request to GRAPHQL_URL
+            will be set to this value. Defaults to None
+        :param dict files: Files to be sent via request.FILES. Defaults to None.
+        :param django.test.Client client: Test client. Defaults to django.test.Client.
+        :param str graphql_url: URL to graphql endpoint. Defaults to "/graphql"
+        :return: Response object from client
+        """
+        return file_graphql_query(
+            query,
+            op_name=op_name,
+            input_data=input_data,
+            variables=variables,
+            headers=headers,
+            files=files,
+            client_post=self.client_post,
+            graphql_url=self.GRAPHQL_URL,
+        )
+
+    def assertResponseNoErrors(self, resp, msg=None):  # pylint: disable=C0103
+        """
+        Assert that the call went through correctly. 200 means the syntax is ok,
+        if there are no `errors`, the call was fine.
+        :param Response resp: HttpResponse
+        :param str msg: Error message.
+        """
+        content = json.loads(resp.content)
+        self.assertEqual(resp.status_code, 200, msg or content)
+        self.assertNotIn("errors", list(content.keys()), msg or content)
+
+    def assertResponseHasErrors(self, resp, msg=None):  # pylint: disable=C0103
+        """
+        Assert that the call was failing. Take care: Even with errors,
+        GraphQL returns status 200!
+        :param Response resp: HttpResponse
+        :param str msg: Error message.
+        """
+        content = json.loads(resp.content)
+        self.assertIn("errors", list(content.keys()), msg or content)

tests/test_flask.py

@@ -6,6 +6,8 @@
 from .flask_app import create_app
 from .schema import schema
 
+from graphene_file_upload.flask.testing import file_graphql_query
+
 
 def response_utf8_json(resp):
     return json.loads(resp.data.decode())
@@ -63,3 +65,45 @@ def test_single_file(client, file_text, expected_first_line):
             },
         }
     }
+
+
+@pytest.mark.parametrize(
+    'client,file_text,expected_first_line',
+    (
+        (None, u'Fake Data\nLine2\n', u'Fake Data'),
+        # Try the fire emoji
+        (None, u'\U0001F525\nLine2\nLine3\n', u'\U0001F525'),
+    ),
+    indirect=['client']
+)
+def test_file_graphql_query(client, file_text, expected_first_line):
+
+    query = '''
+        mutation testMutation($file: Upload!) {
+            myUpload(fileIn: $file) {
+                ok
+                firstLine
+            }
+        }
+    '''
+
+    with NamedTemporaryFile() as t_file:
+        t_file.write(file_text.encode('utf-8'))
+        t_file.seek(0)
+
+        response = file_graphql_query(
+            query,
+            op_name='testMutation',
+            files={'file': t_file},
+            client=client,
+            graphql_url='/graphql',
+        )
+    assert response.status_code == 200
+    assert response_utf8_json(response) == {
+        'data': {
+            'myUpload': {
+                'ok': True,
+                'firstLine': expected_first_line,
+            },
+        }
+    }