Add mockapi source files and tests
This commit is contained in:
@@ -16,23 +16,12 @@ class MockOperationResolver(Resolver):
|
|||||||
"""Custom operation resolver for mock API."""
|
"""Custom operation resolver for mock API."""
|
||||||
|
|
||||||
def __init__(self, mock_server_generator):
|
def __init__(self, mock_server_generator):
|
||||||
"""Initialize the resolver.
|
"""Initialize the resolver."""
|
||||||
|
|
||||||
Args:
|
|
||||||
mock_server_generator: The MockServerGenerator instance
|
|
||||||
"""
|
|
||||||
super().__init__()
|
super().__init__()
|
||||||
self.mock_server_generator = mock_server_generator
|
self.mock_server_generator = mock_server_generator
|
||||||
|
|
||||||
def resolve(self, operation):
|
def resolve(self, operation):
|
||||||
"""Resolve an operation to a mock function.
|
"""Resolve an operation to a mock function."""
|
||||||
|
|
||||||
Args:
|
|
||||||
operation: The operation object from connexion
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Resolution object with the mock function
|
|
||||||
"""
|
|
||||||
operation_id = self.resolve_operation_id(operation)
|
operation_id = self.resolve_operation_id(operation)
|
||||||
for path, path_item in self.mock_server_generator.spec.get("paths", {}).items():
|
for path, path_item in self.mock_server_generator.spec.get("paths", {}).items():
|
||||||
for method, op_def in path_item.items():
|
for method, op_def in path_item.items():
|
||||||
@@ -47,12 +36,7 @@ class MockServerGenerator:
|
|||||||
"""Generates a mock server from an OpenAPI specification."""
|
"""Generates a mock server from an OpenAPI specification."""
|
||||||
|
|
||||||
def __init__(self, spec: Dict[str, Any], config: Optional[Config] = None):
|
def __init__(self, spec: Dict[str, Any], config: Optional[Config] = None):
|
||||||
"""Initialize the mock server generator.
|
"""Initialize the mock server generator."""
|
||||||
|
|
||||||
Args:
|
|
||||||
spec: The OpenAPI specification dictionary
|
|
||||||
config: Configuration object
|
|
||||||
"""
|
|
||||||
self.spec = spec
|
self.spec = spec
|
||||||
self.config = config or Config()
|
self.config = config or Config()
|
||||||
schemas = spec.get("components", {}).get("schemas", {})
|
schemas = spec.get("components", {}).get("schemas", {})
|
||||||
@@ -60,11 +44,7 @@ class MockServerGenerator:
|
|||||||
self.app: Optional[App] = None
|
self.app: Optional[App] = None
|
||||||
|
|
||||||
def generate(self) -> App:
|
def generate(self) -> App:
|
||||||
"""Generate the connexion application.
|
"""Generate the connexion application."""
|
||||||
|
|
||||||
Returns:
|
|
||||||
Configured connexion App instance
|
|
||||||
"""
|
|
||||||
self.app = connexion.App(__name__, specification_dir=".")
|
self.app = connexion.App(__name__, specification_dir=".")
|
||||||
|
|
||||||
self.app.add_api(
|
self.app.add_api(
|
||||||
@@ -84,14 +64,7 @@ class MockServerGenerator:
|
|||||||
return self.app
|
return self.app
|
||||||
|
|
||||||
def _create_mock_function(self, operation: Dict[str, Any]):
|
def _create_mock_function(self, operation: Dict[str, Any]):
|
||||||
"""Create a mock function for an operation.
|
"""Create a mock function for an operation."""
|
||||||
|
|
||||||
Args:
|
|
||||||
operation: The OpenAPI operation definition
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Function that generates mock responses
|
|
||||||
"""
|
|
||||||
|
|
||||||
def mock_function(*args, **kwargs):
|
def mock_function(*args, **kwargs):
|
||||||
mock_config = operation.get("x-mock-config", {})
|
mock_config = operation.get("x-mock-config", {})
|
||||||
@@ -113,26 +86,12 @@ class MockServerGenerator:
|
|||||||
return mock_function
|
return mock_function
|
||||||
|
|
||||||
def _should_return_error(self, mock_config: Dict[str, Any]) -> bool:
|
def _should_return_error(self, mock_config: Dict[str, Any]) -> bool:
|
||||||
"""Determine if we should return an error response.
|
"""Determine if we should return an error response."""
|
||||||
|
|
||||||
Args:
|
|
||||||
mock_config: x-mock-config extension data
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
True if error should be returned
|
|
||||||
"""
|
|
||||||
error_probability = mock_config.get("errorProbability", 0)
|
error_probability = mock_config.get("errorProbability", 0)
|
||||||
return random.random() < error_probability
|
return random.random() < error_probability
|
||||||
|
|
||||||
def _generate_error_response(self, mock_config: Dict[str, Any]) -> tuple:
|
def _generate_error_response(self, mock_config: Dict[str, Any]) -> tuple:
|
||||||
"""Generate an error response.
|
"""Generate an error response."""
|
||||||
|
|
||||||
Args:
|
|
||||||
mock_config: x-mock-config extension data
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Tuple of (response_body, status_code)
|
|
||||||
"""
|
|
||||||
status_code = mock_config.get("errorCode", 500)
|
status_code = mock_config.get("errorCode", 500)
|
||||||
error_message = mock_config.get("errorMessage", "Mock error")
|
error_message = mock_config.get("errorMessage", "Mock error")
|
||||||
return {"error": error_message}, status_code
|
return {"error": error_message}, status_code
|
||||||
@@ -142,14 +101,6 @@ def create_mock_server(
|
|||||||
spec: Dict[str, Any],
|
spec: Dict[str, Any],
|
||||||
config: Optional[Config] = None,
|
config: Optional[Config] = None,
|
||||||
) -> App:
|
) -> App:
|
||||||
"""Create a mock server from an OpenAPI spec.
|
"""Create a mock server from an OpenAPI spec."""
|
||||||
|
|
||||||
Args:
|
|
||||||
spec: OpenAPI specification dictionary
|
|
||||||
config: Configuration object
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Configured connexion App
|
|
||||||
"""
|
|
||||||
generator = MockServerGenerator(spec, config)
|
generator = MockServerGenerator(spec, config)
|
||||||
return generator.generate()
|
return generator.generate()
|
||||||
Reference in New Issue
Block a user