79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
"""Unit coverage for bounded Activation Seam telemetry."""
|
|
|
|
from meshnet_node.seam_telemetry import GenerationTelemetry
|
|
|
|
|
|
def test_seam_telemetry_aggregates_bytes_latency_and_correlates_ids():
|
|
telemetry = GenerationTelemetry("route-session-1", report_every=2, report_interval=100, now=0.0)
|
|
|
|
assert telemetry.record_seam(
|
|
activation_id="activation-1", phase="prefill", hop=0, node="node-a",
|
|
latency_seconds=0.012, wire_bytes=120, response_bytes=240,
|
|
connection_reused=False, now=0.1,
|
|
)
|
|
telemetry.mark_reported(now=0.1)
|
|
assert telemetry.record_seam(
|
|
activation_id="activation-2", phase="prefill", hop=0, node="node-a",
|
|
latency_seconds=0.008, wire_bytes=80, response_bytes=160,
|
|
connection_reused=True, now=0.2,
|
|
)
|
|
telemetry.note_tokens(4)
|
|
|
|
snapshot = telemetry.snapshot(now=2.0)
|
|
assert snapshot["session_id"] == "route-session-1"
|
|
assert snapshot["tokens_per_sec"] == 2.0
|
|
assert snapshot["seams"] == [{
|
|
"phase": "prefill", "hop": 0, "node": "node-a", "activations": 2,
|
|
"latency_ms": 20.0, "avg_latency_ms": 10.0, "wire_bytes": 200,
|
|
"response_bytes": 400, "connection_reuse": 1,
|
|
"compression_input_bytes": 0, "compression_output_bytes": 0, "compression_ms": 0.0,
|
|
"decompression_input_bytes": 0, "decompression_output_bytes": 0, "decompression_ms": 0.0,
|
|
"last_activation_id": "activation-2",
|
|
}]
|
|
|
|
|
|
def test_seam_telemetry_includes_compression_work_and_byte_counters():
|
|
telemetry = GenerationTelemetry("route-session-compression", now=0.0)
|
|
telemetry.record_compression(
|
|
phase="prefill", hop=0, node="node-a", input_bytes=1000, output_bytes=200,
|
|
elapsed_seconds=0.003,
|
|
)
|
|
telemetry.record_compression(
|
|
phase="prefill", hop=0, node="node-a", input_bytes=200, output_bytes=1000,
|
|
elapsed_seconds=0.001, decompression=True,
|
|
)
|
|
seam = telemetry.snapshot(now=1.0)["seams"][0]
|
|
assert seam["compression_input_bytes"] == 1000
|
|
assert seam["compression_output_bytes"] == 200
|
|
assert seam["compression_ms"] == 3.0
|
|
assert seam["decompression_input_bytes"] == 200
|
|
assert seam["decompression_output_bytes"] == 1000
|
|
assert seam["decompression_ms"] == 1.0
|
|
|
|
|
|
def test_seam_telemetry_reports_on_bounded_cadence_and_cleans_up():
|
|
telemetry = GenerationTelemetry("route-session-2", report_every=3, report_interval=5, now=0.0)
|
|
|
|
for count in range(1, 4):
|
|
due = telemetry.record_seam(
|
|
activation_id=f"activation-{count}", phase="decode", hop=1, node="node-b",
|
|
latency_seconds=0.001, wire_bytes=10, response_bytes=20,
|
|
connection_reused=True, now=float(count),
|
|
)
|
|
if due:
|
|
telemetry.mark_reported(now=float(count))
|
|
assert not telemetry.report_due
|
|
assert telemetry.record_seam(
|
|
activation_id="activation-4", phase="decode", hop=1, node="node-b",
|
|
latency_seconds=0.001, wire_bytes=10, response_bytes=20,
|
|
connection_reused=True, now=9.0,
|
|
)
|
|
|
|
telemetry.close()
|
|
assert telemetry.snapshot(now=10.0)["seams"] == []
|
|
assert not telemetry.record_seam(
|
|
activation_id="late", phase="decode", hop=1, node="node-b",
|
|
latency_seconds=0.001, wire_bytes=10, response_bytes=20,
|
|
connection_reused=True, now=10.0,
|
|
)
|