From fec05923708249a590f9e6e76f199e1f510061e1 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Wed, 4 Feb 2026 13:39:52 +0000 Subject: [PATCH] Add server and snapshot tests --- tests/test_server.py | 375 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 375 insertions(+) create mode 100644 tests/test_server.py diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..24db564 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,375 @@ +"""Tests for the mock server module.""" + +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 +) +from api_snapshot.recorder.recorder import ( + RecordedRequest, + RecordedResponse, + RequestResponsePair +) + + +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