154 lines
4.2 KiB
Python
154 lines
4.2 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
from typing import Dict, Any, List
|
|
|
|
import pytest
|
|
|
|
from api_snapshot.recorder.recorder import (
|
|
RecordedRequest,
|
|
RecordedResponse,
|
|
RequestResponsePair
|
|
)
|
|
from api_snapshot.snapshot.manager import (
|
|
Snapshot,
|
|
SnapshotMetadata,
|
|
SnapshotManager
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for tests."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield tmpdir
|
|
|
|
|
|
@pytest.fixture
|
|
def snapshot_manager(temp_dir):
|
|
"""Create a SnapshotManager instance with temp directory."""
|
|
return SnapshotManager(temp_dir)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_request():
|
|
"""Create a sample recorded request."""
|
|
return RecordedRequest(
|
|
method="GET",
|
|
url="https://api.example.com/users",
|
|
headers={"Accept": "application/json", "Authorization": "Bearer token123"},
|
|
body=None
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_response():
|
|
"""Create a sample recorded response."""
|
|
return RecordedResponse(
|
|
status_code=200,
|
|
headers={"Content-Type": "application/json"},
|
|
body='[{"id": 1, "name": "John"}]',
|
|
latency_ms=150
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_pair(sample_request, sample_response):
|
|
"""Create a sample request-response pair."""
|
|
return RequestResponsePair(request=sample_request, response=sample_response)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_snapshot(sample_pair):
|
|
"""Create a sample snapshot."""
|
|
metadata = SnapshotMetadata(
|
|
version="1.0",
|
|
timestamp="2024-01-01T00:00:00",
|
|
description="Test snapshot"
|
|
)
|
|
return Snapshot(metadata=metadata, requests=[sample_pair])
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_snapshot_dict(sample_snapshot):
|
|
"""Create a sample snapshot as dictionary."""
|
|
return sample_snapshot.to_dict()
|
|
|
|
|
|
@pytest.fixture
|
|
def snapshot_file(temp_dir):
|
|
"""Create a temporary snapshot file."""
|
|
data = {
|
|
"metadata": {
|
|
"version": "1.0",
|
|
"timestamp": "2024-01-01T00:00:00",
|
|
"description": "Test snapshot",
|
|
"source_url": "https://api.example.com",
|
|
"tags": ["test", "sample"]
|
|
},
|
|
"requests": [
|
|
{
|
|
"request": {
|
|
"method": "GET",
|
|
"url": "https://api.example.com/users",
|
|
"headers": {"Accept": "application/json"},
|
|
"body": None,
|
|
"timestamp": "2024-01-01T00:00:00"
|
|
},
|
|
"response": {
|
|
"status_code": 200,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": '[{"id": 1, "name": "John"}]',
|
|
"latency_ms": 100
|
|
}
|
|
},
|
|
{
|
|
"request": {
|
|
"method": "POST",
|
|
"url": "https://api.example.com/users",
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": '{"name": "Jane"}',
|
|
"timestamp": "2024-01-01T00:00:01"
|
|
},
|
|
"response": {
|
|
"status_code": 201,
|
|
"headers": {"Content-Type": "application/json"},
|
|
"body": '{"id": 2, "name": "Jane"}',
|
|
"latency_ms": 200
|
|
}
|
|
}
|
|
]
|
|
}
|
|
|
|
path = os.path.join(temp_dir, "test_snapshot.json")
|
|
with open(path, "w") as f:
|
|
json.dump(data, f)
|
|
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def multiple_endpoints_snapshot(temp_dir):
|
|
"""Create a snapshot with multiple endpoints."""
|
|
manager = SnapshotManager(temp_dir)
|
|
|
|
pairs = []
|
|
|
|
for i in range(3):
|
|
req = RecordedRequest(
|
|
method=["GET", "POST", "DELETE"][i],
|
|
url=f"https://api.example.com/items/{i}",
|
|
headers={},
|
|
body=None
|
|
)
|
|
resp = RecordedResponse(
|
|
status_code=[200, 201, 204][i],
|
|
headers={"Content-Type": "application/json"},
|
|
body=f'{{"id": {i}}}',
|
|
latency_ms=50 * (i + 1)
|
|
)
|
|
pairs.append(RequestResponsePair(request=req, response=resp))
|
|
|
|
manager.save_snapshot("multi-endpoint", requests=pairs, description="Multiple endpoints")
|
|
return manager
|