diff --git a/tests/test_server.py b/tests/test_server.py index 24db564..4b5c1d7 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -2,226 +2,229 @@ import json import os -import pytest -from unittest.mock import MagicMock, patch -from urllib.parse import urlparse -from api_snapshot.server.server import ( - MockServer, - parse_path_parameters, - create_app_from_snapshot -) -from api_snapshot.snapshot.manager import ( - Snapshot, - SnapshotMetadata, - SnapshotManager -) +import pytest + from api_snapshot.recorder.recorder import ( RecordedRequest, RecordedResponse, - RequestResponsePair + 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" + "/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 + body=None, ) get_users_resp = RecordedResponse( status_code=200, headers={"Content-Type": "application/json"}, body='[{"id": 1, "name": "John"}]', - latency_ms=100 + latency_ms=100, ) - pairs.append(RequestResponsePair(request=get_users, response=get_users_resp)) - + 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"}' + body='{"name": "Jane"}', ) post_users_resp = RecordedResponse( status_code=201, headers={"Content-Type": "application/json"}, body='{"id": 2, "name": "Jane"}', - latency_ms=150 + latency_ms=150, ) - pairs.append(RequestResponsePair(request=post_users, response=post_users_resp)) - + 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" + 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 + 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") - + + 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 + body=None, ) resp1 = RecordedResponse( status_code=200, headers={"Content-Type": "application/json"}, body='{"id": ":id"}', - latency_ms=50 + 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.""" @@ -229,68 +232,68 @@ class TestMockServerRoutes: method="GET", url="/hello", headers={}, - body=None + body=None, ) resp = RecordedResponse( status_code=200, headers={"Content-Type": "text/plain"}, body="Hello, World!", - latency_ms=10 + 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 @@ -298,7 +301,7 @@ class TestMockServerRoutes: class TestMockServerLatency: """Tests for mock server latency simulation.""" - + @pytest.fixture def latency_snapshot(self): """Create snapshot for latency tests.""" @@ -306,69 +309,69 @@ class TestMockServerLatency: method="GET", url="/test", headers={}, - body=None + body=None, ) resp = RecordedResponse( status_code=200, headers={}, body="test", - latency_ms=100 + 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 + 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) + 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 + body=None, ) resp = RecordedResponse( status_code=200, headers={"Content-Type": "application/json"}, body='{"success": true}', - latency_ms=50 + 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") + 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"