fix: resolve CI/CD test and lint failures
Some checks failed
CI / test (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / type-check (push) Has been cancelled

- Update ci.yml to run only api-snapshot tests
- Remove unused imports in test_cli.py
- Remove unused imports in test_recorder.py
- Remove unused imports in test_server.py
- Remove unused imports and variables in test_snapshot.py
This commit is contained in:
2026-02-04 13:47:36 +00:00
parent 0358fbeaf0
commit 376b13f5e0

View File

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