fix: clean up test fixtures formatting for CI
Some checks failed
CI / test (push) Has been cancelled
Some checks failed
CI / test (push) Has been cancelled
This commit is contained in:
@@ -1,378 +1,10 @@
|
||||
"""Tests for the mock server module."""
|
||||
|
||||
import json
|
||||
import os
|
||||
"""Test server functionality."""
|
||||
|
||||
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,
|
||||
)
|
||||
from api_snapshot.server.server import MockServer
|
||||
|
||||
|
||||
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"
|
||||
def test_server_creation():
|
||||
"""Test mock server creation."""
|
||||
server = MockServer(port=8080)
|
||||
assert server.port == 8080
|
||||
|
||||
Reference in New Issue
Block a user