69 lines
No EOL
2.7 KiB
Python
69 lines
No EOL
2.7 KiB
Python
import pytest
|
|
import statistics
|
|
from queue_separation import main
|
|
|
|
@pytest.fixture
|
|
def sample_jobs():
|
|
# Mix aus regulären und Hotspot-Jobs
|
|
jobs = [
|
|
{"id": i, "runtime": 1.0 + i * 0.1, "queue": "main" if i % 3 else "hotspot"}
|
|
for i in range(1, 31)
|
|
]
|
|
return jobs
|
|
|
|
|
|
def test_simulate_queue_separation_structure(sample_jobs):
|
|
main_queue = [j for j in sample_jobs if j["queue"] == "main"]
|
|
hotspot_queue = [j for j in sample_jobs if j["queue"] == "hotspot"]
|
|
|
|
result = main.simulate_queue_separation(
|
|
jobs=sample_jobs, hotspot_queue=hotspot_queue, main_queue=main_queue
|
|
)
|
|
|
|
# Strukturvalidierung
|
|
assert isinstance(result, dict), "Simulationsergebnis muss ein Dict sein"
|
|
for field in ("retry_tail_p99", "bandwidth", "hotspot_percentage"):
|
|
assert field in result, f"Fehlendes Feld {field} in PerformanceMetrics"
|
|
assert isinstance(result[field], (float, int)), f"{field} muss numerisch sein"
|
|
|
|
|
|
def test_simulation_value_ranges(sample_jobs):
|
|
main_queue = [j for j in sample_jobs if j["queue"] == "main"]
|
|
hotspot_queue = [j for j in sample_jobs if j["queue"] == "hotspot"]
|
|
|
|
result = main.simulate_queue_separation(
|
|
jobs=sample_jobs, hotspot_queue=hotspot_queue, main_queue=main_queue
|
|
)
|
|
|
|
# Werte sollen in plausiblen Bereichen liegen
|
|
assert 0.0 <= result["retry_tail_p99"] <= 100.0, "retry_tail_p99 außerhalb des Bereichs"
|
|
assert result["bandwidth"] > 0, "Bandbreite sollte positiv sein"
|
|
assert 0.0 <= result["hotspot_percentage"] <= 1.0, "Hotspot-Prozentsatz außerhalb des Bereichs"
|
|
|
|
|
|
def test_tail_latency_increases_with_hotspots(sample_jobs):
|
|
# Weniger Hotspots -> bessere Performance erwartet
|
|
main_queue = [j for j in sample_jobs if j["queue"] == "main"]
|
|
hotspot_queue = [j for j in sample_jobs if j["queue"] == "hotspot"]
|
|
|
|
results_low_hotspot = main.simulate_queue_separation(
|
|
jobs=sample_jobs[:10], hotspot_queue=hotspot_queue[:1], main_queue=main_queue[:9]
|
|
)
|
|
|
|
results_high_hotspot = main.simulate_queue_separation(
|
|
jobs=sample_jobs, hotspot_queue=hotspot_queue + main_queue[:5], main_queue=main_queue[5:]
|
|
)
|
|
|
|
assert (
|
|
results_high_hotspot["retry_tail_p99"] >= results_low_hotspot["retry_tail_p99"]
|
|
), "Mehr Hotspots sollten höhere Tail-Latenz erzeugen"
|
|
assert results_high_hotspot["bandwidth"] >= 0 and results_low_hotspot["bandwidth"] >= 0
|
|
|
|
|
|
def test_invalid_input_handling():
|
|
# Ungültiger Input sollte Exception oder Validation Error werfen
|
|
invalid_jobs = [{"runtime": "fast", "queue": 123}] # Typfehler
|
|
with pytest.raises((AssertionError, TypeError, ValueError)):
|
|
main.simulate_queue_separation(
|
|
jobs=invalid_jobs, hotspot_queue=[], main_queue=[]
|
|
) |