Test Endpoint

#3

Author: dbanty

openapi_python_client/openapi_parser/openapi.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from enum import Enum
 from typing import Any, Dict, Generator, Iterable, List, Optional, Set
 
@@ -52,14 +52,14 @@ class Endpoint:
     method: str
     description: Optional[str]
     name: str
-    query_parameters: List[Property]
-    path_parameters: List[Property]
-    responses: List[Response]
     requires_security: bool
-    form_body_reference: Optional[Reference]
-    json_body: Optional[Property]
     tag: str
-    relative_imports: Set[str]
+    relative_imports: Set[str] = field(default_factory=set)
+    query_parameters: List[Property] = field(default_factory=list)
+    path_parameters: List[Property] = field(default_factory=list)
+    responses: List[Response] = field(default_factory=list)
+    form_body_reference: Optional[Reference] = None
+    json_body: Optional[Property] = None
 
     @staticmethod
     def parse_request_form_body(body: Dict[str, Any], /) -> Optional[Reference]:
@@ -79,62 +79,61 @@ def parse_request_json_body(body: Dict[str, Any], /) -> Optional[Property]:
             return property_from_dict("json_body", required=True, data=json_body["schema"])
         return None
 
-    @staticmethod
-    def from_data(*, data: Dict[str, Any], path: str, method: str) -> Endpoint:
-        """ Construct an endpoint from the OpenAPI data """
-        query_parameters: List[Property] = []
-        path_parameters: List[Property] = []
-        responses: List[Response] = []
-        tag = data.get("tags", ["default"])[0]
-        relative_imports: Set[str] = set()
+    def _add_body(self, data: Dict[str, Any]) -> None:
+        """ Adds form or JSON body to Endpoint if included in data """
+        if "requestBody" not in data:
+            return
+
+        self.form_body_reference = Endpoint.parse_request_form_body(data["requestBody"])
+        self.json_body = Endpoint.parse_request_json_body(data["requestBody"])
+
+        if self.form_body_reference:
+            self.relative_imports.add(import_string_from_reference(self.form_body_reference, prefix="..models"))
+        if (
+            self.json_body is not None
+            and isinstance(self.json_body, (ListProperty, RefProperty, EnumProperty))
+            and self.json_body.reference is not None
+        ):
+            self.relative_imports.add(import_string_from_reference(self.json_body.reference, prefix="..models"))
+
+    def _add_responses(self, data: Dict[str, Any]) -> None:
+        for code, response_dict in data["responses"].items():
+            response = response_from_dict(status_code=int(code), data=response_dict)
+            if isinstance(response, (RefResponse, ListRefResponse)):
+                self.relative_imports.add(import_string_from_reference(response.reference, prefix="..models"))
+            self.responses.append(response)
 
+    def _add_parameters(self, data: Dict[str, Any]) -> None:
         for param_dict in data.get("parameters", []):
             prop = property_from_dict(
                 name=param_dict["name"], required=param_dict["required"], data=param_dict["schema"]
             )
             if isinstance(prop, (ListProperty, RefProperty, EnumProperty)) and prop.reference:
-                relative_imports.add(import_string_from_reference(prop.reference, prefix="..models"))
+                self.relative_imports.add(import_string_from_reference(prop.reference, prefix="..models"))
             if param_dict["in"] == ParameterLocation.QUERY:
-                query_parameters.append(prop)
+                self.query_parameters.append(prop)
             elif param_dict["in"] == ParameterLocation.PATH:
-                path_parameters.append(prop)
+                self.path_parameters.append(prop)
             else:
                 raise ValueError(f"Don't know where to put this parameter: {param_dict}")
 
-        for code, response_dict in data["responses"].items():
-            response = response_from_dict(status_code=int(code), data=response_dict)
-            if isinstance(response, (RefResponse, ListRefResponse)):
-                relative_imports.add(import_string_from_reference(response.reference, prefix="..models"))
-            responses.append(response)
-        form_body_reference = None
-        json_body = None
-        if "requestBody" in data:
-            form_body_reference = Endpoint.parse_request_form_body(data["requestBody"])
-            json_body = Endpoint.parse_request_json_body(data["requestBody"])
-
-        if form_body_reference:
-            relative_imports.add(import_string_from_reference(form_body_reference, prefix="..models"))
-        if (
-            json_body is not None
-            and isinstance(json_body, (ListProperty, RefProperty, EnumProperty))
-            and json_body.reference is not None
-        ):
-            relative_imports.add(import_string_from_reference(json_body.reference, prefix="..models"))
+    @staticmethod
+    def from_data(*, data: Dict[str, Any], path: str, method: str) -> Endpoint:
+        """ Construct an endpoint from the OpenAPI data """
 
