proto-socket/python/test/test_stress_harness.py

"""Tests for the stress harness helpers: LatencyReservoir and summarize(request_count).

These tests validate the bounded latency sampling introduced in the
m-performance-hotspot-optimization / python-sustained-memory-metric plan.
"""

import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import bench.stress as stress_module
from bench.stress import LatencyReservoir, SUSTAINED_MAX_LATENCY_SAMPLES, Stability


class TestLatencyReservoir:
    """Verify bounded sample count and unbounded request count tracking."""

    def test_reservoir_tracks_all_count(self) -> None:
        """count must reflect the total number of add() calls, not just stored samples."""
        sampler = LatencyReservoir(max_samples=10)
        for i in range(500):
            sampler.add(float(i))
        assert sampler.count == 500

    def test_reservoir_bounds_samples_to_max(self) -> None:
        """samples list must never exceed max_samples length."""
        sampler = LatencyReservoir(max_samples=SUSTAINED_MAX_LATENCY_SAMPLES)
        for i in range(200_000):
            sampler.add(float(i))
        assert len(sampler.samples) == SUSTAINED_MAX_LATENCY_SAMPLES

    def test_reservoir_initial_state(self) -> None:
        assert LatencyReservoir(100).count == 0
        assert LatencyReservoir(100).samples == []

    def test_reservoir_small_values_fully_preserved(self) -> None:
        """When inputs are within the max, all values are stored."""
        sampler = LatencyReservoir(max_samples=5)
        for i in range(5):
            sampler.add(float(i * 10))
        assert len(sampler.samples) == 5
        assert sampler.count == 5

    def test_reservoir_deterministic_slot_mapping(self) -> None:
        """The slot formula must be deterministic for the same count."""
        sampler = LatencyReservoir(max_samples=3)
        sampler.add(1.0)
        sampler.add(2.0)
        sampler.add(3.0)
        sampler.add(4.0)
        sampler2 = LatencyReservoir(max_samples=3)
        for v in (1.0, 2.0, 3.0, 4.0):
            sampler2.add(v)
        assert sampler.samples == sampler2.samples


class TestSummarizeRequestCount:
    """Verify that summarize() actually passes request_count to emit_row().

    These tests use monkeypatch to intercept emit_row() kwargs so that the
    real summarize() function is exercised. A mock that reimplements the same
    logic would pass even if summarize() had a bug — this would not.
    """

    def test_summarize_uses_request_count_for_throughput(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """When request_count=200, emit_row must receive requests=200 and throughput=200.0."""
        captured: dict = {}

        def fake_emit_row(**kwargs: object) -> None:
            captured.update(kwargs)

        monkeypatch.setattr(stress_module, "emit_row", fake_emit_row)

        stability = Stability()
        stress_module.summarize(
            profile="sustained",
            axis="duration=1000ms",
            transport="tcp",
            latencies=[1.0, 2.0, 3.0],
            elapsed_ms=1000.0,
            client_count=1,
            stability=stability,
            mem=12.3,
            request_count=200,
        )

        assert captured.get("requests") == 200
        assert pytest.approx(float(captured["throughput"])) == 200.0
        # p50/p95/p99 are computed from the bounded sample [1.0, 2.0, 3.0], NOT from request_count
        assert pytest.approx(float(captured["p50"])) == 2.0
        assert pytest.approx(float(captured["p95"])) == 3.0
        assert pytest.approx(float(captured["p99"])) == 3.0

    def test_summarize_defaults_to_len_latencies(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Without request_count, emit_row must receive requests=len(latencies)."""
        captured: dict = {}

        def fake_emit_row(**kwargs: object) -> None:
            captured.update(kwargs)

        original_emit = stress_module.emit_row
        monkeypatch.setattr(stress_module, "emit_row", fake_emit_row)

        stability = Stability()
        stress_module.summarize(
            profile="roundtrip",
            axis="concurrency=1",
            transport="tcp",
            latencies=[1.5, 2.5, 3.5],
            elapsed_ms=1000.0,
            client_count=1,
            stability=stability,
            mem=10.0,
        )

        assert captured.get("requests") == 3
        assert pytest.approx(float(captured["throughput"])) == 3.0

    def test_summarize_emits_correct_payload_and_status(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Verify that non-latency fields (payload_bytes, status, mem) are also correct."""
        captured: dict = {}

        def fake_emit_row(**kwargs: object) -> None:
            captured.update(kwargs)

        monkeypatch.setattr(stress_module, "emit_row", fake_emit_row)

        stability = Stability()
        stress_module.summarize(
            profile="sustained",
            axis="duration=2000ms",
            transport="tcp",
            latencies=[5.0],
            elapsed_ms=2000.0,
            client_count=1,
            stability=stability,
            mem=45.6,
            request_count=1000,
        )

        assert captured.get("profile") == "sustained"
        assert captured.get("transport") == "tcp"
        assert captured.get("client_count") == 1
        assert captured.get("requests") == 1000
        assert pytest.approx(float(captured["throughput"])) == 1000.0 / 2000.0 * 1000.0
        assert captured.get("mem") == 45.6
        assert captured.get("payload_bytes") == stress_module.test_data_payload_bytes("req-0")
        assert captured.get("p50") == 5.0  # single sample => all percentiles = 5.0