radar_system/python_app/tests/test_orchestration_recovery.py

"""Orchestration & recovery tests.

Pins the agreed semantics:
  * crash auto-restart retries FOREVER with capped exponential back-off (never gives
    up — unattended appliance); the failure streak resets when data flows again;
  * config writes are atomic — a serialization failure leaves no partial output and
    never corrupts an existing config;
  * the acquisition producer command is selected by radar.model;
  * process exit reports classify clean vs unexpected exits correctly.
"""

from __future__ import annotations

import json
import os
import sys
import tempfile
import time
import unittest
from pathlib import Path

from python_app.models.run_config_model import RunConfigModel
from python_app.orchestration.config_writer import ConfigWriter
from python_app.orchestration.process_supervisor import ProcessExitReport, ProcessSupervisor
from python_app.orchestration.restart_policy import RestartPolicy


class RestartPolicyTest(unittest.TestCase):
    def test_backoff_grows_and_caps(self) -> None:
        policy = RestartPolicy(min_interval_s=3.0, max_interval_s=60.0, backoff_factor=2.0)
        self.assertEqual(policy.backoff_for(0), 3.0)
        self.assertEqual(policy.backoff_for(1), 6.0)
        self.assertEqual(policy.backoff_for(2), 12.0)
        self.assertEqual(policy.backoff_for(3), 24.0)
        self.assertEqual(policy.backoff_for(100), 60.0)  # capped

    def test_never_gives_up_even_after_huge_streak(self) -> None:
        # No attempt cap: a long failure streak still yields a finite, capped wait.
        policy = RestartPolicy()
        self.assertEqual(policy.backoff_for(10_000), policy.max_interval_s)

    def test_should_restart_respects_backoff_window(self) -> None:
        policy = RestartPolicy(min_interval_s=3.0, max_interval_s=60.0)
        # First failure (streak 0): needs >= 3s since last restart.
        self.assertFalse(policy.should_restart_now(now_s=102.0, last_restart_s=100.0, consecutive_failures=0))
        self.assertTrue(policy.should_restart_now(now_s=103.0, last_restart_s=100.0, consecutive_failures=0))
        # After 2 failures the window is 12s.
        self.assertFalse(policy.should_restart_now(now_s=111.0, last_restart_s=100.0, consecutive_failures=2))
        self.assertTrue(policy.should_restart_now(now_s=112.0, last_restart_s=100.0, consecutive_failures=2))

    def test_first_restart_is_immediate(self) -> None:
        # last_restart defaults to 0.0 in the mixin, so the very first crash restarts now.
        self.assertTrue(RestartPolicy().should_restart_now(now_s=5_000.0, last_restart_s=0.0, consecutive_failures=0))


class ConfigWriterTest(unittest.TestCase):
    def setUp(self) -> None:
        self._dir = tempfile.TemporaryDirectory()
        self.addCleanup(self._dir.cleanup)
        self.root = Path(self._dir.name)
        self.writer = ConfigWriter(self.root / "runtime")

    def test_writes_loadable_config_and_leaves_no_tmp(self) -> None:
        out = self.root / "run_config.json"
        self.writer.write(RunConfigModel(), out)
        self.assertTrue(out.exists())
        self.assertFalse(out.with_suffix(".json.tmp").exists())
        RunConfigModel.from_dict(json.loads(out.read_text()))  # round-trips back through the schema

    def test_nan_fails_loudly_without_corrupting_existing(self) -> None:
        out = self.root / "run_config.json"
        self.writer.write(RunConfigModel(), out)
        original = out.read_text()

        broken = RunConfigModel()
        broken.radar.sweep.if_bandwidth_hz = float("nan")
        with self.assertRaises(ValueError):  # allow_nan=False
            self.writer.write(broken, out)

        self.assertEqual(out.read_text(), original)  # existing config untouched
        self.assertFalse(out.with_suffix(".json.tmp").exists())  # no half-written tmp left


