test_get_response

django_connexion/apis/django_api.py

@@ -7,7 +7,7 @@
 import logging
 
 from connexion.apis.abstract import AbstractAPI
-from connexion.lifecycle import ConnexionRequest
+from connexion.lifecycle import ConnexionRequest, ConnexionResponse
 from connexion.utils import yamldumper
 from django.http import HttpResponse, JsonResponse
 from django.urls import path as django_path
@@ -67,7 +67,6 @@ def _add_operation_internal(self, method, path, operation):
         Adds the operation according to the user framework in use.
         It will be used to register the operation on the user framework router.
         """
-        # print('_add_operation_internal', method, path, operation)
         operation_id = operation.operation_id
         logger.debug('... Adding %s -> %s', method.upper(), operation_id,
                      extra=vars(operation))
@@ -131,10 +130,43 @@ def _is_framework_response(cls, response):
     @classmethod
     def _framework_to_connexion_response(cls, response, mimetype):
         """ Cast framework response class to ConnexionResponse used for schema validation """
+        content_type = response.headers['Content-Type']
+
+        if response.streaming:
+            body = response.streaming_content
+            if not content_type:
+                content_type = 'application/octet-stream'
+        else:
+            body = response.content
+
+        if not mimetype:
+            try:
+                mimetype, _ = content_type.split(';', 1)
+            except ValueError:
+                mimetype = content_type
+
+        return ConnexionResponse(
+            status_code=response.status_code,
+            mimetype=mimetype,
+            content_type=content_type,
+            headers=response.headers,
+            body=body,
+        )
 
     @classmethod
     def _connexion_to_framework_response(cls, response, mimetype, extra_context=None):
         """ Cast ConnexionResponse to framework response class """
+        content_type = response.content_type
+        if not content_type:
+            content_type = f'{mimetype or response.mimetype}; charset=utf-8'
+
+        django_response = HttpResponse(
+            status=response.status_code,
+            content_type=content_type,
+            content=response.body,
+            headers=response.headers,
+        )
+        return django_response
 
     @classmethod
     def _build_response(cls, data, mimetype, content_type=None, status_code=None, headers=None,
@@ -153,7 +185,34 @@ def _build_response(cls, data, mimetype, content_type=None, status_code=None, he
         :return A framework response.
         :rtype Response
         """
-        breakpoint()
+        if cls._is_framework_response(data):
+            return HttpResponse(data, status_code=status_code, headers=headers)
+
+        data, status_code, serialized_mimetype = cls._prepare_body_and_status_code(
+            data=data, mimetype=mimetype, status_code=status_code, extra_context=extra_context)
+
+        if data is None:
+            data = b''
+
+        mimetype = mimetype or serialized_mimetype
+        if content_type is None:
+            if mimetype:
+                content_type = mimetype
+            elif isinstance(data, bytes):
+                content_type = 'application/octet-stream'
+            else:
+                content_type = 'text/plain'
+
+        if isinstance(data, (str, dict)):
+            content_type += '; charset=utf-8'
+
+        kwargs = {
+            'content_type': content_type,
+            'headers': headers,
+            'status': status_code
+        }
+        kwargs = {k: v for k, v in kwargs.items() if v is not None}
+        return HttpResponse(data, **kwargs)
 
     @property
     def urls(self):

django_connexion/apis/django_utils.py

@@ -3,7 +3,7 @@
 import re
 import string
 
-from django.http import HttpResponse
+from django.http.response import HttpResponseBase
 
 PATH_PARAMETER = re.compile(r'\{([^}]*)\}')
 
