tegwick
bfd1a2d46d
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
136 lines
4.7 KiB
Python
136 lines
4.7 KiB
Python
import importlib.util
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"pilot", Path(__file__).with_name("synthetic_pilot.py")
|
|
)
|
|
pilot = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(pilot)
|
|
|
|
|
|
class SyntheticContractTests(unittest.TestCase):
|
|
def config(self):
|
|
return dict(
|
|
users=2,
|
|
workers=2,
|
|
iterations=20,
|
|
tenders=100,
|
|
document_bytes=1024,
|
|
think_seconds=0.2,
|
|
timeout_seconds=30,
|
|
ordinary_p95_seconds=2,
|
|
duration_seconds=180,
|
|
idle_seconds=30,
|
|
)
|
|
|
|
def test_invalid_workload_fails_before_runtime(self):
|
|
for key, value in [
|
|
("users", 0),
|
|
("users", True),
|
|
("workers", -1),
|
|
("iterations", 1.2),
|
|
("document_bytes", 7),
|
|
("timeout_seconds", 0),
|
|
("duration_seconds", 0),
|
|
("idle_seconds", -1),
|
|
("ordinary_p95_seconds", float("nan")),
|
|
("think_seconds", -1),
|
|
]:
|
|
with self.subTest(key=key, value=value), self.assertRaises(ValueError):
|
|
pilot.validate(self.config() | {key: value})
|
|
|
|
def test_nearest_rank_p95_and_errors_include_failed_requests(self):
|
|
rows = [{"seconds": n, "ok": n != 20} for n in range(1, 21)]
|
|
self.assertEqual(pilot.summarize(rows), {"requests": 20, "errors": 1, "p95_seconds": 19})
|
|
|
|
def test_empty_measurement_is_unknown(self):
|
|
self.assertIsNone(pilot.summarize([])["p95_seconds"])
|
|
|
|
def test_default_is_two_users_and_two_seconds(self):
|
|
self.assertEqual(pilot.validate(self.config())["users"], 2)
|
|
self.assertEqual(self.config()["ordinary_p95_seconds"], 2)
|
|
|
|
|
|
class LifecycleTests(unittest.TestCase):
|
|
def test_process_group_cleanup_stops_real_child(self):
|
|
process = subprocess.Popen(
|
|
[sys.executable, "-c", "import time; time.sleep(60)"], start_new_session=True
|
|
)
|
|
try:
|
|
raise RuntimeError("injected worker failure")
|
|
except RuntimeError:
|
|
pilot.stop_process(process)
|
|
self.assertIsNotNone(process.poll())
|
|
|
|
def test_sampler_exception_is_retained_as_gap(self):
|
|
errors = []
|
|
|
|
def broken_sampler():
|
|
raise OSError("injected collection failure")
|
|
|
|
pilot.guarded_sampler(broken_sampler, errors)
|
|
gaps = pilot.sampling_gaps([], [], errors)
|
|
self.assertEqual(gaps[0]["error_type"], "OSError")
|
|
self.assertTrue(gaps)
|
|
|
|
def test_sparse_samples_are_not_complete(self):
|
|
rows = [dict(monotonic_seconds=0), dict(monotonic_seconds=5)]
|
|
self.assertTrue(pilot.sampling_gaps(rows, [{}], []))
|
|
|
|
|
|
class SizingTests(unittest.TestCase):
|
|
def test_failed_measurement_does_not_recommend(self):
|
|
spec = importlib.util.spec_from_file_location(
|
|
"sizing", Path(__file__).with_name("summarize_synthetic_sizing.py")
|
|
)
|
|
sizing = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(sizing)
|
|
with self.assertRaises(ValueError):
|
|
sizing.recommend({"runtime_samples": [], "complete": False}, {})
|
|
|
|
def test_sampler_gap_prevents_recommendation(self):
|
|
spec = importlib.util.spec_from_file_location(
|
|
"sizing", Path(__file__).with_name("summarize_synthetic_sizing.py")
|
|
)
|
|
sizing = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(sizing)
|
|
with self.assertRaisesRegex(ValueError, "sampling gaps"):
|
|
sizing.recommend({"measurement_complete": False}, {})
|
|
|
|
def test_cpu_delta_and_rounding(self):
|
|
spec = importlib.util.spec_from_file_location(
|
|
"sizing", Path(__file__).with_name("summarize_synthetic_sizing.py")
|
|
)
|
|
sizing = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(sizing)
|
|
rows = [
|
|
dict(
|
|
phase="ordinary",
|
|
pids=[1],
|
|
monotonic_seconds=i,
|
|
cpu_seconds=i * 0.1,
|
|
rss_bytes=100 * 1024**2,
|
|
)
|
|
for i in range(3)
|
|
]
|
|
margin = dict(
|
|
cpu_request_multiplier=1.5,
|
|
cpu_limit_multiplier=1.5,
|
|
memory_request_multiplier=1.25,
|
|
memory_limit_multiplier=1.5,
|
|
)
|
|
result = sizing.recommend(
|
|
dict(runtime_samples=rows, complete=True, provisional_pass=True, config={"workers": 2}),
|
|
margin,
|
|
)
|
|
self.assertEqual(
|
|
result["suggested_application"],
|
|
dict(cpu_request_m=150, cpu_limit_m=200, memory_request_mib=128, memory_limit_mib=192),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|