OpenAPI Generator Testing — Automated Testing of Generated API Clients and Servers

import pytest
import requests
from openapi_client import ApiClient, Configuration
from openapi_client.api import threats_api
from openapi_client.model.threat import Threat

class TestThreatContract:
    @pytest.fixture
    def client(self):
        config = Configuration(
            host="http://localhost:8080",
            api_key={"ApiKeyAuth": "test-key"}
        )
        return threats_api.ThreatsApi(ApiClient(config))

    def test_create_threat(self, client):
        """Verify POST /threats creates and returns a valid Threat."""
        new_threat = Threat(
            name="Test Threat",
            severity="high",
            indicator_type="IP"
        )
        created = client.create_threat(new_threat)

        assert created.id is not None
        assert created.name == "Test Threat"
        assert created.severity == "high"
        assert created.created_at is not None

    def test_list_threats_pagination(self, client):
        """Verify GET /threats returns paginated results."""
        threats = client.list_threats(page=1, per_page=10)

        assert threats.data is not None
        assert len(threats.data) <= 10
        assert threats.total is not None
        assert threats._links.next is not None

    def test_get_nonexistent_threat(self, client):
        """Verify 404 for nonexistent resource."""
        with pytest.raises(requests.exceptions.HTTPError) as exc:
            client.get_threat("nonexistent-id")
        assert exc.value.response.status_code == 404

import json
import pytest
import requests
from pathlib import Path

class SpecDrivenTester:
    def __init__(self, spec_path, base_url):
        with open(spec_path) as f:
            self.spec = json.load(f)
        self.base_url = base_url

    def generate_tests(self):
        """Generate test cases from OpenAPI spec paths."""
        tests = []
        for path, methods in self.spec.get('paths', {}).items():
            for method, operation in methods.items():
                test = self._create_test(path, method, operation)
                if test:
                    tests.append(test)
        return tests

    def _create_test(self, path, method, operation):
        """Create a test case from a single operation."""
        url = f"{self.base_url}{path}"
        status_code = list(operation.get('responses', {}).keys())[0]

        def test_func():
            # Build request parameters from spec examples
            params = self._extract_params(operation)
            response = requests.request(method, url, **params)
            assert response.status_code == int(status_code), \
                f"Expected {status_code}, got {response.status_code}"

        test_func.__name__ = f"test_{operation.get('operationId', f'{method}_{path}')}"
        return test_func

    def _extract_params(self, operation):
        """Extract example parameters from the spec."""
        return {}  # Simplified for example

# Generate and run tests
tester = SpecDrivenTester('openapi.yaml', 'http://localhost:8080')
for test in tester.generate_tests():
    test()

import subprocess
import sys
import os

def run_generated_tests(spec_file, languages, server_port=8080):
    """
    CI pipeline: generate code, start server, run tests.
    """
    results = {'passed': [], 'failed': []}

    for lang in languages:
        print(f"\n=== Testing {lang} SDK ===")

        # Step 1: Generate client
        gen_cmd = [
            'openapi-generator-cli', 'generate',
            '-i', spec_file,
            '-g', lang,
            '-o', f'/tmp/sdks/{lang}',
            '--additional-properties=skipFormModel=true'
        ]
        subprocess.run(gen_cmd, check=True)

        # Step 2: Generate server stub
        server_dir = '/tmp/server'
        subprocess.run([
            'openapi-generator-cli', 'generate',
            '-i', spec_file,
            '-g', 'python-flask',
            '-o', server_dir
        ], check=True)

        # Step 3: Start server
        server_process = subprocess.Popen(
            ['python3', f'{server_dir}/app.py', '--port', str(server_port)]
        )

        try:
            # Step 4: Install and run client tests
            client_dir = f'/tmp/sdks/{lang}'
            subprocess.run(
                ['pip3', 'install', '-e', client_dir],
                capture_output=True
            )
            test_result = subprocess.run(
                ['pytest', f'{client_dir}/test/', '-v'],
                capture_output=True, text=True
            )

            if test_result.returncode == 0:
                results['passed'].append(lang)
                print(f"  PASSED: {lang} SDK tests")
            else:
                results['failed'].append(lang)
                print(f"  FAILED: {lang} SDK tests")
                print(test_result.stdout[-500:])

        finally:
            server_process.terminate()

    return results

# Usage
result = run_generated_tests(
    'openapi.yaml',
    languages=['python', 'typescript', 'java', 'go'],
    server_port=8080
)
print(f"Passed: {len(result['passed'])}, Failed: {len(result['failed'])}")