157 lines
6.8 KiB
Python
157 lines
6.8 KiB
Python
import time
|
|
import json
|
|
import random
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.main import Telemetry, create_app, parse_line
|
|
|
|
|
|
def json_record(**changes):
|
|
record = dict(timestamp="1720000000.123", host="example.test", method="POST",
|
|
status="201", incoming="6460500000", outgoing="240", request_time="180.0",
|
|
upstream_response_time="0.100, 0.200", client_address="2001:db8::1",
|
|
path='/upload/"file"/<script>')
|
|
record.update(changes)
|
|
return json.dumps(record)
|
|
|
|
|
|
def test_json_details_and_legacy_compatibility():
|
|
event = parse_line(json_record())
|
|
assert event["client_address"] == "2001:db8::1"
|
|
assert event["path"] == '/upload/"file"/<script>'
|
|
assert event["method"] == "POST"
|
|
assert event["incoming"] == 6460500000
|
|
assert event["upstream_time"] == pytest.approx(.15)
|
|
assert parse_line(json_record(path="/upload?token=secret"))["path"] == "/upload"
|
|
for upstream in ("", "-"):
|
|
assert parse_line(json_record(upstream_response_time=upstream))["upstream_time"] is None
|
|
legacy = parse_line("1720000000.123 example.test GET 200 120 240 .1 -")
|
|
assert legacy["client_address"] is None
|
|
assert legacy["path"] is None
|
|
assert legacy["method"] == "GET"
|
|
|
|
|
|
@pytest.mark.parametrize("line", [
|
|
'{"timestamp":', '{}', '{"timestamp":null}',
|
|
json_record(incoming="-1"), json_record(outgoing="2.5"),
|
|
json_record(timestamp="NaN"), json_record(request_time="Infinity"),
|
|
json_record(status="600"), json_record(client_address=[]),
|
|
json_record(upstream_response_time="not-a-time"),
|
|
])
|
|
def test_invalid_json_records_are_ignored(line):
|
|
assert parse_line(line) is None
|
|
|
|
|
|
def test_top_requests_matches_full_sort_and_window_boundaries(monkeypatch):
|
|
now = 1720004000
|
|
monkeypatch.setattr("app.main.time.time", lambda: now)
|
|
telemetry = Telemetry(3600)
|
|
randomizer = random.Random(42)
|
|
all_records = []
|
|
# Dense seconds exercise per-bucket truncation, while boundary seconds
|
|
# exercise membership in every selectable window.
|
|
ages = (0, 59, 60, 299, 300, 899, 900, 3599, 3600)
|
|
for age in ages:
|
|
for _ in range(30):
|
|
event = parse_line(json_record(timestamp=str(now-age),
|
|
incoming=str(randomizer.randrange(100)),
|
|
outgoing=str(randomizer.randrange(100))))
|
|
telemetry.ingest(event)
|
|
all_records.append(dict(event, id=telemetry.next_request_id,
|
|
size=max(event["incoming"], event["outgoing"])))
|
|
snapshot = telemetry.snapshot(60)
|
|
for window in (60, 300, 900, 3600):
|
|
expected = sorted((record for record in all_records if now-window+1 <= record["timestamp"] <= now),
|
|
key=lambda record: (record["size"], record["timestamp"], record["id"]), reverse=True)[:10]
|
|
actual = snapshot["top_requests_by_window"][str(window)]
|
|
assert [record["id"] for record in actual] == [record["id"] for record in expected]
|
|
assert all(len(records) <= 10 for records in telemetry.request_buckets.values())
|
|
assert len(telemetry.request_buckets) == len(ages)-1
|
|
assert snapshot["summary"]["incoming"] == sum(r["incoming"] for r in all_records if r["timestamp"] >= now-59)/60
|
|
assert telemetry.total_incoming == sum(r["incoming"] for r in all_records)
|
|
|
|
|
|
def test_directions_ties_unique_ids_and_idle_expiration(monkeypatch):
|
|
now = 1720004000
|
|
monkeypatch.setattr("app.main.time.time", lambda: now)
|
|
telemetry = Telemetry(60)
|
|
for incoming, outgoing in ((100, 100), (100, 100), (1, 200), (300, 1)):
|
|
telemetry.ingest(parse_line(json_record(timestamp=str(now), incoming=str(incoming), outgoing=str(outgoing))))
|
|
records = telemetry.snapshot(60)["top_requests_by_window"]["60"]
|
|
assert [record["id"] for record in records] == [4, 3, 2, 1]
|
|
assert [record["direction"] for record in records] == ["inbound", "outbound", "inbound", "inbound"]
|
|
now += 59
|
|
assert len(telemetry.snapshot(60)["top_requests_by_window"]["60"]) == 4
|
|
now += 1
|
|
assert all(not records for records in telemetry.snapshot(3600)["top_requests_by_window"].values())
|
|
assert telemetry.request_buckets == {}
|
|
|
|
|
|
def test_out_of_order_old_and_future_requests(monkeypatch):
|
|
now = 1720004000
|
|
monkeypatch.setattr("app.main.time.time", lambda: now)
|
|
telemetry = Telemetry(60)
|
|
for age in (0, 20, 59, 61, -100):
|
|
telemetry.ingest(parse_line(json_record(timestamp=str(now-age))))
|
|
records = telemetry.snapshot(60)["top_requests_by_window"]["60"]
|
|
assert len(records) == 4
|
|
assert records[0]["timestamp"] == now+100
|
|
assert len(telemetry.request_buckets) == 3
|
|
|
|
|
|
def test_rest_and_sse_include_request_details(monkeypatch):
|
|
now = 1720004000
|
|
monkeypatch.setattr("app.main.time.time", lambda: now)
|
|
app = create_app()
|
|
app.state.telemetry.ingest(parse_line(json_record(timestamp=str(now))))
|
|
endpoints = {route.path: route.endpoint for route in app.routes if hasattr(route, "endpoint")}
|
|
|
|
async def check():
|
|
rest = await endpoints["/api/telemetry"](window=60)
|
|
response = await endpoints["/api/events"](request=None, window=60)
|
|
try:
|
|
first = await anext(response.body_iterator)
|
|
streamed = json.loads(first.removeprefix("data: ").strip())
|
|
assert streamed["top_requests_by_window"] == rest["top_requests_by_window"]
|
|
record = streamed["top_requests_by_window"]["60"][0]
|
|
assert record["client_address"] == "2001:db8::1"
|
|
assert record["size"] == 6460500000
|
|
finally:
|
|
await response.body_iterator.aclose()
|
|
assert not app.state.telemetry.subscribers
|
|
|
|
asyncio.run(check())
|
|
|
|
|
|
def test_parse_line_and_multiple_upstreams():
|
|
event = parse_line("1720000000.123 example.test GET 200 120 240 0.150 0.100, 0.200")
|
|
assert event["host"] == "example.test"
|
|
assert event["incoming"] == 120
|
|
assert event["upstream_time"] == pytest.approx(0.15)
|
|
|
|
|
|
def test_parse_line_with_empty_trailing_upstream_time():
|
|
event = parse_line("1720000000.123 example.test GET 200 120 240 0.150")
|
|
assert event is not None
|
|
assert event["host"] == "example.test"
|
|
assert event["upstream_time"] is None
|
|
|
|
|
|
def test_malformed_lines_are_ignored():
|
|
assert parse_line("not enough fields") is None
|
|
assert parse_line("1 host GET nope 1 2 0.1 -") is None
|
|
assert parse_line("1 host GET 600 1 2 0.1 -") is None
|
|
|
|
|
|
def test_telemetry_is_aggregated_and_bounded():
|
|
telemetry = Telemetry(60)
|
|
base = int(time.time()) - 30
|
|
for second in range(70):
|
|
telemetry.ingest(parse_line(f"{base + second} host GET 200 10 20 .1 -"))
|
|
assert telemetry.total_incoming == 700
|
|
assert len(telemetry.global_series.buckets) <= 60
|
|
snapshot = telemetry.snapshot(60)
|
|
assert snapshot["summary"]["requests"] >= 1
|
|
assert snapshot["hosts"] == ["host"]
|