188 lines
6.5 KiB
Python
188 lines
6.5 KiB
Python
"""Tests for traffic analyzer and stats generator."""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
|
|
from http_log_explorer.analyzers import StatsGenerator, TrafficAnalyzer
|
|
from http_log_explorer.models import FilterCriteria, HTTPEntry, Request, Response
|
|
|
|
|
|
def make_entry(
|
|
method: str = "GET",
|
|
url: str = "https://api.example.com/test",
|
|
status: int = 200,
|
|
body: str | None = None,
|
|
duration_ms: float = 100.0,
|
|
) -> HTTPEntry:
|
|
"""Create a test HTTPEntry."""
|
|
return HTTPEntry(
|
|
id=f"entry-{method}-{status}",
|
|
request=Request(
|
|
method=method,
|
|
url=url,
|
|
headers={"Content-Type": "application/json"},
|
|
body=body,
|
|
),
|
|
response=Response(
|
|
status=status,
|
|
status_text="OK",
|
|
body=body,
|
|
content_type="application/json",
|
|
response_time_ms=duration_ms,
|
|
),
|
|
timestamp=datetime.now(),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_entries():
|
|
"""Create sample HTTP entries for testing."""
|
|
return [
|
|
make_entry("GET", "https://api.example.com/users", 200, duration_ms=50),
|
|
make_entry("POST", "https://api.example.com/users", 201, duration_ms=100),
|
|
make_entry("GET", "https://api.example.com/users/1", 200, duration_ms=75),
|
|
make_entry("PUT", "https://api.example.com/users/1", 200, duration_ms=80),
|
|
make_entry("DELETE", "https://api.example.com/users/1", 204, duration_ms=30),
|
|
make_entry("GET", "https://api.example.com/posts", 404, duration_ms=25),
|
|
make_entry("POST", "https://api.example.com/posts", 500, duration_ms=200),
|
|
]
|
|
|
|
|
|
class TestTrafficAnalyzer:
|
|
"""Tests for TrafficAnalyzer."""
|
|
|
|
def test_filter_by_method(self, sample_entries):
|
|
"""Test filtering by HTTP method."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
get_entries = analyzer.by_method(["GET"])
|
|
|
|
assert len(get_entries) == 3
|
|
assert all(e.request.method == "GET" for e in get_entries)
|
|
|
|
def test_filter_by_status(self, sample_entries):
|
|
"""Test filtering by status code."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
success_entries = analyzer.by_status([200, 201, 204])
|
|
|
|
assert len(success_entries) == 5
|
|
assert all(e.response.status < 300 for e in success_entries)
|
|
|
|
def test_filter_by_url_pattern(self, sample_entries):
|
|
"""Test filtering by URL pattern."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
user_entries = analyzer.by_url(r"/users")
|
|
|
|
assert len(user_entries) == 5
|
|
assert all("/users" in e.request.url for e in user_entries)
|
|
|
|
def test_filter_by_content_type(self, sample_entries):
|
|
"""Test filtering by content type."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
json_entries = analyzer.by_content_type(["application/json"])
|
|
|
|
assert len(json_entries) == 7
|
|
|
|
def test_successful_requests(self, sample_entries):
|
|
"""Test getting successful requests (2xx)."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
success = analyzer.successful_requests()
|
|
|
|
assert len(success) == 5
|
|
assert all(200 <= e.response.status < 300 for e in success)
|
|
|
|
def test_client_errors(self, sample_entries):
|
|
"""Test getting client errors (4xx)."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
errors = analyzer.client_errors()
|
|
|
|
assert len(errors) == 1
|
|
assert errors[0].response.status == 404
|
|
|
|
def test_server_errors(self, sample_entries):
|
|
"""Test getting server errors (5xx)."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
errors = analyzer.server_errors()
|
|
|
|
assert len(errors) == 1
|
|
assert errors[0].response.status == 500
|
|
|
|
def test_search(self, sample_entries):
|
|
"""Test search across URL and bodies."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
results = analyzer.search("users")
|
|
|
|
assert len(results) == 5
|
|
|
|
def test_get_entry_by_id(self, sample_entries):
|
|
"""Test getting entry by ID."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
entry = analyzer.get_entry_by_id("entry-GET-200")
|
|
|
|
assert entry is not None
|
|
assert entry.request.method == "GET"
|
|
|
|
def test_get_entry_by_id_not_found(self, sample_entries):
|
|
"""Test getting non-existent entry."""
|
|
analyzer = TrafficAnalyzer(sample_entries)
|
|
entry = analyzer.get_entry_by_id("non-existent")
|
|
|
|
assert entry is None
|
|
|
|
|
|
class TestStatsGenerator:
|
|
"""Tests for StatsGenerator."""
|
|
|
|
def test_total_requests(self, sample_entries):
|
|
"""Test total request count."""
|
|
generator = StatsGenerator(sample_entries)
|
|
stats = generator.generate()
|
|
|
|
assert stats.total_requests == 7
|
|
|
|
def test_method_distribution(self, sample_entries):
|
|
"""Test method distribution."""
|
|
generator = StatsGenerator(sample_entries)
|
|
stats = generator.generate()
|
|
|
|
assert stats.method_distribution.get("GET") == 3
|
|
assert stats.method_distribution.get("POST") == 2
|
|
assert stats.method_distribution.get("PUT") == 1
|
|
assert stats.method_distribution.get("DELETE") == 1
|
|
|
|
def test_status_breakdown(self, sample_entries):
|
|
"""Test status code breakdown."""
|
|
generator = StatsGenerator(sample_entries)
|
|
stats = generator.generate()
|
|
|
|
assert stats.status_breakdown.get(200) == 3
|
|
assert stats.status_breakdown.get(201) == 1
|
|
assert stats.status_breakdown.get(204) == 1
|
|
assert stats.status_breakdown.get(404) == 1
|
|
assert stats.status_breakdown.get(500) == 1
|
|
|
|
def test_response_time_stats(self, sample_entries):
|
|
"""Test response time statistics."""
|
|
generator = StatsGenerator(sample_entries)
|
|
stats = generator.generate()
|
|
|
|
assert stats.response_time_stats["min"] == 25.0
|
|
assert stats.response_time_stats["max"] == 200.0
|
|
assert stats.response_time_stats["avg"] == pytest.approx(80.0, rel=1)
|
|
|
|
def test_endpoint_count(self, sample_entries):
|
|
"""Test endpoint counting."""
|
|
generator = StatsGenerator(sample_entries)
|
|
stats = generator.generate()
|
|
|
|
assert "/users" in stats.endpoint_count
|
|
assert "/posts" in stats.endpoint_count
|
|
|
|
def test_to_dict(self, sample_entries):
|
|
"""Test converting stats to dictionary."""
|
|
generator = StatsGenerator(sample_entries)
|
|
stats_dict = generator.to_dict()
|
|
|
|
assert stats_dict["total_requests"] == 7
|
|
assert "method_distribution" in stats_dict
|
|
assert "status_breakdown" in stats_dict
|