Added support for octet-stream content type (#116) (#157)

* Added support for octet-stream content type (#116)

Co-authored-by: Ethan Mann <[email protected]>

Author: emann

CHANGELOG.md

@@ -5,6 +5,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+## 0.5.4 - Unreleased
+### Additions
+- Added support for octet-stream content type (#116)
+
+
 ## 0.5.3 - 2020-08-13
 ### Security
 - All values that become file/directory names are sanitized to address path traversal vulnerabilities (CVE-2020-15141)

end_to_end_tests/fastapi_app/__init__.py

@@ -7,6 +7,7 @@
 
 from fastapi import APIRouter, Body, FastAPI, File, Header, Query, UploadFile
 from pydantic import BaseModel
+from starlette.responses import FileResponse
 
 app = FastAPI(title="My Test API", description="An API for testing openapi-python-client",)
 
@@ -85,6 +86,15 @@ def test_defaults(
     return
 
 
+@test_router.get(
+    "/test_octet_stream",
+    response_class=FileResponse,
+    responses={200: {"content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}}}},
+)
+def test_octet_stream():
+    return
+
+
 app.include_router(test_router, prefix="/tests", tags=["tests"])
 
 

end_to_end_tests/fastapi_app/openapi.json

@@ -334,6 +334,28 @@
                     }
                 }
             }
+        },
+        "/tests/test_octet_stream": {
+            "get": {
+                "tags": [
+                    "tests"
+                ],
+                "summary": "Test Octet Stream",
+                "operationId": "test_octet_stream_tests_test_octet_stream_get",
+                "responses": {
+                    "200": {
+                        "description": "Successful Response",
+                        "content": {
+                            "application/octet-stream": {
+                                "schema": {
+                                    "type": "string",
+                                    "format": "binary"
+                                }
+                            }
+                        }
+                    }
+                }
+            }
         }
     },
     "components": {

end_to_end_tests/golden-master/my_test_api_client/api/tests.py

@@ -172,3 +172,18 @@ def test_defaults_tests_test_defaults_post(
         return HTTPValidationError.from_dict(cast(Dict[str, Any], response.json()))
     else:
         raise ApiResponseError(response=response)
+
+
+def test_octet_stream_tests_test_octet_stream_get(*, client: Client,) -> bytes:
+
+    """  """
+    url = "{}/tests/test_octet_stream".format(client.base_url)
+
+    headers: Dict[str, Any] = client.get_headers()
+
+    response = httpx.get(url=url, headers=headers,)
+
+    if response.status_code == 200:
+        return bytes(response.content)
+    else:
+        raise ApiResponseError(response=response)

end_to_end_tests/golden-master/my_test_api_client/async_api/tests.py

@@ -176,3 +176,19 @@ async def test_defaults_tests_test_defaults_post(
         return HTTPValidationError.from_dict(cast(Dict[str, Any], response.json()))
     else:
         raise ApiResponseError(response=response)
+
+
+async def test_octet_stream_tests_test_octet_stream_get(*, client: Client,) -> bytes:
+
+    """  """
+    url = "{}/tests/test_octet_stream".format(client.base_url,)
+
+    headers: Dict[str, Any] = client.get_headers()
+
+    async with httpx.AsyncClient() as _client:
+        response = await _client.get(url=url, headers=headers,)
+
+    if response.status_code == 200:
+        return bytes(response.content)
+    else:
+        raise ApiResponseError(response=response)

openapi_python_client/parser/responses.py

@@ -70,6 +70,21 @@ def constructor(self) -> str:
         return f"{self.python_type}(response.text)"
 
 
+@dataclass
+class BytesResponse(Response):
+    """ Response is a basic type """
+
+    python_type: str = "bytes"
+
+    def return_string(self) -> str:
+        """ How this Response should be represented as a return type """
+        return self.python_type
+
+    def constructor(self) -> str:
+        """ How the return value of this response should be constructed """
+        return f"{self.python_type}(response.content)"
+
+
 openapi_types_to_python_type_strings = {
     "string": "str",
     "number": "float",
@@ -88,6 +103,8 @@ def response_from_data(*, status_code: int, data: Union[oai.Response, oai.Refere
     schema_data = None
     if "application/json" in content:
         schema_data = data.content["application/json"].media_type_schema
+    elif "application/octet-stream" in content:
+        return BytesResponse(status_code=status_code)
     elif "text/html" in content:
         schema_data = data.content["text/html"].media_type_schema
 

tests/test_openapi_parser/test_responses.py

@@ -94,6 +94,22 @@ def test_constructor(self):
         assert r.constructor() == "bool(response.text)"
 
 
+class TestBytesResponse:
+    def test_return_string(self):
+        from openapi_python_client.parser.responses import BytesResponse
+
+        b = BytesResponse(200)
+
+        assert b.return_string() == "bytes"
+
+    def test_constructor(self):
+        from openapi_python_client.parser.responses import BytesResponse
+
+        b = BytesResponse(200)
+
+        assert b.constructor() == "bytes(response.content)"
+
+
 class TestResponseFromData:
     def test_response_from_data_no_content(self, mocker):
         from openapi_python_client.parser.responses import response_from_data
@@ -199,3 +215,15 @@ def test_response_from_dict_unsupported_type(self):
         )
 
         assert response_from_data(status_code=200, data=data) == ParseError(data=data, detail="Unrecognized type BLAH")
+
+    def test_response_from_data_octet_stream(self, mocker):
+        status_code = mocker.MagicMock(autospec=int)
+        data = oai.Response.construct(
+            content={"application/octet-stream": oai.MediaType.construct(media_type_schema=mocker.MagicMock())}
+        )
+        BytesResponse = mocker.patch(f"{MODULE_NAME}.BytesResponse")
+        from openapi_python_client.parser.responses import response_from_data
+
+        response = response_from_data(status_code=status_code, data=data)
+
+        assert response == BytesResponse()