vergabe-teilnahme/scripts/test_synthetic_pilot.py

import importlib.util
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

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 PlanTests(unittest.TestCase):
    def test_edit_dependency_and_unknown_operation_fail_before_runtime(self):
        for operations in [
            ["task-edit"],
            ["tender-create", "lot-edit"],
            ["unknown"],
            ["tender-create", "tender-create"],
        ]:
            with self.subTest(operations=operations), self.assertRaises(ValueError):
                pilot.operation_plan(operations)

    def test_repeated_reads_change_derived_expectation(self):
        plan = pilot.operation_plan(["list", "list", "tender-create", "detail", "tender-edit"])
        self.assertEqual(pilot.expected_requests([2, 3], plan, 2), 29)

    def test_restart_requires_container_and_unique_names(self):
        base = dict(duration_seconds=1, think_seconds=0)
        with self.assertRaises(ValueError):
            pilot.phase_plan(base | {"phases": [{"name": "restart", "kind": "restart"}]})
        with self.assertRaises(ValueError):
            pilot.phase_plan(
                base
                | {
                    "phases": [
                        dict(name="x", kind="load", duration_seconds=1, think_seconds=0),
                        dict(name="x", kind="load", duration_seconds=1, think_seconds=0),
                    ]
                }
            )

    def test_planned_restart_gap_does_not_hide_same_phase_gap(self):
        rows = [
            dict(monotonic_seconds=0, phase="ordinary"),
            dict(expected_unavailable=True, phase="restart"),
            dict(monotonic_seconds=10, phase="recovery"),
        ]
        self.assertEqual(pilot.sampling_gaps(rows, [{}], []), [])
        rows.append(dict(monotonic_seconds=15, phase="recovery"))
        self.assertTrue(pilot.sampling_gaps(rows, [{}], []))


class ComparisonTests(unittest.TestCase):
    def test_restart_counter_reset_is_not_averaged_across_phase_boundary(self):
        samples = []
        for phase, start in [("ordinary", 0), ("recovery", 10)]:
            for offset in range(2):
                samples.append(
                    dict(
                        phase=phase,
                        monotonic_seconds=start + offset,
                        cpu_seconds=1 + offset,
                        rss_bytes=1024**2,
                        cpu_stat=dict(
                            nr_throttled=offset, nr_periods=offset, throttled_usec=offset
                        ),
                    )
                )
        record = dict(
            measurement_complete=True,
            measurement_gaps=[],
            app_runtime=dict(nano_cpus=1000000000, memory_bytes=1073741824),
            runtime_samples=samples,
            phase_results=[dict(name=n, kind="load") for n in ["ordinary", "recovery"]],
            database_samples=[dict(phase="ordinary", docker_stats={"CPUPerc": "1%"})],
            ordinary={},
            document_transfer={},
            complete=True,
            provisional_pass=True,
            config={},
            completed_iterations={},
        )
        with tempfile.TemporaryDirectory() as tmp:
            source, output = Path(tmp) / "input.json", Path(tmp) / "output.json"
            source.write_text(json.dumps(record))
            subprocess.run(
                [
                    sys.executable,
                    str(Path(__file__).with_name("compare_synthetic_limits.py")),
                    str(source),
                    "--output",
                    str(output),
                ],
                check=True,
                stdout=subprocess.DEVNULL,
            )
            measured = json.loads(output.read_text())["application_measured"]
            self.assertEqual(measured["cpu_mean_m"], 1000)
            self.assertEqual(measured["throttled_periods"], 2)


class LifecycleTests(unittest.TestCase):
    def test_restart_checks_new_loopback_binding(self):
        with (
            mock.patch.object(pilot.subprocess, "run"),
            mock.patch.object(pilot, "container_url", return_value="http://127.0.0.1:49002"),
            mock.patch.object(pilot, "wait_ready") as ready,
        ):
            self.assertEqual(pilot.restart_container("synthetic"), "http://127.0.0.1:49002")
            ready.assert_called_once_with("http://127.0.0.1:49002")

    def test_public_binding_is_rejected(self):
        with mock.patch.object(pilot.subprocess, "check_output", return_value="0.0.0.0:49002"):
            with self.assertRaises(ValueError):
                pilot.container_url("synthetic")

    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),
        )


class RemoteRuntimeTests(unittest.TestCase):
    def config(self):
        return SyntheticContractTests.config(None) | {
            "runtime": "remote",
            "phases": [
                {"name": "ordinary", "kind": "load", "duration_seconds": 1, "think_seconds": 0},
                {"name": "restart", "kind": "restart"},
            ],
        }

    def test_remote_runtime_accepts_restart_and_host_rejects_it(self):
        self.assertEqual(pilot.validate(self.config())["runtime"], "remote")
        with self.assertRaises(ValueError):
            pilot.validate(self.config() | {"runtime": "host"})

    def test_remote_mode_skips_local_sampler_coverage(self):
        self.assertEqual(pilot.sampling_gaps([], [], [], remote=True), [])
        self.assertTrue(pilot.sampling_gaps([], [], []))

    def test_settings_text_sets_allowed_hosts_and_disables_sso(self):
        text = pilot.settings_text("/m", "/i.db", ["vergabe-app"])
        self.assertIn("ALLOWED_HOSTS = ['vergabe-app']", text)
        self.assertIn("NETKINGDOM_ENABLED = False", text)

    def test_restart_deletes_single_ready_pod_and_waits_for_replacement(self):
        ready = {"type": "Ready", "status": "True"}

        def pod(name, conditions=(ready,), deleting=False):
            meta = {"name": name} | ({"deletionTimestamp": "x"} if deleting else {})
            return {"metadata": meta, "status": {"conditions": list(conditions)}}

        listings = [
            {"items": [pod("old")]},
            {"items": [pod("old", deleting=True), pod("new", conditions=())]},
            {"items": [pod("new")]},
        ]
        calls = []

        def request(method, path):
            calls.append((method, path))
            return listings.pop(0) if method == "GET" else {}

        with tempfile.TemporaryDirectory() as directory:
            Path(directory, "namespace").write_text("bench\n")
            with (
                mock.patch.object(pilot, "wait_ready") as waited,
                mock.patch.object(pilot.time, "sleep"),
            ):
                url = pilot.restart_kubernetes_pod(
                    "http://app", "app=x", sa_dir=Path(directory), request=request
                )
        self.assertEqual(url, "http://app")
        self.assertIn(("DELETE", "/api/v1/namespaces/bench/pods/old"), calls)
        waited.assert_called_once()

    def test_restart_refuses_ambiguous_selection(self):
        with tempfile.TemporaryDirectory() as directory:
            Path(directory, "namespace").write_text("bench")
            with self.assertRaises(RuntimeError):
                pilot.restart_kubernetes_pod(
                    "http://app",
                    "app=x",
                    sa_dir=Path(directory),
                    request=lambda method, path: {"items": []},
                )


if __name__ == "__main__":
    unittest.main()