7000pctAUTO
0358fbeaf0
- 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
316 lines
10 KiB
Python
316 lines
10 KiB
Python
"""Tests for the recorder module."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from api_snapshot.recorder.recorder import (
|
|
RecordedRequest,
|
|
RecordedResponse,
|
|
RecordingSession,
|
|
RequestResponsePair,
|
|
record_multiple,
|
|
record_session,
|
|
)
|
|
|
|
|
|
class TestRecordedRequest:
|
|
"""Tests for RecordedRequest class."""
|
|
|
|
def test_create_request(self):
|
|
"""Test creating a recorded request."""
|
|
req = RecordedRequest(
|
|
method="GET",
|
|
url="https://api.example.com/users",
|
|
headers={"Accept": "application/json"},
|
|
body=None,
|
|
)
|
|
|
|
assert req.method == "GET"
|
|
assert req.url == "https://api.example.com/users"
|
|
assert req.headers == {"Accept": "application/json"}
|
|
assert req.body is None
|
|
assert req.timestamp is not None
|
|
|
|
def test_request_to_dict(self):
|
|
"""Test converting request to dictionary."""
|
|
req = RecordedRequest(
|
|
method="POST",
|
|
url="https://api.example.com/data",
|
|
headers={"Content-Type": "application/json"},
|
|
body='{"key": "value"}',
|
|
)
|
|
|
|
data = req.to_dict()
|
|
|
|
assert data["method"] == "POST"
|
|
assert data["url"] == "https://api.example.com/data"
|
|
assert data["headers"] == {"Content-Type": "application/json"}
|
|
assert data["body"] == '{"key": "value"}'
|
|
assert "timestamp" in data
|
|
|
|
def test_request_from_dict(self):
|
|
"""Test creating request from dictionary."""
|
|
data = {
|
|
"method": "PUT",
|
|
"url": "https://api.example.com/resource/1",
|
|
"headers": {"Authorization": "Bearer token"},
|
|
"body": "updated data",
|
|
"timestamp": "2024-01-01T12:00:00",
|
|
}
|
|
|
|
req = RecordedRequest.from_dict(data)
|
|
|
|
assert req.method == "PUT"
|
|
assert req.url == "https://api.example.com/resource/1"
|
|
assert req.headers == {"Authorization": "Bearer token"}
|
|
assert req.body == "updated data"
|
|
assert req.timestamp == "2024-01-01T12:00:00"
|
|
|
|
|
|
class TestRecordedResponse:
|
|
"""Tests for RecordedResponse class."""
|
|
|
|
def test_create_response(self):
|
|
"""Test creating a recorded response."""
|
|
resp = RecordedResponse(
|
|
status_code=200,
|
|
headers={"Content-Type": "application/json"},
|
|
body='{"success": true}',
|
|
latency_ms=150,
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.headers == {"Content-Type": "application/json"}
|
|
assert resp.body == '{"success": true}'
|
|
assert resp.latency_ms == 150
|
|
|
|
def test_response_to_dict(self):
|
|
"""Test converting response to dictionary."""
|
|
resp = RecordedResponse(
|
|
status_code=404,
|
|
headers={"Content-Type": "text/plain"},
|
|
body="Not Found",
|
|
latency_ms=50,
|
|
)
|
|
|
|
data = resp.to_dict()
|
|
|
|
assert data["status_code"] == 404
|
|
assert data["headers"] == {"Content-Type": "text/plain"}
|
|
assert data["body"] == "Not Found"
|
|
assert data["latency_ms"] == 50
|
|
|
|
def test_response_from_dict(self):
|
|
"""Test creating response from dictionary."""
|
|
data = {
|
|
"status_code": 500,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": '{"error": "internal"}',
|
|
"latency_ms": 300,
|
|
}
|
|
|
|
resp = RecordedResponse.from_dict(data)
|
|
|
|
assert resp.status_code == 500
|
|
assert resp.headers == {"Content-Type": "application/json"}
|
|
assert resp.body == '{"error": "internal"}'
|
|
assert resp.latency_ms == 300
|
|
|
|
|
|
class TestRequestResponsePair:
|
|
"""Tests for RequestResponsePair class."""
|
|
|
|
def test_create_pair(self, sample_request, sample_response):
|
|
"""Test creating a request-response pair."""
|
|
pair = RequestResponsePair(
|
|
request=sample_request,
|
|
response=sample_response,
|
|
)
|
|
|
|
assert pair.request == sample_request
|
|
assert pair.response == sample_response
|
|
|
|
def test_pair_to_dict(self, sample_pair):
|
|
"""Test converting pair to dictionary."""
|
|
data = sample_pair.to_dict()
|
|
|
|
assert "request" in data
|
|
assert "response" in data
|
|
assert data["request"]["method"] == "GET"
|
|
assert data["response"]["status_code"] == 200
|
|
|
|
def test_pair_from_dict(self, sample_pair):
|
|
"""Test creating pair from dictionary."""
|
|
data = sample_pair.to_dict()
|
|
restored = RequestResponsePair.from_dict(data)
|
|
|
|
assert restored.request.method == sample_pair.request.method
|
|
assert restored.response.status_code == sample_pair.response.status_code
|
|
|
|
|
|
class TestRecordingSession:
|
|
"""Tests for RecordingSession class."""
|
|
|
|
def test_init_session(self):
|
|
"""Test initializing a recording session."""
|
|
session = RecordingSession()
|
|
|
|
assert session.base_url is None
|
|
assert session.on_request is None
|
|
assert session.recordings == []
|
|
|
|
def test_init_with_base_url(self):
|
|
"""Test initializing with base URL."""
|
|
session = RecordingSession(base_url="https://api.example.com")
|
|
|
|
assert session.base_url == "https://api.example.com"
|
|
|
|
def test_build_url_absolute(self):
|
|
"""Test building URL for absolute URL."""
|
|
session = RecordingSession()
|
|
|
|
url = session._build_url("https://other.com/api")
|
|
|
|
assert url == "https://other.com/api"
|
|
|
|
def test_build_url_relative(self):
|
|
"""Test building URL for relative URL."""
|
|
session = RecordingSession(base_url="https://api.example.com")
|
|
|
|
url = session._build_url("/users")
|
|
|
|
assert url == "https://api.example.com/users"
|
|
|
|
def test_build_url_relative_no_base(self):
|
|
"""Test building URL for relative URL without base."""
|
|
session = RecordingSession()
|
|
|
|
url = session._build_url("/users")
|
|
|
|
assert url == "/users"
|
|
|
|
def test_record_request_adds_to_recordings(self):
|
|
"""Test that record_request adds to recordings."""
|
|
session = RecordingSession()
|
|
|
|
with patch(
|
|
"api_snapshot.recorder.recorder.requests.Session.request"
|
|
) as mock_request:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_response.text = '{"success": true}'
|
|
mock_request.return_value = mock_response
|
|
|
|
session.record_request("GET", "https://api.example.com/users")
|
|
|
|
assert len(session.recordings) == 1
|
|
assert session.recordings[0].request.method == "GET"
|
|
assert session.recordings[0].response.status_code == 200
|
|
|
|
def test_on_request_callback(self):
|
|
"""Test on_request callback is called."""
|
|
callback = MagicMock()
|
|
session = RecordingSession(on_request=callback)
|
|
|
|
with patch(
|
|
"api_snapshot.recorder.recorder.requests.Session.request"
|
|
) as mock_request:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {}
|
|
mock_response.text = "{}"
|
|
mock_request.return_value = mock_response
|
|
|
|
session.record_request("GET", "https://api.example.com")
|
|
|
|
callback.assert_called_once()
|
|
|
|
def test_clear_recordings(self):
|
|
"""Test clearing recordings."""
|
|
session = RecordingSession()
|
|
|
|
with patch(
|
|
"api_snapshot.recorder.recorder.requests.Session.request"
|
|
) as mock_request:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {}
|
|
mock_response.text = "{}"
|
|
mock_request.return_value = mock_response
|
|
|
|
session.record_request("GET", "https://api.example.com")
|
|
assert len(session.recordings) == 1
|
|
|
|
session.clear()
|
|
assert len(session.recordings) == 0
|
|
|
|
def test_get_recordings(self):
|
|
"""Test getting recordings."""
|
|
session = RecordingSession()
|
|
|
|
with patch(
|
|
"api_snapshot.recorder.recorder.requests.Session.request"
|
|
) as mock_request:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {}
|
|
mock_response.text = "{}"
|
|
mock_request.return_value = mock_response
|
|
|
|
session.record_request("GET", "https://api.example.com/1")
|
|
session.record_request("GET", "https://api.example.com/2")
|
|
|
|
recordings = session.get_recordings()
|
|
|
|
assert len(recordings) == 2
|
|
assert recordings[0].request.url.endswith("/1")
|
|
assert recordings[1].request.url.endswith("/2")
|
|
|
|
|
|
class TestRecordFunctions:
|
|
"""Tests for record_session and record_multiple functions."""
|
|
|
|
def test_record_session(self):
|
|
"""Test record_session function."""
|
|
with patch(
|
|
"api_snapshot.recorder.recorder.requests.Session.request"
|
|
) as mock_request:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {"Content-Type": "application/json"}
|
|
mock_response.text = '{"data": "test"}'
|
|
mock_request.return_value = mock_response
|
|
|
|
pairs = record_session(
|
|
url="https://api.example.com/api",
|
|
method="GET",
|
|
)
|
|
|
|
assert len(pairs) == 1
|
|
assert pairs[0].request.method == "GET"
|
|
assert pairs[0].response.status_code == 200
|
|
|
|
def test_record_multiple(self):
|
|
"""Test record_multiple function."""
|
|
with patch(
|
|
"api_snapshot.recorder.recorder.requests.Session.request"
|
|
) as mock_request:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {}
|
|
mock_response.text = "{}"
|
|
mock_request.return_value = mock_response
|
|
|
|
config = [
|
|
{"method": "GET", "url": "https://api.example.com/1"},
|
|
{"method": "POST", "url": "https://api.example.com/2"},
|
|
{"method": "DELETE", "url": "https://api.example.com/3"},
|
|
]
|
|
|
|
pairs = record_multiple(requests_config=config)
|
|
|
|
assert len(pairs) == 3
|
|
assert pairs[0].request.method == "GET"
|
|
assert pairs[1].request.method == "POST"
|
|
assert pairs[2].request.method == "DELETE"
|