"""U-036 tests: Deploy Supervisor process skeleton -- lifecycle only (PID
file, status file, start/stop), no actual deploy logic (that's U-037+).
GRAPHLEMON_HOME is always overridden to a throwaway tempdir so these tests
never touch the real user's ~/.graphlemon.

Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v
"""

import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import unittest

from supervisor import deploy_supervisor as ds
from src import paths


class DeploySupervisorTestBase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.mkdtemp(prefix="deploy-supervisor-test-")
        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
        self.old_home = os.environ.get("GRAPHLEMON_HOME")
        os.environ["GRAPHLEMON_HOME"] = self.tmp

    def tearDown(self):
        if self.old_home is None:
            os.environ.pop("GRAPHLEMON_HOME", None)
        else:
            os.environ["GRAPHLEMON_HOME"] = self.old_home


class RuntimePathsTest(DeploySupervisorTestBase):
    def test_runtime_dir_created_under_platform_home(self):
        d = ds.runtime_dir()
        self.assertTrue(os.path.isdir(d))
        self.assertTrue(d.startswith(paths.platform_home()))

    def test_pid_and_status_file_paths_live_under_runtime_dir(self):
        self.assertEqual(os.path.dirname(ds.pid_file_path()), ds.runtime_dir())
        self.assertEqual(os.path.dirname(ds.status_file_path()), ds.runtime_dir())


class IsAliveTest(unittest.TestCase):
    def test_current_process_is_alive(self):
        self.assertTrue(ds.is_alive(os.getpid()))

    def test_zero_or_negative_pid_is_not_alive(self):
        self.assertFalse(ds.is_alive(0))
        self.assertFalse(ds.is_alive(-1))

    def test_non_int_is_not_alive(self):
        self.assertFalse(ds.is_alive("not-a-pid"))

    def test_a_pid_that_does_not_exist_is_not_alive(self):
        # PID_MAX on Linux/macOS is well below this; astronomically unlikely
        # to collide with a real running process.
        self.assertFalse(ds.is_alive(2**30))


class StatusFileTest(DeploySupervisorTestBase):
    def test_write_then_read_status_round_trips(self):
        ds.write_status("running", detail="test detail")
        status = ds.read_status()
        self.assertEqual(status["status"], "running")
        self.assertEqual(status["detail"], "test detail")
        self.assertEqual(status["pid"], os.getpid())

    def test_read_status_returns_none_when_no_file_yet(self):
        self.assertIsNone(ds.read_status())

    def test_write_status_is_atomic_no_tmp_file_left_behind(self):
        ds.write_status("running")
        entries = os.listdir(ds.runtime_dir())
        self.assertEqual(entries, ["supervisor.status.json"])


class IsRunningTest(DeploySupervisorTestBase):
    def test_false_when_no_pid_file(self):
        self.assertFalse(ds.is_running())

    def test_true_when_pid_file_points_to_a_live_process(self):
        with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
            handle.write(str(os.getpid()))
        self.assertTrue(ds.is_running())

    def test_false_when_pid_file_points_to_a_dead_process(self):
        with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
            handle.write(str(2**30))
        self.assertFalse(ds.is_running())

    def test_false_when_pid_file_is_corrupt(self):
        with open(ds.pid_file_path(), "w", encoding="utf-8") as handle:
            handle.write("not-a-number")
        self.assertFalse(ds.is_running())


class RunForegroundTest(DeploySupervisorTestBase):
    def test_writes_pid_file_and_running_status_during_the_loop(self):
        seen_running = []

        def poll_fn():
            seen_running.append(ds.is_running())
            seen_running.append(ds.read_status()["status"])

        ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1)

        self.assertEqual(seen_running, [True, "running"])

    def test_calls_poll_fn_exactly_iterations_times(self):
        calls = []
        ds.run_foreground(poll_fn=lambda: calls.append(1), poll_interval=0, iterations=3)
        self.assertEqual(len(calls), 3)

    def test_writes_stopped_status_and_removes_pid_file_on_normal_completion(self):
        ds.run_foreground(poll_fn=None, poll_interval=0, iterations=1)
        self.assertEqual(ds.read_status()["status"], "stopped")
        self.assertFalse(os.path.exists(ds.pid_file_path()))

    def test_sigterm_breaks_the_loop_and_still_writes_stopped_status(self):
        calls = []

        def poll_fn():
            calls.append(1)
            if len(calls) == 2:
                os.kill(os.getpid(), signal.SIGTERM)

        # iterations=None (would run forever) -- SIGTERM is the only exit.
        ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=None)

        self.assertEqual(len(calls), 2)
        self.assertEqual(ds.read_status()["status"], "stopped")
        self.assertFalse(os.path.exists(ds.pid_file_path()))

    def test_poll_fn_exception_propagates_but_cleanup_still_runs(self):
        def poll_fn():
            raise RuntimeError("boom")

        with self.assertRaises(RuntimeError):
            ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1)

        self.assertEqual(ds.read_status()["status"], "stopped")
        self.assertFalse(os.path.exists(ds.pid_file_path()))


class RealSubprocessLifecycleTest(DeploySupervisorTestBase):
    """Not mocked -- actually spawns `python3 -m supervisor.deploy_supervisor`
    as a real independent OS process and stops it for real, per this Unit's
    Done criterion ("骨架能独立启停")."""

    def test_start_detached_then_stop(self):
        env = dict(os.environ)
        # start_detached itself calls is_running()/Popen using the current
        # process's env var (already patched by setUp); the child process
        # inherits it via Popen's default env=None -> os.environ passthrough.
        pid = ds.start_detached(poll_interval=0.1)
        try:
            deadline = time.time() + 5
            while time.time() < deadline and not ds.is_running():
                time.sleep(0.05)
            self.assertTrue(ds.is_running())
            self.assertTrue(ds.is_alive(pid))

            stopped = ds.stop(timeout=5)
            self.assertTrue(stopped)
            self.assertFalse(ds.is_running())
        finally:
            if ds.is_alive(pid):
                os.kill(pid, signal.SIGKILL)

    def test_start_detached_twice_raises_while_already_running(self):
        pid = ds.start_detached(poll_interval=0.1)
        try:
            deadline = time.time() + 5
            while time.time() < deadline and not ds.is_running():
                time.sleep(0.05)
            self.assertTrue(ds.is_running())
            with self.assertRaises(RuntimeError):
                ds.start_detached(poll_interval=0.1)
        finally:
            ds.stop(timeout=5)
            if ds.is_alive(pid):
                os.kill(pid, signal.SIGKILL)

    def test_stop_returns_false_when_nothing_is_running(self):
        self.assertFalse(ds.stop(timeout=1))


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