379 lines
11 KiB
Python
379 lines
11 KiB
Python
"""Tests for the mock server module."""
|
|
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from api_snapshot.recorder.recorder import (
|
|
RecordedRequest,
|
|
RecordedResponse,
|
|
RequestResponsePair,
|
|
)
|
|
from api_snapshot.server.server import (
|
|
MockServer,
|
|
create_app_from_snapshot,
|
|
parse_path_parameters,
|
|
)
|
|
from api_snapshot.snapshot.manager import (
|
|
Snapshot,
|
|
SnapshotManager,
|
|
SnapshotMetadata,
|
|
)
|
|
|
|
|
|
class TestParsePathParameters:
|
|
"""Tests for parse_path_parameters function."""
|
|
|
|
def test_exact_path_match(self):
|
|
"""Test with exact path match (no parameters)."""
|
|
params = parse_path_parameters("/users", "/users")
|
|
|
|
assert params == {}
|
|
|
|
def test_single_parameter(self):
|
|
"""Test extracting single path parameter."""
|
|
params = parse_path_parameters("/users/123", "/users/:id")
|
|
|
|
assert params == {"id": "123"}
|
|
|
|
def test_multiple_parameters(self):
|
|
"""Test extracting multiple path parameters."""
|
|
params = parse_path_parameters(
|
|
"/users/123/posts/456",
|
|
"/users/:userId/posts/:postId",
|
|
)
|
|
|
|
assert params == {"userId": "123", "postId": "456"}
|
|
|
|
def test_brace_parameters(self):
|
|
"""Test extracting parameters with braces format."""
|
|
params = parse_path_parameters("/items/abc", "/items/{id}")
|
|
|
|
assert params == {"id": "abc"}
|
|
|
|
|
|
class TestMockServer:
|
|
"""Tests for MockServer class."""
|
|
|
|
@pytest.fixture
|
|
def sample_pairs(self):
|
|
"""Create sample request-response pairs."""
|
|
pairs = []
|
|
|
|
get_users = RecordedRequest(
|
|
method="GET",
|
|
url="/users",
|
|
headers={"Accept": "application/json"},
|
|
body=None,
|
|
)
|
|
get_users_resp = RecordedResponse(
|
|
status_code=200,
|
|
headers={"Content-Type": "application/json"},
|
|
body='[{"id": 1, "name": "John"}]',
|
|
latency_ms=100,
|
|
)
|
|
pairs.append(RequestResponsePair(request=get_users,
|
|
response=get_users_resp))
|
|
|
|
post_users = RecordedRequest(
|
|
method="POST",
|
|
url="/users",
|
|
headers={"Content-Type": "application/json"},
|
|
body='{"name": "Jane"}',
|
|
)
|
|
post_users_resp = RecordedResponse(
|
|
status_code=201,
|
|
headers={"Content-Type": "application/json"},
|
|
body='{"id": 2, "name": "Jane"}',
|
|
latency_ms=150,
|
|
)
|
|
pairs.append(RequestResponsePair(request=post_users,
|
|
response=post_users_resp))
|
|
|
|
return pairs
|
|
|
|
def test_init_server(self, sample_pairs):
|
|
"""Test initializing mock server."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(
|
|
snapshot=snapshot,
|
|
host="127.0.0.1",
|
|
port=8080,
|
|
latency_mode="original",
|
|
)
|
|
|
|
assert server.snapshot == snapshot
|
|
assert server.host == "127.0.0.1"
|
|
assert server.port == 8080
|
|
assert server.latency_mode == "original"
|
|
|
|
def test_get_latency_original(self, sample_pairs):
|
|
"""Test latency calculation in original mode."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(snapshot=snapshot, latency_mode="original")
|
|
|
|
latency = server._get_latency_ms(100)
|
|
|
|
assert latency == 100
|
|
|
|
def test_get_latency_fixed(self, sample_pairs):
|
|
"""Test latency calculation in fixed mode."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(
|
|
snapshot=snapshot,
|
|
latency_mode="fixed",
|
|
fixed_latency_ms=200,
|
|
)
|
|
|
|
latency = server._get_latency_ms(50)
|
|
|
|
assert latency == 200
|
|
|
|
def test_get_latency_none(self, sample_pairs):
|
|
"""Test latency calculation with no latency."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(snapshot=snapshot, latency_mode="none")
|
|
|
|
latency = server._get_latency_ms(500)
|
|
|
|
assert latency == 0
|
|
|
|
def test_parse_query_params(self, sample_pairs):
|
|
"""Test parsing query parameters from URL."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(snapshot=snapshot)
|
|
|
|
params = server._parse_query_params(
|
|
"https://api.example.com/users?page=1&limit=10"
|
|
)
|
|
|
|
assert params["page"] == ["1"]
|
|
assert params["limit"] == ["10"]
|
|
|
|
def test_match_request_exact(self, sample_pairs):
|
|
"""Test matching exact request."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(snapshot=snapshot)
|
|
|
|
pair = server._match_request("GET", "/users", {})
|
|
|
|
assert pair is not None
|
|
assert pair.request.method == "GET"
|
|
|
|
def test_match_request_method_mismatch(self, sample_pairs):
|
|
"""Test matching with method mismatch."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(snapshot=snapshot)
|
|
|
|
pair = server._match_request("DELETE", "/users", {})
|
|
|
|
assert pair is None
|
|
|
|
def test_match_request_not_found(self, sample_pairs):
|
|
"""Test matching when route not found."""
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=sample_pairs)
|
|
|
|
server = MockServer(snapshot=snapshot)
|
|
|
|
pair = server._match_request("GET", "/nonexistent", {})
|
|
|
|
assert pair is None
|
|
|
|
def test_match_request_with_path_params(self, sample_pairs):
|
|
"""Test matching request with path parameters."""
|
|
pair1 = RecordedRequest(
|
|
method="GET",
|
|
url="/users/:id",
|
|
headers={},
|
|
body=None,
|
|
)
|
|
resp1 = RecordedResponse(
|
|
status_code=200,
|
|
headers={"Content-Type": "application/json"},
|
|
body='{"id": ":id"}',
|
|
latency_ms=50,
|
|
)
|
|
pair = RequestResponsePair(request=pair1, response=resp1)
|
|
|
|
metadata = SnapshotMetadata()
|
|
snapshot = Snapshot(metadata=metadata, requests=[pair])
|
|
|
|
server = MockServer(snapshot=snapshot)
|
|
|
|
matched = server._match_request("GET", "/users/123", {})
|
|
|
|
assert matched is not None
|
|
assert matched.response.status_code == 200
|
|
|
|
|
|
class TestMockServerRoutes:
|
|
"""Tests for mock server route handling."""
|
|
|
|
@pytest.fixture
|
|
def simple_snapshot(self):
|
|
"""Create a simple snapshot for route tests."""
|
|
req = RecordedRequest(
|
|
method="GET",
|
|
url="/hello",
|
|
headers={},
|
|
body=None,
|
|
)
|
|
resp = RecordedResponse(
|
|
status_code=200,
|
|
headers={"Content-Type": "text/plain"},
|
|
body="Hello, World!",
|
|
latency_ms=10,
|
|
)
|
|
pair = RequestResponsePair(request=req, response=resp)
|
|
|
|
metadata = SnapshotMetadata()
|
|
return Snapshot(metadata=metadata, requests=[pair])
|
|
|
|
def test_route_returns_correct_status(self, simple_snapshot):
|
|
"""Test that route returns correct status code."""
|
|
server = MockServer(snapshot=simple_snapshot)
|
|
|
|
with server.app.test_client() as client:
|
|
response = client.get("/hello")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_route_returns_correct_body(self, simple_snapshot):
|
|
"""Test that route returns correct body."""
|
|
server = MockServer(snapshot=simple_snapshot)
|
|
|
|
with server.app.test_client() as client:
|
|
response = client.get("/hello")
|
|
|
|
assert response.data.decode() == "Hello, World!"
|
|
|
|
def test_route_returns_correct_headers(self, simple_snapshot):
|
|
"""Test that route returns correct headers."""
|
|
server = MockServer(snapshot=simple_snapshot)
|
|
|
|
with server.app.test_client() as client:
|
|
response = client.get("/hello")
|
|
|
|
assert "text/plain" in response.content_type
|
|
|
|
def test_route_not_found(self, simple_snapshot):
|
|
"""Test 404 for unmatched routes."""
|
|
server = MockServer(snapshot=simple_snapshot)
|
|
|
|
with server.app.test_client() as client:
|
|
response = client.get("/nonexistent")
|
|
|
|
assert response.status_code == 404
|
|
|
|
data = json.loads(response.data)
|
|
assert "error" in data
|
|
assert "available_routes" in data
|
|
|
|
def test_snapshot_info_endpoint(self, simple_snapshot):
|
|
"""Test the __snapshot-info endpoint."""
|
|
server = MockServer(snapshot=simple_snapshot)
|
|
|
|
with server.app.test_client() as client:
|
|
response = client.get("/__snapshot-info")
|
|
|
|
assert response.status_code == 200
|
|
|
|
data = json.loads(response.data)
|
|
assert "endpoints" in data
|
|
assert "latency_mode" in data
|
|
|
|
|
|
class TestMockServerLatency:
|
|
"""Tests for mock server latency simulation."""
|
|
|
|
@pytest.fixture
|
|
def latency_snapshot(self):
|
|
"""Create snapshot for latency tests."""
|
|
req = RecordedRequest(
|
|
method="GET",
|
|
url="/test",
|
|
headers={},
|
|
body=None,
|
|
)
|
|
resp = RecordedResponse(
|
|
status_code=200,
|
|
headers={},
|
|
body="test",
|
|
latency_ms=100,
|
|
)
|
|
pair = RequestResponsePair(request=req, response=resp)
|
|
|
|
metadata = SnapshotMetadata()
|
|
return Snapshot(metadata=metadata, requests=[pair])
|
|
|
|
def test_latency_mode_fixed(self, latency_snapshot):
|
|
"""Test fixed latency mode."""
|
|
server = MockServer(
|
|
snapshot=latency_snapshot,
|
|
latency_mode="fixed",
|
|
fixed_latency_ms=50,
|
|
)
|
|
|
|
assert server.latency_mode == "fixed"
|
|
assert server.fixed_latency_ms == 50
|
|
|
|
def test_latency_mode_random(self, latency_snapshot):
|
|
"""Test random latency mode."""
|
|
server = MockServer(
|
|
snapshot=latency_snapshot,
|
|
latency_mode="random",
|
|
random_latency_range=(10, 100),
|
|
)
|
|
|
|
assert server.latency_mode == "random"
|
|
assert server.random_latency_range == (10, 100)
|
|
|
|
|
|
class TestCreateAppFromSnapshot:
|
|
"""Tests for create_app_from_snapshot function."""
|
|
|
|
def test_create_app_from_snapshot(self, temp_dir):
|
|
"""Test creating Flask app from snapshot."""
|
|
manager = SnapshotManager(temp_dir)
|
|
|
|
req = RecordedRequest(
|
|
method="GET",
|
|
url="/api/test",
|
|
headers={},
|
|
body=None,
|
|
)
|
|
resp = RecordedResponse(
|
|
status_code=200,
|
|
headers={"Content-Type": "application/json"},
|
|
body='{"success": true}',
|
|
latency_ms=50,
|
|
)
|
|
pair = RequestResponsePair(request=req, response=resp)
|
|
|
|
manager.save_snapshot("test", requests=[pair])
|
|
|
|
app, server = create_app_from_snapshot(
|
|
snapshot_path=os.path.join(temp_dir, "test.json"),
|
|
)
|
|
|
|
assert app is not None
|
|
assert server is not None
|
|
assert server.host == "127.0.0.1"
|
|
assert server.port == 8080
|