path: root/app/tests/test_backends_servers.py

"""Tests for the server lifecycle module (backends/servers.py)."""

import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch

from backends import ServerSpec, servers


class StartTests(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.dir = Path(self._tmp.name)
        # A fake executable so Path(argv[0]).exists() passes.
        self.exe = self.dir / "fake_server"
        self.exe.write_bytes(b"#!/bin/sh\n")
        self.spec = ServerSpec("test", "http://127.0.0.1:9999",
                               [str(self.exe), "--port", "9999"])

    def tearDown(self):
        self._tmp.cleanup()

    def test_returns_false_when_executable_missing(self):
        spec = ServerSpec("nope", "http://127.0.0.1:1", ["/no/such/binary"])
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertFalse(servers.start(spec))

    def test_noop_when_already_running(self):
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("backends.common.server_running", return_value=True), \
                patch("subprocess.Popen") as mk:
            self.assertTrue(servers.start(self.spec))
        mk.assert_not_called()

    def test_happy_path_spawns_and_polls_until_ready(self):
        proc = MagicMock()
        proc.pid = 4242
        proc.poll.return_value = None  # process still running
        # server_running: False on the pre-check, True once inside the loop.
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc) as mk, \
                patch("backends.common.server_running",
                      side_effect=[False, True]), \
                patch("time.sleep"):
            ok = servers.start(self.spec)
        self.assertTrue(ok)
        mk.assert_called_once()
        # Pid file written.
        self.assertEqual(
            (self.dir / "test-server.pid").read_text(encoding="utf-8"),
            "4242")

    def test_returns_false_when_process_exits_early(self):
        proc = MagicMock()
        proc.pid = 99
        proc.poll.return_value = 1  # exited with code 1
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc), \
                patch("backends.common.server_running", return_value=False), \
                patch("time.sleep"):
            ok = servers.start(self.spec)
        self.assertFalse(ok)
        # Pid file cleaned up after early exit.
        self.assertFalse((self.dir / "test-server.pid").exists())

    def test_returns_false_on_timeout(self):
        proc = MagicMock()
        proc.pid = 7
        proc.poll.return_value = None
        # time.time: first call < deadline loop entry, then past deadline.
        times = iter([0.0, float(servers.SERVER_START_TIMEOUT + 1)])
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc), \
                patch("backends.common.server_running", return_value=False), \
                patch("time.sleep"), \
                patch("time.time", side_effect=lambda: next(times)):
            ok = servers.start(self.spec)
        self.assertFalse(ok)

    def _boot_proc(self):
        proc = MagicMock()
        proc.pid = 4242
        proc.poll.return_value = None
        return proc

    def test_cwd_passed_to_popen(self):
        """A spec with a cwd spawns the server in that working directory.

        audio.cpp resolves model_specs/<family>.json relative to its
        process working directory, so the hub must start it from the
        checkout.
        """
        spec = ServerSpec("test", "http://127.0.0.1:9999",
                          [str(self.exe)], cwd=Path("/opt/audio.cpp"))
        proc = self._boot_proc()
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc) as mk, \
                patch("backends.common.server_running",
                      side_effect=[False, True]), \
                patch("time.sleep"):
            ok = servers.start(spec)
        self.assertTrue(ok)
        kwargs = mk.call_args.kwargs
        self.assertEqual(kwargs.get("cwd"), "/opt/audio.cpp")

    def test_identity_spec_waits_for_http_identity(self):
        """Readiness needs the server to answer HTTP as its identity.

        A TCP-accepting but still-booting server (lazy model load, slow
        listen-before-serve) must not count as ready.
        """
        spec = ServerSpec("test", "http://127.0.0.1:9999",
                          [str(self.exe)], identity="audiocpp")
        proc = self._boot_proc()
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc), \
                patch("backends.common.server_running", return_value=True), \
                patch.object(servers.probe, "identify_server",
                             side_effect=[None, None, "audiocpp"]), \
                patch("time.sleep"):
            ok = servers.start(spec)
        self.assertTrue(ok)

    def test_faster_identity_requires_model_loaded(self):
        """The faster identity additionally waits for /health model_loaded."""
        spec = ServerSpec("test", "http://127.0.0.1:9999",
                          [str(self.exe)], identity="faster")
        proc = self._boot_proc()
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc), \
                patch("backends.common.server_running", return_value=True), \
                patch.object(servers.probe, "identify_server",
                             return_value="faster"), \
                patch.object(servers.probe, "faster_model_loaded",
                             side_effect=[False, True]), \
                patch("time.sleep"):
            ok = servers.start(spec)
        self.assertTrue(ok)

    def test_progress_receives_boot_events(self):
        events = []
        proc = self._boot_proc()
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc), \
                patch("backends.common.server_running",
                      side_effect=[False, True]), \
                patch("time.sleep"):
            ok = servers.start(self.spec, progress=events.append)
        self.assertTrue(ok)
        kinds = [event["kind"] for event in events]
        self.assertEqual(kinds, ["starting", "ready"])
        self.assertEqual(events[0]["pid"], 4242)
        self.assertIn("--port", events[0]["argv"])

    def test_cancel_aborts_boot_kills_process_and_reports(self):
        import threading
        cancel = threading.Event()
        cancel.set()
        proc = self._boot_proc()
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch("subprocess.Popen", return_value=proc), \
                patch("backends.common.server_running", return_value=False), \
                patch.object(servers, "_kill_pid", return_value=True) as mk, \
                patch("time.sleep"):
            ok = servers.start(self.spec, cancel=cancel)
        self.assertFalse(ok)
        mk.assert_called_once_with(4242)
        self.assertFalse((self.dir / "test-server.pid").exists())

    def test_console_progress_prints_events(self):
        import io
        from contextlib import redirect_stdout
        buf = io.StringIO()
        with redirect_stdout(buf):
            servers._console_progress({"kind": "running", "name": "test",
                                       "url": "http://127.0.0.1:9999"})
            servers._console_progress({"kind": "error", "message": "boom"})
        out = buf.getvalue()
        self.assertIn("already running", out)
        self.assertIn("boom", out)


