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