Optimise tests to prefer non-Docker and be cross-platform

Author: SamuelMarks

scripts/run_petstore_test.py

@@ -3,6 +3,14 @@
 import shutil
 import subprocess
 import time
+import socket
+
+
+def find_free_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        return s.getsockname()[1]
 
 
 def main():
@@ -16,6 +24,7 @@ def main():
 
     json_file = sys.argv[1]
     tmp_dir = sys.argv[2]
+    tmp_server_dir = tmp_dir + "_server"
 
     input_path = os.path.join(test_harness_dir, json_file)
     if not os.path.exists(input_path):
@@ -38,40 +47,102 @@ def main():
         print(f"Running petstore SDK generation against {input_path} -> {tmp_dir}")
         subprocess.run(cmd, check=True)
 
-        # Test against Docker JVM Petstore server
-        container_name = f"petstore_server_{os.getpid()}"
-        image_name = (
-            "swaggerapi/petstore-v3"
-            if "oas3" in json_file.lower()
-            else "swaggerapi/petstore"
-        )
-
-        fallback_image_name = (
-            "openapitools/openapi-petstore"
-            if "oas3" in json_file.lower()
-            else "swaggerapi/swagger-petstore"  # Fallback guess
-        )
+        # Try to run Python-based generated server first
+        host_port = None
+        server_process = None
+        is_oas3 = "oas3" in json_file.lower()
 
         try:
-            print(f"Starting docker container {image_name}...")
+            print("Attempting to run Python-based mock server...")
             subprocess.run(
                 [
-                    "docker",
+                    "uv",
                     "run",
-                    "-d",
-                    "-P",
-                    "--name",
-                    container_name,
-                    image_name,
+                    "cdd-python",
+                    "from_openapi",
+                    "to_server",
+                    "-i",
+                    input_path,
+                    "-o",
+                    tmp_server_dir,
                 ],
                 check=True,
-                capture_output=True,
             )
+            host_port = str(find_free_port())
+
+            env = os.environ.copy()
+            env["PORT"] = host_port
+            env.pop("VIRTUAL_ENV", None)
+
+            subprocess.run(["uv", "venv"], cwd=tmp_server_dir, env=env, check=True)
+            subprocess.run(
+                ["uv", "pip", "install", "-e", "."],
+                cwd=tmp_server_dir,
+                env=env,
+                check=True,
+            )
+
+            server_process = subprocess.Popen(
+                [
+                    os.path.join(".venv", "bin", "python"),
+                    "main.py",
+                    "--ephemeral",
+                    "--seed",
+                ],
+                cwd=tmp_server_dir,
+                env=env,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+            )
+
+            import urllib.request
+
+            server_ready = False
+            for _ in range(30):
+                time.sleep(1)
+                try:
+                    urllib.request.urlopen(
+                        f"http://localhost:{host_port}/openapi.json", timeout=1
+                    )
+                    server_ready = True
+                    break
+                except Exception:
+                    pass
+
+            if not server_ready:
+                print("Python mock server failed to start. Output:")
+                server_process.terminate()
+                try:
+                    stdout, _ = server_process.communicate(timeout=5)
+                    print(
+                        f"--- SERVER OUTPUT ---\n{stdout.decode('utf-8', errors='ignore')}\n-------------------"
+                    )
+                except Exception as e:
+                    print(f"Failed to get server output: {e}")
+                server_process = None
+                host_port = None
+
         except Exception as e:
-            print(
-                f"Failed to start JVM image {image_name}, falling back to {fallback_image_name}: {e}"
+            print(f"Failed to start Python mock server: {e}")
+            if server_process:
+                server_process.terminate()
+                server_process = None
+            host_port = None
+
+        container_name = None
+        if not host_port:
+            print("Falling back to Docker JVM Petstore server...")
+            container_name = f"petstore_server_{os.getpid()}"
+            image_name = "swaggerapi/petstore-v3" if is_oas3 else "swaggerapi/petstore"
+
+            fallback_image_name = (
+                "openapitools/openapi-petstore"
+                if is_oas3
+                else "swaggerapi/swagger-petstore"  # Fallback guess
             )
+
             try:
+                print(f"Starting docker container {image_name}...")
                 subprocess.run(
                     [
                         "docker",
@@ -80,20 +151,55 @@ def main():
                         "-P",
                         "--name",
                         container_name,
-                        fallback_image_name,
+                        image_name,
                     ],
                     check=True,
                     capture_output=True,
                 )
-            except Exception as e2:
+            except Exception as e:
                 print(
-                    f"Fallback docker test failed (maybe docker not available?): {e2}"
+                    f"Failed to start JVM image {image_name}, falling back to {fallback_image_name}: {e}"
                 )
-                return
+                try:
+                    subprocess.run(
+                        [
+                            "docker",
+                            "run",
+                            "-d",
+                            "-P",
+                            "--name",
+                            container_name,
+                            fallback_image_name,
+                        ],
+                        check=True,
+                        capture_output=True,
+                    )
+                except Exception as e2:
+                    print(
+                        f"Fallback docker test failed (maybe docker not available?): {e2}"
+                    )
+                    return
+
+            try:
+                import urllib.request
+
+                for _ in range(10):
+                    time.sleep(3)
+                    try:
+                        port_res = subprocess.run(
+                            ["docker", "port", container_name, "8080"],
+                            check=True,
+                            capture_output=True,
+                            text=True,
+                        )
+                        host_port = port_res.stdout.strip().split(":")[-1]
+                        urllib.request.urlopen(
+                            f"http://localhost:{host_port}/api/swagger.json"
+                        )
+                        break
+                    except Exception:
+                        pass
 
-        try:
-            for _ in range(10):
-                time.sleep(3)
                 try:
                     port_res = subprocess.run(
                         ["docker", "port", container_name, "8080"],
@@ -102,35 +208,27 @@ def main():
                         text=True,
                     )
                     host_port = port_res.stdout.strip().split(":")[-1]
-                    import urllib.request
-
-                    urllib.request.urlopen(
-                        f"http://localhost:{host_port}/api/swagger.json"
-                    )
-                    break
-                except Exception:
-                    pass
-
-            try:
-                port_res = subprocess.run(
-                    ["docker", "port", container_name, "8080"],
-                    check=True,
-                    capture_output=True,
-                    text=True,
-                )
-                host_port = port_res.stdout.strip().split(":")[-1]
-            except subprocess.CalledProcessError as e:
-                print("Failed to get docker port:", e.stderr)
-                print("Failing gracefully.")
+                except subprocess.CalledProcessError as e:
+                    print("Failed to get docker port:", e.stderr)
+                    print("Failing gracefully.")
+                    return
+            except Exception as e:
+                print(f"Docker fallback setup failed: {e}")
                 return
 
+        try:
             # Use SDK to get inventory
             env = os.environ.copy()
-            env["API_BASE_URL"] = (
-                f"http://localhost:{host_port}/api/v3"
-                if "oas3" in json_file.lower()
-                else f"http://localhost:{host_port}/api"
-            )
+            if server_process:
+                # Python mock server has endpoints at root
+                env["API_BASE_URL"] = f"http://localhost:{host_port}"
+            else:
+                env["API_BASE_URL"] = (
+                    f"http://localhost:{host_port}/api/v3"
+                    if is_oas3
+                    else f"http://localhost:{host_port}/api"
+                )
+
             test_cmd = [
                 "uv",
                 "run",
@@ -143,16 +241,27 @@ def main():
                 _ = subprocess.run(
                     test_cmd, env=env, check=True, capture_output=True, text=True
                 )
+                print("SDK test passed.")
             except subprocess.CalledProcessError as e:
                 print("SDK test failed with error:", e.stderr)
                 print("Failing gracefully.")
-            pass
         finally:
-            subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
+            if server_process:
+                server_process.terminate()
+                try:
+                    server_process.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    server_process.kill()
+            if container_name:
+                subprocess.run(
+                    ["docker", "rm", "-f", container_name], capture_output=True
+                )
 
     finally:
         if os.path.exists(tmp_dir):
             shutil.rmtree(tmp_dir)
+        if os.path.exists(tmp_server_dir):
+            shutil.rmtree(tmp_server_dir)
 
 
 if __name__ == "__main__":  # pragma: no cover

tests/test_coverage_cli_branches.py

@@ -15,11 +15,11 @@ def test_serve_json_rpc_break(monkeypatch, capsys):
 
 
 def test_apply_env_vars_store_true():
+    """Test test_apply_env_vars_store_true."""
     from openapi_client.cli import apply_env_vars_to_parser
     import argparse
     import os
 
-    """Test test_apply_env_vars_store_true."""
     parser = argparse.ArgumentParser()
     parser.add_argument("--test-true", action="store_true")
     parser.add_argument("--test-false", action="store_false")

tests/test_coverage_gap.py

@@ -168,9 +168,9 @@ def test_mcp_read_resource_error():
 
 
 def test_mcp_client_notifications():
+    """Test building various notifications."""
     from openapi_client.mcp_client.client import MCPClient
 
-    """Test building various notifications."""
     client = MCPClient()
     assert client.build_cancelled_notification(1)["method"] == "notifications/cancelled"
     assert (
@@ -204,9 +204,9 @@ def test_mcp_client_notifications():
 
 
 def test_mcp_client_cursor():
+    """Test building various requests with cursor."""
     from openapi_client.mcp_client.client import MCPClient
 
-    """Test building various requests with cursor."""
     client = MCPClient()
     assert (
         client.build_list_resource_templates_request("cursor")["params"]["cursor"]
@@ -215,9 +215,9 @@ def test_mcp_client_cursor():
 
 
 def test_mcp_client_more_methods():
+    """Test test_mcp_client_more_methods."""
     from openapi_client.mcp_client.client import MCPClient
 
-    """Test test_mcp_client_more_methods."""
     client = MCPClient()
     assert client.build_ping_request()["method"] == "ping"
     assert client.build_initialize_request()["method"] == "initialize"
@@ -270,9 +270,8 @@ def test_mcp_all_methods():
 
 
 def test_mcp_client_more_missing():
-    from openapi_client.mcp_client.client import MCPClient
-
     """Test test_mcp_client_more_missing."""
+    from openapi_client.mcp_client.client import MCPClient
 
     client = MCPClient()
     assert (
@@ -281,9 +280,8 @@ def test_mcp_client_more_missing():
 
 
 def test_mcp_client_cursor_none_no_params():
-    from openapi_client.mcp_client.client import MCPClient
-
     """Test test_mcp_client_cursor_none_no_params."""
+    from openapi_client.mcp_client.client import MCPClient
 
     client = MCPClient()
 

tests/test_coverage_write_generated_local.py

@@ -1,3 +1,5 @@
+"""Tests for test_coverage_write_generated_local.py."""
+
 import unittest.mock
 from openapi_client.cli import sync_dir