test: achieve 100% test coverage codebase wide

Author: SamuelMarks

.gitignore

@@ -151,3 +151,11 @@ celerybeat.pid
 # Pyenv / Misc
 .python-version
 *.snapshot.json
+
+# Profiling and Patching
+*.patch
+*.diff
+*.pyprof
+*.lprof
+*.pstats
+*.cprof

src/openapi_client/functions/emit.py

@@ -36,6 +36,8 @@ def emit_function(method: str, path: str, operation: Operation) -> cst.FunctionD
     has_query = False
     has_body_param = False
     body_param_name = "body"
+    has_form_data = False
+    form_data_params = []
 
     # Process query, header, path, cookie parameters
     if operation.parameters:
@@ -86,6 +88,10 @@ def emit_function(method: str, path: str, operation: Operation) -> cst.FunctionD
 
                 # Process query params
                 p_in = getattr(param, "in_", getattr(param, "in", None))
+                if p_in in ("formData", "form"):
+                    has_form_data = True
+                    form_data_params.append((param.name, param_name))
+
                 if p_in == "query":
                     style = getattr(param, "style", "form")
                     delim = (
@@ -203,11 +209,14 @@ def emit_function(method: str, path: str, operation: Operation) -> cst.FunctionD
         )
     body_statements.extend(query_statements)
 
-    url_value = cst.BinaryOperation(
-        left=cst.Attribute(value=cst.Name("self"), attr=cst.Name("base_url")),
-        operator=cst.Add(),
-        right=cst.SimpleString(f'"{path}"'),
-    )
+    if "{" in path and "}" in path:
+        url_value = cst.parse_expression(f'self.base_url + f"{path}"')
+    else:
+        url_value = cst.BinaryOperation(
+            left=cst.Attribute(value=cst.Name("self"), attr=cst.Name("base_url")),
+            operator=cst.Add(),
+            right=cst.SimpleString(f'"{path}"'),
+        )
     if has_query:
         url_value = cst.BinaryOperation(
             left=url_value,
@@ -281,7 +290,23 @@ def emit_function(method: str, path: str, operation: Operation) -> cst.FunctionD
                                     )
                                 ]
                                 if operation.requestBody or has_body_param
-                                else []
+                                else (
+                                    [
+                                        cst.Arg(
+                                            keyword=cst.Name("fields"),
+                                            value=cst.Dict(
+                                                elements=[
+                                                    cst.DictElement(
+                                                        key=cst.SimpleString(f'"{k}"'),
+                                                        value=cst.Name(v)
+                                                    )
+                                                    for k, v in form_data_params
+                                                ]
+                                            )
+                                        )
+                                    ]
+                                    if has_form_data else []
+                                )
                             ),
                         ),
                     )

src/openapi_client/tests/emit.py

@@ -31,13 +31,21 @@ def get_dummy_value_for_schema(schema_obj) -> cst.BaseExpression:
     elif ann.startswith("dict"):
         return cst.Dict(elements=[])
     else:
+        # Fallback to a string instead of None so urllib3 doesn't crash on encode_multipart_formdata
+        t = getattr(schema_obj, "type", None) if schema_obj else None
+        if t == "file":
+            return cst.SimpleString('b"dummy_content"')
         return cst.Name("None")
 
 
-def get_stub_body(schema_obj):
+def get_stub_body(schema_obj, spec=None):
     """
     Generate a stub body AST node based on an OpenAPI schema object.
     """
+    if schema_obj and getattr(schema_obj, "ref", None) and spec and spec.components and spec.components.schemas:
+        ref_name = schema_obj.ref.split('/')[-1]
+        schema_obj = spec.components.schemas.get(ref_name, schema_obj)
+
     if schema_obj and getattr(schema_obj, "properties", None):
         elements = []
         for prop_name, prop_schema in schema_obj.properties.items():
@@ -48,7 +56,10 @@ def get_stub_body(schema_obj):
                     elements=[cst.Element(cst.SimpleString('"http://dummy"'))]
                 )
             else:
-                val = get_dummy_value_for_schema(prop_schema)
+                if getattr(prop_schema, "ref", None) or getattr(prop_schema, "type", None) in ("object", "array"):
+                    val = get_stub_body(prop_schema, spec)
+                else:
+                    val = get_dummy_value_for_schema(prop_schema)
             elements.append(
                 cst.DictElement(
                     key=cst.SimpleString(f'"{prop_name}"'), value=val
@@ -60,7 +71,7 @@ def get_stub_body(schema_obj):
         if items_schema and getattr(items_schema, "type", None) in ("string", "integer", "number", "boolean"):
             item_val = get_dummy_value_for_schema(items_schema)
         else:
-            item_val = get_stub_body(items_schema)
+            item_val = get_stub_body(items_schema, spec)
         return cst.List(
             elements=[
                 cst.Element(item_val)
@@ -70,7 +81,7 @@ def get_stub_body(schema_obj):
         return cst.Dict(elements=[])
 
 def emit_operation_test(
-    method: str, path: str, operation: Operation, composable: bool = False
+    method: str, path: str, operation: Operation, composable: bool = False, spec=None
 ) -> cst.FunctionDef:
     """
     Emit a pytest unit test for an API operation.
@@ -96,11 +107,13 @@ def emit_operation_test(
                     schema_obj = getattr(
                         param, "schema_", getattr(param, "schema", None)
                     )
-                    val = get_stub_body(schema_obj)
+                    val = get_stub_body(schema_obj, spec)
                 else:
                     schema_obj = getattr(
                         param, "schema_", getattr(param, "schema", None)
                     )
+                    if not schema_obj and getattr(param, "type", None):
+                        schema_obj = param
                     val = get_dummy_value_for_schema(schema_obj)
                 args.append(cst.Arg(keyword=cst.Name(param_name), value=val))
 
@@ -129,7 +142,7 @@ def emit_operation_test(
                     getattr(content[first_key], "schema", None),
                 )
 
-            val = get_stub_body(schema_obj)
+            val = get_stub_body(schema_obj, spec)
         except Exception:
             val = cst.Dict(elements=[])
 
@@ -463,7 +476,7 @@ def emit_tests(spec: OpenAPI, composable: bool = False) -> cst.Module:
                     if operation:
                         body.append(
                             emit_operation_test(
-                                method, path, operation, composable=composable
+                                method, path, operation, composable=composable, spec=spec
                             )
                         )
                         body.append(cst.EmptyLine())

test/test_client.py

@@ -6,5 +6,5 @@
 
 @pytest.fixture
 def client():
-    return Client(os.getenv("API_URL", "http://localhost:8080/v2"))
+    return Client(os.getenv("API_URL", "http://localhost:8080/v2"), api_key = "special-key")
 

tests/test_compliance_updates.py

@@ -205,8 +205,8 @@ def test_emit_tests_exception():
 
     # Bypass pydantic validation to force an exception in the try block
     class Bad:
-        def __bool__(self):
-            return True
+        def __contains__(self, item):
+            raise Exception("Mock error")
 
     rb = RequestBody.model_construct(content=Bad())
     op = Operation(operationId="err", requestBody=rb)

tests/test_coverage_emit_functions_extra.py

@@ -0,0 +1,83 @@
+import pytest
+from openapi_client.models import Operation, Parameter, RequestBody, MediaType, Schema
+from openapi_client.functions.emit import emit_function
+
+def test_emit_function_formData_and_no_path_vars():
+    # Test path without "{}"
+    path = "/no/vars/here"
+    method = "post"
+    
+    # Test param with in="formData"
+    param = Parameter(
+        name="my_form_param",
+        in_="formData",
+        required=True,
+        schema=Schema(type="string")
+    )
+    
+    op = Operation(
+        operationId="test_formData_op",
+        parameters=[param]
+    )
+    
+    func_node = emit_function(method, path, op)
+    assert func_node is not None
+    assert func_node.name.value == "test_formData_op"
+
+from openapi_client.tests.emit import get_dummy_value_for_schema, get_stub_body, emit_operation_test
+from openapi_client.models import Schema, OpenAPI, Components, Reference
+
+def test_get_dummy_value_for_file():
+    schema = Schema(type="file")
+    val = get_dummy_value_for_schema(schema)
+    assert val.value == 'b"dummy_content"'
+
+def test_get_stub_body_ref():
+    ref_schema = Reference(ref="#/components/schemas/Target")
+    target_schema = Schema(type="object", properties={"some_prop": Schema(type="string")})
+    spec = OpenAPI(
+        openapi="3.2.0",
+        info={"title": "test", "version": "1.0"},
+        paths={},
+        components=Components(schemas={"Target": target_schema})
+    )
+    val = get_stub_body(ref_schema, spec)
+    assert val is not None
+
+def test_get_stub_body_prop_ref():
+    ref_schema = Reference(ref="#/components/schemas/Target")
+    target_schema = Schema(type="object", properties={"some_prop": Schema(type="string")})
+    main_schema = Schema(
+        type="object",
+        properties={
+            "my_ref": ref_schema,
+            "my_obj": Schema(type="object", properties={"inner": Schema(type="string")})
+        }
+    )
+    spec = OpenAPI(
+        openapi="3.2.0",
+        info={"title": "test", "version": "1.0"},
+        paths={},
+        components=Components(schemas={"Target": target_schema})
+    )
+    val = get_stub_body(main_schema, spec)
+    assert val is not None
+
+def test_emit_operation_test_param_no_schema():
+    param = Parameter(name="p", in_="query", type="string")
+    op = Operation(operationId="test_param_type", parameters=[param])
+    node = emit_operation_test("get", "/test", op)
+    assert node is not None
+
+def test_emit_function_with_path_vars():
+    path = "/users/{id}"
+    method = "get"
+    
+    op = Operation(
+        operationId="get_user",
+        parameters=[]
+    )
+    
+    func_node = emit_function(method, path, op)
+    assert func_node is not None
+    assert func_node.name.value == "get_user"

tests/test_coverage_emit_test.py

@@ -57,13 +57,13 @@ def test_emit_tests_exception_content():
     from openapi_client.tests.emit import emit_operation_test
     from openapi_client.models import Operation, RequestBody
 
-    class BadDict(dict):
-        def keys(self):
+    class BadDict:
+        def __contains__(self, item):
             raise Exception("Mock error")
 
-    op = Operation(
+    op = Operation.model_construct(
         operationId="test_exception",
-        requestBody=RequestBody(content=BadDict())
+        requestBody=RequestBody.model_construct(content=BadDict())
     )
     func = emit_operation_test("post", "/test", op)
     assert func.name.value == "test_test_exception"
@@ -73,17 +73,16 @@ def test_emit_tests_body_param_exception():
     from openapi_client.tests.emit import emit_operation_test
     from openapi_client.models import Operation, Parameter
 
-    class BadParameter(Parameter):
+    class BadParameter:
+        name = 'body_param'
+        in_ = 'body'
         @property
         def schema_(self):
             raise Exception("Mock error")
-        @property
-        def schema(self):
-            raise Exception("Mock error")
 
-    op = Operation(
+    op = Operation.model_construct(
         operationId="test_body_param_exception",
-        parameters=[BadParameter(name="body_param", in_="body")]
+        parameters=[BadParameter()]
     )
     try:
         func = emit_operation_test("post", "/test", op)
@@ -106,53 +105,36 @@ def test_emit_tests_request_body_exception():
     from openapi_client.tests.emit import emit_operation_test
     from openapi_client.models import Operation, RequestBody
 
-    class BadRequestBody(RequestBody):
+    class BadRequestBody:
+        pass
         @property
         def content(self):
             raise Exception("Mock error")
 
-    op = Operation(
+    op = Operation.model_construct(
         operationId="test_request_body_exception",
         requestBody=BadRequestBody()
     )
     func = emit_operation_test("post", "/test", op)
     assert func.name.value == "test_test_request_body_exception"
 
 
-def test_emit_tests_body_param_exception_schema():
-    from openapi_client.tests.emit import emit_operation_test
-    from openapi_client.models import Operation, Parameter
-
-    class BadParameter2(Parameter):
-        @property
-        def schema(self):
-            raise Exception("Mock error")
-
-    op = Operation(
-        operationId="test_body_param_exception_2",
-        parameters=[BadParameter2(name="body_param_2", in_="body")]
-    )
-    try:
-        func = emit_operation_test("post", "/test", op)
-    except Exception:
-        pass
 
 def test_emit_tests_other_param_exception():
     from openapi_client.tests.emit import emit_operation_test
     from openapi_client.models import Operation, Parameter
 
-    class BadParameter3(Parameter):
-        @property
-        def schema(self):
-            raise Exception("Mock error")
+    class BadParameter3:
+        name = 'other_param_3'
+        in_ = 'query'
 
         @property
         def schema_(self):
             raise Exception("Mock error")
 
-    op = Operation(
+    op = Operation.model_construct(
         operationId="test_other_param_exception",
-        parameters=[BadParameter3(name="other_param_3", in_="query")]
+        parameters=[BadParameter3()]
     )
     try:
         func = emit_operation_test("post", "/test", op)