class StopTests(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.dir = Path(self._tmp.name)

    def tearDown(self):
        self._tmp.cleanup()

    def _write_pid(self, name, pid):
        (self.dir / f"{name}-server.pid").write_text(str(pid),
                                                     encoding="utf-8")

    def test_returns_false_when_no_pid_file(self):
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertFalse(servers.stop("test"))

    def test_stops_alive_process_and_removes_pid_file(self):
        self._write_pid("test", 1234)
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch.object(servers, "_pid_alive", return_value=True), \
                patch.object(servers, "_kill_pid", return_value=True) as mk:
            ok = servers.stop("test")
        self.assertTrue(ok)
        mk.assert_called_once_with(1234)
        self.assertFalse((self.dir / "test-server.pid").exists())

    def test_already_dead_returns_true_and_cleans_pid_file(self):
        self._write_pid("test", 1234)
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch.object(servers, "_pid_alive", return_value=False), \
                patch.object(servers, "_kill_pid") as mk:
            ok = servers.stop("test")
        self.assertTrue(ok)
        mk.assert_not_called()
        self.assertFalse((self.dir / "test-server.pid").exists())

    def test_corrupt_pid_file_returns_false_and_cleans(self):
        (self.dir / "test-server.pid").write_text("not-a-number",
                                                   encoding="utf-8")
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertFalse(servers.stop("test"))
        self.assertFalse((self.dir / "test-server.pid").exists())


class ManagesTests(unittest.TestCase):
    """manages(): a live recorded pid marks a server as ours."""

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.dir = Path(self._tmp.name)
        self.specs = [ServerSpec("test", "http://127.0.0.1:9999", [])]

    def tearDown(self):
        self._tmp.cleanup()

    def _write_pid(self, name, pid):
        (self.dir / f"{name}-server.pid").write_text(str(pid),
                                                     encoding="utf-8")

    def test_false_without_pid_file(self):
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertFalse(servers.manages(self.specs))

    def test_true_with_live_recorded_pid(self):
        self._write_pid("test", 4242)
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch.object(servers, "_pid_alive", return_value=True):
            self.assertTrue(servers.manages(self.specs))

    def test_false_with_dead_recorded_pid(self):
        self._write_pid("test", 4242)
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch.object(servers, "_pid_alive", return_value=False):
            self.assertFalse(servers.manages(self.specs))

    def test_false_with_corrupt_pid_file(self):
        (self.dir / "test-server.pid").write_text("junk",
                                                  encoding="utf-8")
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch.object(servers, "_pid_alive",
                             return_value=True) as mk_alive:
            self.assertFalse(servers.manages(self.specs))
        mk_alive.assert_not_called()

    def test_true_when_any_spec_is_ours(self):
        other = ServerSpec("other", "http://127.0.0.1:9998", [])
        with patch.object(servers, "LOG_DIR", self.dir), \
                patch.object(servers, "_pid_alive", return_value=True):
            self._write_pid("test", 4242)
            self.assertTrue(servers.manages([other] + self.specs))
        # The live pid belongs to 'test'; 'other' alone stays unmanaged.
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertFalse(servers.manages([other]))


class PidForTests(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.dir = Path(self._tmp.name)

    def tearDown(self):
        self._tmp.cleanup()

    def test_none_when_no_pid_file(self):
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertIsNone(servers.pid_for("test"))

    def test_returns_pid_from_file(self):
        (self.dir / "test-server.pid").write_text("555\n",
                                                   encoding="utf-8")
        with patch.object(servers, "LOG_DIR", self.dir):
            self.assertEqual(servers.pid_for("test"), 555)


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