class RadarModelCommandTest(unittest.TestCase):
    def setUp(self) -> None:
        self._dir = tempfile.TemporaryDirectory()
        self.addCleanup(self._dir.cleanup)
        self.root = Path(self._dir.name)
        self.supervisor = ProcessSupervisor(self.root)

    def _config_with_model(self, model: str) -> Path:
        path = self.root / "cfg.json"
        path.write_text(json.dumps({"radar": {"model": model}}))
        return path

    def test_read_radar_model(self) -> None:
        self.assertEqual(ProcessSupervisor._read_radar_model(self._config_with_model("kamil_adc")), "kamil_adc")

    def test_read_radar_model_defaults_on_malformed(self) -> None:
        path = self.root / "bad.json"
        path.write_text("[]")  # not an object
        self.assertEqual(ProcessSupervisor._read_radar_model(path), "librevna")

    def test_matrix_producer_for_multi_and_sn9000(self) -> None:
        for model in ("librevna_multi", "sn9000"):
            cmd = self.supervisor._acquisition_command(self._config_with_model(model))
            self.assertEqual(cmd[:3], [sys.executable, "-m", "python_app.scripts.matrix_raw_producer"])

    def test_kamil_producer_for_kamil_adc(self) -> None:
        cmd = self.supervisor._acquisition_command(self._config_with_model("kamil_adc"))
        self.assertEqual(cmd[:3], [sys.executable, "-m", "python_app.scripts.kamil_adc_raw_producer"])

    def test_native_orchestrator_for_librevna(self) -> None:
        cmd = self.supervisor._acquisition_command(self._config_with_model("librevna"))
        self.assertTrue(cmd[0].endswith("build/bin/sweep_orchestrator"))


class ProcessExitReportTest(unittest.TestCase):
    @staticmethod
    def _report(*, code: int, clean: bool) -> ProcessExitReport:
        return ProcessExitReport(
            name="data_processor",
            command=["x"],
            working_directory=Path("/tmp"),
            return_code=code,
            stdout_path=Path("/nonexistent.out"),
            stderr_path=Path("/nonexistent.err"),
            expected_clean_exit=clean,
        )

    def test_clean_exit_is_info(self) -> None:
        report = self._report(code=0, clean=True)
        self.assertEqual(report.level, "INFO")
        self.assertIn("completed normally", report.format())

    def test_nonzero_exit_is_error(self) -> None:
        report = self._report(code=3, clean=False)
        self.assertEqual(report.level, "ERROR")
        self.assertIn("exited with code 3", report.format())

    def test_unexpected_zero_exit_is_error(self) -> None:
        report = self._report(code=0, clean=False)
        self.assertEqual(report.level, "ERROR")
        self.assertIn("exited unexpectedly with code 0", report.format())


class SupervisorLifecycleTest(unittest.TestCase):
    def setUp(self) -> None:
        self._dir = tempfile.TemporaryDirectory()
        self.addCleanup(self._dir.cleanup)
        self.supervisor = ProcessSupervisor(Path(self._dir.name))
        self.addCleanup(self.supervisor.stop_all)

    def _await_reports(self, timeout_s: float = 3.0) -> list[ProcessExitReport]:
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            reports = self.supervisor.collect_exit_reports()
            if reports:
                return reports
            time.sleep(0.02)
        return []

    def test_spawn_alive_then_stop(self) -> None:
        self.supervisor._spawn("data_preprocessor",
                               [sys.executable, "-c", "import time; time.sleep(30)"],
                               allow_clean_exit=False)
        self.assertTrue(self.supervisor.is_running())
        self.assertIn("data_preprocessor", self.supervisor.pids())
        self.supervisor.stop()
        self.assertFalse(self.supervisor.is_running())

    def test_unexpected_exit_reported_as_error(self) -> None:
        self.supervisor._spawn("data_processor",
                               [sys.executable, "-c", "import sys; sys.exit(3)"],
                               allow_clean_exit=False)
        reports = self._await_reports()
        self.assertEqual(len(reports), 1)
        self.assertEqual(reports[0].return_code, 3)
        self.assertEqual(reports[0].level, "ERROR")

    def test_clean_exit_reported_as_info(self) -> None:
        self.supervisor._spawn("sweep_orchestrator",
                               [sys.executable, "-c", "import sys; sys.exit(0)"],
                               allow_clean_exit=True)
        reports = self._await_reports()
        self.assertEqual(len(reports), 1)
        self.assertTrue(reports[0].expected_clean_exit)
        self.assertEqual(reports[0].level, "INFO")

    def test_stale_pid_guard_rejects_unrelated_processes(self) -> None:
        # The reap guard must never kill a recycled PID that is not one of our binaries.
        self.assertFalse(self.supervisor._is_stale_pipeline_pid(2_000_000_000))  # no such pid
        self.assertFalse(self.supervisor._is_stale_pipeline_pid(os.getpid()))    # the test runner


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