from __future__ import annotations import asyncio import tempfile import unittest from pathlib import Path from unittest.mock import AsyncMock, patch from src.web_process_backend import ( WebProcessBackend, _ffmpeg_pid_matches_process, build_pm2_start_command, pm2_process_entry_needs_recreate, preferred_python_interpreter, ) class WebProcessBackendTests(unittest.TestCase): def test_build_pm2_start_command_for_python_script_uses_repo_context(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) script = base_dir / "youtube.py" venv_python = base_dir / ".venv" / "bin" / "python" venv_python.parent.mkdir(parents=True, exist_ok=True) venv_python.write_text("#!/usr/bin/env python3\n", encoding="utf-8") venv_python.chmod(0o755) script.write_text("print('ok')\n", encoding="utf-8") cmd = build_pm2_start_command(base_dir, "youtube2__lane1", "youtube.py") self.assertIsNotNone(cmd) assert cmd is not None self.assertIn("LIVE_PM2_PROCESS_NAME=youtube2__lane1", cmd) self.assertIn("--name youtube2__lane1", cmd) self.assertIn("--cwd", cmd) self.assertIn(str(script), cmd) self.assertIn(str(venv_python), cmd) def test_preferred_python_interpreter_falls_back_to_system_python_when_venv_is_missing(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) with patch("src.web_process_backend.shutil_which", side_effect=lambda name: f"/usr/bin/{name}" if name == "python3" else None): with patch("src.web_process_backend.sys.executable", "/definitely/missing/python"): chosen = preferred_python_interpreter(base_dir) self.assertEqual(chosen, "/usr/bin/python3") def test_pm2_process_entry_needs_recreate_when_python_interpreter_is_missing(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) script = base_dir / "youtube.py" script.write_text("print('ok')\n", encoding="utf-8") item = { "pm_exec_path": str(script), "pm_cwd": str(base_dir), "pm2_env": { "exec_interpreter": str(base_dir / ".venv" / "bin" / "python"), }, } self.assertTrue(pm2_process_entry_needs_recreate(base_dir, "youtube", "youtube.py", item)) def test_start_recreates_stale_pm2_entry_instead_of_reusing_broken_interpreter(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) script = base_dir / "youtube.py" script.write_text("print('ok')\n", encoding="utf-8") backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend._use_pm2 = True stale_item = { "pm_exec_path": str(script), "pm_cwd": str(base_dir), "pm2_env": { "exec_interpreter": str(base_dir / ".venv" / "bin" / "python"), }, } with patch.object(backend, "ensure_pm2_runtime", AsyncMock()): with patch.object(backend, "_pm2_jlist_index", AsyncMock(return_value={"youtube": stale_item})): with patch.object(backend, "clear_process_logs", AsyncMock()): with patch("src.web_process_backend.build_pm2_start_command", return_value="pm2 start fresh") as build_cmd: with patch("src.web_process_backend.run_pm2_result", new=AsyncMock(return_value=(0, "deleted"))) as run_pm2_result: with patch("src.web_process_backend._shell_out", new=AsyncMock(return_value=(0, "created"))) as shell_out: with patch.object(backend, "_set_pm2_desired_running", AsyncMock()) as set_desired: with patch.object(backend, "_pm2_save", AsyncMock()): text = asyncio.run(backend.start("youtube", "youtube.py")) self.assertEqual(text, "created") build_cmd.assert_called_once_with(base_dir, "youtube", "youtube.py") run_pm2_result.assert_awaited_once_with("delete youtube", timeout=20.0) shell_out.assert_awaited_once_with("pm2 start fresh", timeout=45.0) set_desired.assert_awaited_once_with("youtube", True) def test_parse_pm2_describe_marks_missing_jlist_process_as_not_found(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend.process_infos = AsyncMock(return_value={"youtube": {"process_status": "not_found", "out_path": None, "err_path": None, "pid": 0}}) # type: ignore[method-assign] info = asyncio.run(backend.parse_pm2_describe("youtube")) self.assertEqual(info["process_status"], "not_found") self.assertIsNone(info["out_path"]) self.assertIsNone(info["err_path"]) self.assertEqual(info["pid"], 0) def test_parse_pm2_describe_reads_status_and_log_paths_from_jlist_item(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend.process_infos = AsyncMock( # type: ignore[method-assign] return_value={ "youtube": { "process_status": "online", "out_path": "/home/live/.pm2/logs/youtube-out.log", "err_path": "/home/live/.pm2/logs/youtube-error.log", "pid": 4567, } } ) info = asyncio.run(backend.parse_pm2_describe("youtube")) self.assertEqual(info["process_status"], "online") self.assertEqual(info["out_path"], "/home/live/.pm2/logs/youtube-out.log") self.assertEqual(info["err_path"], "/home/live/.pm2/logs/youtube-error.log") self.assertEqual(info["pid"], 4567) def test_pm2_jlist_items_uses_short_ttl_cache(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") shell_out = AsyncMock(return_value=(0, '[{"name":"youtube","pm2_env":{"status":"online"}}]')) with patch("src.web_process_backend.resolve_pm2_bin", return_value="/usr/bin/pm2"): with patch("src.web_process_backend._shell_out", new=shell_out): first = asyncio.run(backend._pm2_jlist_items()) second = asyncio.run(backend._pm2_jlist_items()) self.assertEqual(first[0]["name"], "youtube") self.assertEqual(second[0]["name"], "youtube") shell_out.assert_awaited_once() def test_ensure_pm2_runtime_resurrects_pm2_when_dump_exists_and_jlist_is_empty(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) pm2_home = base_dir / ".pm2-home" pm2_home.mkdir(parents=True, exist_ok=True) (pm2_home / "pm2.pid").write_text("1234\n", encoding="utf-8") (pm2_home / "dump.pm2").write_text('[{"name":"youtube"}]\n', encoding="utf-8") backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend._use_pm2 = True shell_out = AsyncMock(return_value=(0, "[]")) resurrect = AsyncMock(return_value=(0, "ok")) with patch.dict("os.environ", {"PM2_HOME": str(pm2_home)}, clear=False): with patch("src.web_process_backend.resolve_pm2_bin", return_value="/usr/bin/pm2"): with patch("src.web_process_backend._shell_out", new=shell_out): with patch("src.web_process_backend.run_pm2_result", new=resurrect): asyncio.run(backend.ensure_pm2_runtime()) resurrect.assert_awaited_once_with("resurrect", timeout=45.0) def test_ensure_pm2_runtime_skips_resurrect_without_saved_dump(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) pm2_home = base_dir / ".pm2-home" pm2_home.mkdir(parents=True, exist_ok=True) (pm2_home / "pm2.pid").write_text("1234\n", encoding="utf-8") backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend._use_pm2 = True shell_out = AsyncMock(return_value=(0, "[]")) resurrect = AsyncMock(return_value=(0, "ok")) with patch.dict("os.environ", {"PM2_HOME": str(pm2_home)}, clear=False): with patch("src.web_process_backend.resolve_pm2_bin", return_value="/usr/bin/pm2"): with patch("src.web_process_backend._shell_out", new=shell_out): with patch("src.web_process_backend.run_pm2_result", new=resurrect): asyncio.run(backend.ensure_pm2_runtime()) resurrect.assert_not_awaited() def test_start_does_not_mark_desired_running_when_shell_command_fails(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) script = base_dir / "youtube.py" script.write_text("print('ok')\n", encoding="utf-8") backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend._use_pm2 = True with patch.object(backend, "ensure_mode", AsyncMock()): with patch.object(backend, "_pm2_registered_names", AsyncMock(return_value=set())): with patch.object(backend, "clear_process_logs", AsyncMock()): with patch("src.web_process_backend.build_pm2_start_command", return_value="pm2 start ..."): with patch("src.web_process_backend._shell_out", new=AsyncMock(return_value=(1, "boom"))): with patch.object(backend, "_set_pm2_desired_running", AsyncMock()) as set_desired: with patch.object(backend, "_pm2_save", AsyncMock()): text = asyncio.run(backend.start("youtube", "youtube.py")) self.assertEqual(text, "boom") set_desired.assert_not_awaited() def test_live_ffmpeg_pids_uses_process_tree_of_target_process(self) -> None: with tempfile.TemporaryDirectory() as tmp: base_dir = Path(tmp) backend = WebProcessBackend(base_dir, base_dir / ".pm2" / "logs") backend.ensure_mode = AsyncMock() backend.process_infos = AsyncMock( # type: ignore[method-assign] return_value={"youtube": {"pid": 4321, "process_status": "online"}} ) with patch("src.web_process_backend.os.getpgid", return_value=6789): with patch("src.web_process_backend._linux_ffmpeg_pids_for_process_tree", return_value=[9876]) as finder: pids = asyncio.run(backend.live_ffmpeg_pids("youtube")) self.assertEqual(pids, [9876]) finder.assert_called_once_with(4321, root_pgid=6789) def test_ffmpeg_pid_matches_process_uses_exact_inherited_env_name(self) -> None: env_map = { "LIVE_PM2_PROCESS_NAME": "youtube2__lane1", "PM2_SERVICE_NAME": "", } def fake_env(pid: int, key: str) -> str: return env_map.get(key, "") with patch("src.web_process_backend._read_proc_env_value", side_effect=fake_env): self.assertTrue(_ffmpeg_pid_matches_process(1234, "youtube2__lane1")) self.assertFalse(_ffmpeg_pid_matches_process(1234, "youtube")) if __name__ == "__main__": unittest.main()