@@ -75,4 +75,4 @@ def is_django_response(obj: object) -> bool:
     >>> is_django_response(flask.Response())
     True
     """
-    return isinstance(obj, HttpResponse)
+    return isinstance(obj, HttpResponseBase)

django_connexion/tests/api/test_get_response.py

@@ -0,0 +1,185 @@
+import json
+
+import pytest
+from django_connexion.apis.django_api import DjangoApi
+from connexion.lifecycle import ConnexionResponse
+
+from django.http import HttpResponse, StreamingHttpResponse
+
+
+@pytest.fixture(scope='module')
+def api(django_api_spec_dir):
+    yield DjangoApi(specification=django_api_spec_dir / 'swagger_secure.yaml')
+
+
+def test_get_response_from_django_response(api):
+    django_response = HttpResponse(
+        'foo', status=201, headers={'X-header': 'value'}, content_type='text/plain; charset=utf-8')
+    response = api.get_response(django_response)
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 201
+    assert response.content == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8',
+        'X-header': 'value'
+    }
+
+
+def test_get_response_from_django_stream_response(api):
+    django_stream_response = StreamingHttpResponse(
+        status=201, content_type='application/octet-stream', headers={'X-header': 'value'}
+    )
+    response = api.get_response(django_stream_response)
+    assert isinstance(response, StreamingHttpResponse)
+    assert response.status_code == 201
+    assert dict(response.headers) == {
+        'Content-Type': 'application/octet-stream', 'X-header': 'value'
+    }
+
+
+def test_get_response_from_connexion_response(api):
+    connexion_response = ConnexionResponse(
+        status_code=201, mimetype='text/plain', body='foo', headers={'X-header': 'value'})
+    response = api.get_response(connexion_response)
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 201
+    assert response.content == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'
+    }
+
+
+def test_get_response_from_string(api):
+    response = api.get_response('foo')
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    assert response.content == b'foo'
+    assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
+
+
+def test_get_response_from_string_tuple(api):
+    response = api.get_response(('foo',))
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    assert response.content == b'foo'
+    assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
+
+
+def test_get_response_from_string_status(api):
+    response = api.get_response(('foo', 201))
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 201
+    assert response.content == b'foo'
+    assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
+
+
+def test_get_response_from_string_headers(api):
+    response = api.get_response(('foo', {'X-header': 'value'}))
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    assert response.content == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'
+    }
+
+
+def test_get_response_from_string_status_headers(api):
+    response = api.get_response(('foo', 201, {'X-header': 'value'}))
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 201
+    assert response.content == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'
+    }
+
+
+def test_get_response_from_dict(api):
+    response = api.get_response({'foo': 'bar'})
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    # odd, yes. but backwards compatible. see test_response_with_non_str_and_non_json_body in
+    # tests/aiohttp/test_aiohttp_simple_api.py
+    # TODO: This should be made into JSON when aiohttp and flask serialization can be harmonized.
+    assert response.content == b"{'foo': 'bar'}"
+    assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
+
+
+def test_get_response_from_dict_json(api):
+    response = api.get_response({'foo': 'bar'}, mimetype='application/json')
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    assert json.loads(response.content.decode()) == {"foo": "bar"}
+    assert dict(response.headers) == {'Content-Type': 'application/json; charset=utf-8'}
+
+
+def test_get_response_no_data(api):
+    response = api.get_response(None, mimetype='application/json')
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 204
+    assert response.content == b''
+    assert dict(response.headers) == {'Content-Type': 'application/json'}
+
+
+def test_get_response_binary_json(api):
+    response = api.get_response(b'{"foo":"bar"}', mimetype='application/json')
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    assert json.loads(response.content.decode()) == {"foo": "bar"}
+    assert dict(response.headers) == {'Content-Type': 'application/json'}
+
+
+def test_get_response_binary_no_mimetype(api):
+    response = api.get_response(b'{"foo":"bar"}')
+    assert isinstance(response, HttpResponse)
+    assert response.status_code == 200
+    assert response.content == b'{"foo":"bar"}'
+    assert dict(response.headers) == {'Content-Type': 'application/octet-stream'}
+
+
+def test_get_connexion_response_from_django_response(api):
+    django_response = HttpResponse(
+        'foo', status=201, content_type='text/plain; charset=utf-8', headers={'X-header': 'value'}
+    )
+    response = api.get_connexion_response(django_response)
+    assert isinstance(response, ConnexionResponse)
+    assert response.status_code == 201
+    assert response.body == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'
+    }
+
+
+def test_get_connexion_response_from_connexion_response(api):
+    connexion_response = ConnexionResponse(
+        status_code=201, content_type='text/plain', body='foo', headers={'X-header': 'value'}
+    )
+    response = api.get_connexion_response(connexion_response)
+    assert isinstance(response, ConnexionResponse)
+    assert response.status_code == 201
+    assert response.body == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'
+    }
+
+
+def test_get_connexion_response_from_tuple(api):
+    response = api.get_connexion_response(('foo', 201, {'X-header': 'value'}))
+    assert isinstance(response, ConnexionResponse)
+    assert response.status_code == 201
+    assert response.body == b'foo'
+    assert dict(response.headers) == {
+        'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'
+    }
+
+
+def test_get_connexion_response_from_django_stream_response(api):
+    django_stream_response = StreamingHttpResponse(
+        status=201, content_type='application/octet-stream', headers={'X-header': 'value'}
+    )
+    response = api.get_connexion_response(django_stream_response)
+    assert isinstance(response, ConnexionResponse)
+    assert response.status_code == 201
+    assert bytes(response.body) == b''
+    assert dict(response.headers) == {
+        'Content-Type': 'application/octet-stream', 'X-header': 'value'
+    }

django_connexion/tests/conftest.py

@@ -0,0 +1,11 @@
+import pathlib
+
+import pytest
+
+TEST_FOLDER = pathlib.Path(__file__).parent
+FIXTURES_FOLDER = TEST_FOLDER / 'fixtures'
+
+
+@pytest.fixture(scope='session')
+def django_api_spec_dir():
+    return FIXTURES_FOLDER / 'django'

django_connexion/tests/fakeapi/auth.py

@@ -0,0 +1,14 @@
+import json
+
+
+def fake_basic_auth(username, password, required_scopes=None):
+    if username == password:
+        return {'uid': username}
+    return None
+
+
+def fake_json_auth(token, required_scopes=None):
+    try:
+        return json.loads(token)
+    except ValueError:
+        return None