-        return Endpoint(
+        endpoint = Endpoint(
             path=path,
             method=method,
             description=data.get("description"),
             name=data["operationId"],
-            query_parameters=query_parameters,
-            path_parameters=path_parameters,
-            responses=responses,
-            form_body_reference=form_body_reference,
-            json_body=json_body,
             requires_security=bool(data.get("security")),
-            tag=tag,
-            relative_imports=relative_imports,
+            tag=data.get("tags", ["default"])[0],
         )
+        endpoint._add_parameters(data)
+        endpoint._add_responses(data)
+        endpoint._add_body(data)
+
+        return endpoint
 
 
 @dataclass

tests/test_openapi_parser/test_openapi.py

@@ -226,6 +226,191 @@ def test_parse_request_json_body_no_data(self):
 
         assert result is None
 
+    def test_add_body_no_data(self, mocker):
+        from openapi_python_client.openapi_parser.openapi import Endpoint
+
+        parse_request_form_body = mocker.patch.object(Endpoint, "parse_request_form_body")
+        endpoint = Endpoint(
+            path="path",
+            method="method",
+            description=None,
+            name="name",
+            requires_security=False,
+            tag="tag",
+            relative_imports={"import_3"},
+        )
+
+        endpoint._add_body({})
+
+        parse_request_form_body.assert_not_called()
+
+    def test_add_body_happy(self, mocker):
+        from openapi_python_client.openapi_parser.openapi import Endpoint, Reference, RefProperty
+
+        request_body = mocker.MagicMock()
+        form_body_reference = Reference(ref="a")
+        parse_request_form_body = mocker.patch.object(
+            Endpoint, "parse_request_form_body", return_value=form_body_reference
+        )
+        json_body = RefProperty(name="name", required=True, default=None, reference=Reference("b"))
+        parse_request_json_body = mocker.patch.object(Endpoint, "parse_request_json_body", return_value=json_body)
+        import_string_from_reference = mocker.patch(
+            f"{MODULE_NAME}.import_string_from_reference", side_effect=["import_1", "import_2"]
+        )
+
+        endpoint = Endpoint(
+            path="path",
+            method="method",
+            description=None,
+            name="name",
+            requires_security=False,
+            tag="tag",
+            relative_imports={"import_3"},
+        )
+
+        endpoint._add_body({"requestBody": request_body})
+
+        parse_request_form_body.assert_called_once_with(request_body)
+        parse_request_json_body.assert_called_once_with(request_body)
+        import_string_from_reference.assert_has_calls(
+            [mocker.call(form_body_reference, prefix="..models"), mocker.call(json_body.reference, prefix="..models")]
+        )
+        assert endpoint.relative_imports == {"import_1", "import_2", "import_3"}
+        assert endpoint.json_body == json_body
+        assert endpoint.form_body_reference == form_body_reference
+
+    def test__add_responses(self, mocker):
+        from openapi_python_client.openapi_parser.openapi import Endpoint, RefResponse, Reference
+
+        response_1_data = mocker.MagicMock()
+        response_2_data = mocker.MagicMock()
+        data = {"responses": {"200": response_1_data, "404": response_2_data,}}
+        endpoint = Endpoint(
+            path="path",
+            method="method",
+            description=None,
+            name="name",
+            requires_security=False,
+            tag="tag",
+            relative_imports={"import_3"},
+        )
+        ref_1 = Reference(ref="ref_1")
+        ref_2 = Reference(ref="ref_2")
+        response_1 = RefResponse(status_code=200, content_type="application/json", reference=ref_1)
+        response_2 = RefResponse(status_code=404, content_type="application/json", reference=ref_2)
+        response_from_dict = mocker.patch(f"{MODULE_NAME}.response_from_dict", side_effect=[response_1, response_2])
+        import_string_from_reference = mocker.patch(
+            f"{MODULE_NAME}.import_string_from_reference", side_effect=["import_1", "import_2"]
+        )
+
+        endpoint._add_responses(data)
+
+        response_from_dict.assert_has_calls(
+            [mocker.call(status_code=200, data=response_1_data), mocker.call(status_code=404, data=response_2_data),]
+        )
+        import_string_from_reference.assert_has_calls(
+            [mocker.call(ref_1, prefix="..models"), mocker.call(ref_2, prefix="..models"),]
+        )
+        assert endpoint.responses == [response_1, response_2]
+        assert endpoint.relative_imports == {"import_1", "import_2", "import_3"}
+
+    def test__add_parameters_handles_no_params(self):
+        from openapi_python_client.openapi_parser.openapi import Endpoint
+
+        endpoint = Endpoint(
+            path="path", method="method", description=None, name="name", requires_security=False, tag="tag",
+        )
+        endpoint._add_parameters({})  # Just checking there's no exception here
+
+    def test__add_parameters_fail_loudly_when_location_not_supported(self, mocker):
+        from openapi_python_client.openapi_parser.openapi import Endpoint
+
+        endpoint = Endpoint(
+            path="path", method="method", description=None, name="name", requires_security=False, tag="tag",
+        )
+        mocker.patch(f"{MODULE_NAME}.property_from_dict")
+
+        with pytest.raises(ValueError):
+            endpoint._add_parameters(
+                {"parameters": [{"name": "test", "required": True, "schema": mocker.MagicMock(), "in": "cookie"}]}
+            )
+
+    def test__add_parameters_happy(self, mocker):
+        from openapi_python_client.openapi_parser.openapi import Endpoint, EnumProperty
+
+        endpoint = Endpoint(
+            path="path",
+            method="method",
+            description=None,
+            name="name",
+            requires_security=False,
+            tag="tag",
+            relative_imports={"import_3"},
+        )
+        path_prop = EnumProperty(name="path_enum", required=True, default=None, values={})
+        query_prop = EnumProperty(name="query_enum", required=False, default=None, values={})
+        propety_from_dict = mocker.patch(f"{MODULE_NAME}.property_from_dict", side_effect=[path_prop, query_prop])
+        path_schema = mocker.MagicMock()
+        query_schema = mocker.MagicMock()
+        import_string_from_reference = mocker.patch(
+            f"{MODULE_NAME}.import_string_from_reference", side_effect=["import_1", "import_2"]
+        )
+        data = {
+            "parameters": [
+                {"name": "path_prop_name", "required": True, "schema": path_schema, "in": "path"},
+                {"name": "query_prop_name", "required": False, "schema": query_schema, "in": "query"},
+            ]
+        }
+
+        endpoint._add_parameters(data)
+
+        propety_from_dict.assert_has_calls(
+            [
+                mocker.call(name="path_prop_name", required=True, data=path_schema),
+                mocker.call(name="query_prop_name", required=False, data=query_schema),
+            ]
+        )
+        import_string_from_reference.assert_has_calls(
+            [mocker.call(path_prop.reference, prefix="..models"), mocker.call(query_prop.reference, prefix="..models"),]
+        )
+        assert endpoint.relative_imports == {"import_1", "import_2", "import_3"}
+        assert endpoint.path_parameters == [path_prop]
+        assert endpoint.query_parameters == [query_prop]
+
+    def test_from_data(self, mocker):
+        from openapi_python_client.openapi_parser.openapi import Endpoint
+
+        path = mocker.MagicMock()
+        method = mocker.MagicMock()
+        _add_parameters = mocker.patch.object(Endpoint, "_add_parameters")
+        _add_responses = mocker.patch.object(Endpoint, "_add_responses")
+        _add_body = mocker.patch.object(Endpoint, "_add_body")
+        data = {
+            "description": mocker.MagicMock(),
+            "operationId": mocker.MagicMock(),
+            "security": {"blah": "bloo"},
+        }
+
+        endpoint = Endpoint.from_data(data=data, path=path, method=method)
+
+        assert endpoint.path == path
+        assert endpoint.method == method
+        assert endpoint.description == data["description"]
+        assert endpoint.name == data["operationId"]
+        assert endpoint.requires_security
+        assert endpoint.tag == "default"
+        _add_parameters.assert_called_once_with(data)
+        _add_responses.assert_called_once_with(data)
+        _add_body.assert_called_once_with(data)
+
+        data["tags"] = ["a", "b"]
+        del data["security"]
+
+        endpoint = Endpoint.from_data(data=data, path=path, method=method)
+
+        assert not endpoint.requires_security
+        assert endpoint.tag == "a"
+
 
 class TestImportStringFromReference:
     def test_import_string_from_reference_no_prefix(self, mocker):