245 lines
7.6 KiB
Python
245 lines
7.6 KiB
Python
"""Tests for Mock Server."""
|
|
|
|
import pytest
|
|
import json
|
|
from src.core.server import MockServer, MockServerError, RouteNotFoundError
|
|
|
|
|
|
class TestMockServer:
|
|
"""Tests for MockServer class."""
|
|
|
|
@pytest.fixture
|
|
def temp_spec_file(self, sample_openapi_yaml, tmp_path):
|
|
"""Create a temporary OpenAPI spec file."""
|
|
from pathlib import Path
|
|
spec_file = tmp_path / "openapi.yaml"
|
|
spec_file.write_text(sample_openapi_yaml)
|
|
return str(spec_file)
|
|
|
|
def test_server_initialization(self, temp_spec_file):
|
|
"""Test server initializes with correct parameters."""
|
|
server = MockServer(
|
|
spec_file=temp_spec_file,
|
|
port=8080,
|
|
delay=100,
|
|
fuzzing=False,
|
|
seed=42
|
|
)
|
|
|
|
assert server.port == 8080
|
|
assert server.delay == 100
|
|
assert server.fuzzing is False
|
|
assert server.seed == 42
|
|
|
|
def test_server_loads_spec(self, temp_spec_file):
|
|
"""Test server loads and validates OpenAPI spec."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
spec = server.load_spec()
|
|
|
|
assert spec is not None
|
|
assert spec["openapi"] == "3.0.0"
|
|
|
|
def test_route_cache_creation(self, temp_spec_file):
|
|
"""Test that route cache is created on setup."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
assert len(server._route_cache) > 0
|
|
assert "/users" in server._route_cache
|
|
assert "/users/{userId}" in server._route_cache
|
|
assert "/posts" in server._route_cache
|
|
|
|
def test_find_route_success(self, temp_spec_file):
|
|
"""Test finding a valid route."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
path_item, path_params = server._find_route("/users", "GET")
|
|
|
|
assert path_item is not None
|
|
assert "get" in path_item
|
|
|
|
def test_find_route_with_path_parameter(self, temp_spec_file):
|
|
"""Test finding a route with path parameter."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
path_item, path_params = server._find_route("/users/123", "GET")
|
|
|
|
assert path_item is not None
|
|
assert path_params["userId"] == "123"
|
|
|
|
def test_find_route_not_found(self, temp_spec_file):
|
|
"""Test that RouteNotFoundError is raised for unknown route."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
with pytest.raises(RouteNotFoundError):
|
|
server._find_route("/unknown", "GET")
|
|
|
|
def test_get_response_schema(self, temp_spec_file):
|
|
"""Test getting response schema for an operation."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
paths = server.parser.get_paths()
|
|
path_item = paths["/users"]
|
|
|
|
schema = server._get_response_schema(path_item, "get", "200")
|
|
|
|
assert schema is not None
|
|
|
|
def test_handle_request_success(self, temp_spec_file):
|
|
"""Test handling a valid request."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
response = server._handle_request(
|
|
method="GET",
|
|
path="/users",
|
|
headers={},
|
|
query_params={},
|
|
body=None
|
|
)
|
|
|
|
assert response["status_code"] == 200
|
|
assert "body" in response
|
|
|
|
def test_handle_request_with_path_param(self, temp_spec_file):
|
|
"""Test handling request with path parameter."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
response = server._handle_request(
|
|
method="GET",
|
|
path="/users/123",
|
|
headers={},
|
|
query_params={},
|
|
body=None
|
|
)
|
|
|
|
assert response["status_code"] == 200
|
|
|
|
def test_handle_request_not_found(self, temp_spec_file):
|
|
"""Test handling request for unknown route."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
response = server._handle_request(
|
|
method="GET",
|
|
path="/unknown/route",
|
|
headers={},
|
|
query_params={},
|
|
body=None
|
|
)
|
|
|
|
assert response["status_code"] == 404
|
|
assert "error" in response["body"]
|
|
|
|
def test_handle_request_with_fuzzing(self, temp_spec_file):
|
|
"""Test handling request in fuzzing mode."""
|
|
server = MockServer(spec_file=temp_spec_file, fuzzing=True)
|
|
server.load_spec()
|
|
|
|
response = server._handle_request(
|
|
method="GET",
|
|
path="/users",
|
|
headers={},
|
|
query_params={},
|
|
body=None
|
|
)
|
|
|
|
assert response["status_code"] == 200
|
|
assert "body" in response
|
|
|
|
|
|
class TestMockServerDelay:
|
|
"""Tests for MockServer delay functionality."""
|
|
|
|
@pytest.fixture
|
|
def temp_spec_file(self, sample_openapi_yaml, tmp_path):
|
|
"""Create a temporary OpenAPI spec file."""
|
|
from pathlib import Path
|
|
spec_file = tmp_path / "openapi.yaml"
|
|
spec_file.write_text(sample_openapi_yaml)
|
|
return str(spec_file)
|
|
|
|
def test_get_delay_zero(self, temp_spec_file):
|
|
"""Test that zero delay returns no delay."""
|
|
server = MockServer(spec_file=temp_spec_file, delay=0)
|
|
server.load_spec()
|
|
|
|
assert server._get_delay() == 0.0
|
|
|
|
def test_get_delay_positive(self, temp_spec_file):
|
|
"""Test that positive delay returns correct value."""
|
|
server = MockServer(spec_file=temp_spec_file, delay=100)
|
|
server.load_spec()
|
|
|
|
delay = server._get_delay()
|
|
|
|
assert delay == 0.1
|
|
|
|
|
|
class TestMockServerRequestBody:
|
|
"""Tests for MockServer request body parsing."""
|
|
|
|
@pytest.fixture
|
|
def temp_spec_file(self, sample_openapi_yaml, tmp_path):
|
|
"""Create a temporary OpenAPI spec file."""
|
|
from pathlib import Path
|
|
spec_file = tmp_path / "openapi.yaml"
|
|
spec_file.write_text(sample_openapi_yaml)
|
|
return str(spec_file)
|
|
|
|
def test_parse_request_body_json(self, temp_spec_file):
|
|
"""Test parsing JSON request body."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
body = b'{"name": "test"}'
|
|
result = server._parse_request_body("application/json", body)
|
|
|
|
assert result == {"name": "test"}
|
|
|
|
def test_parse_request_body_empty(self, temp_spec_file):
|
|
"""Test parsing empty request body."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
result = server._parse_request_body("application/json", b"")
|
|
|
|
assert result is None
|
|
|
|
def test_parse_request_body_no_content_type(self, temp_spec_file):
|
|
"""Test parsing request body without content type."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
result = server._parse_request_body(None, b"data")
|
|
|
|
assert result is None
|
|
|
|
def test_parse_request_body_invalid_json(self, temp_spec_file):
|
|
"""Test parsing invalid JSON request body."""
|
|
server = MockServer(spec_file=temp_spec_file)
|
|
server.load_spec()
|
|
|
|
result = server._parse_request_body("application/json", b"invalid json")
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestMockServerErrorHandling:
|
|
"""Tests for MockServer error handling."""
|
|
|
|
def test_port_in_use_error(self):
|
|
"""Test PortInUseError exception."""
|
|
error = MockServerError("Test error")
|
|
assert str(error) == "Test error"
|
|
|
|
def test_route_not_found_error(self):
|
|
"""Test RouteNotFoundError exception."""
|
|
error = RouteNotFoundError("Route not found")
|
|
assert str(error) == "Route not found"
|