from __future__ import annotations import argparse import importlib.util import json import stat import tempfile import unittest from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location( "verify_local_install", ROOT / "tools" / "verify_local_install.py", ) assert SPEC and SPEC.loader verify_local_install = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(verify_local_install) class VerifyLocalInstallTests(unittest.TestCase): def make_install_dir(self, env_text: str) -> Path: tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) install_dir = Path(tmp.name) env_path = install_dir / "config" / "system-stack.env" env_path.parent.mkdir(parents=True) env_path.write_text(env_text, encoding="utf-8") web_index = install_dir / "web-console" / "out" / "index.html" web_index.parent.mkdir(parents=True) web_index.write_text("ok", encoding="utf-8") start = install_dir / "start.sh" start.write_text("#!/usr/bin/env bash\n", encoding="utf-8") start.chmod(start.stat().st_mode | stat.S_IXUSR) return install_dir def test_build_report_covers_control_plane_neko_and_android_screenshot(self) -> None: install_dir = self.make_install_dir( "\n".join( [ "PORT=8100", "HOSTNAME_ALIAS=live", "ENABLE_LIVE_EDGE_PROXY=1", "ENABLE_NEKO=1", "NEKO_INSTANCE_COUNT=2", "ENABLE_ANDROID_PANEL=1", "ENABLE_REDROID=1", "", ] ) ) def fake_http_json(url: str, *, timeout: float = 5.0): if url.endswith("/health"): return True, 200, {"status": "ok"}, "", 2.0 if url.endswith("/deploy/check"): return True, 200, {"ready": True, "pass_count": 7, "total_count": 7}, "", 3.0 if url.endswith("/android/devices"): return ( True, 200, { "available": True, "devices": [{"serial": "127.0.0.1:5555", "state": "device"}], }, "", 4.0, ) if url.endswith("/youtube_pro_channels"): return True, 200, {"channels": [{"id": "lane1"}, {"id": "lane2"}]}, "", 2.0 self.fail(f"unexpected JSON URL: {url}") def fake_http_bytes(url: str, *, timeout: float = 5.0): if url == "http://127.0.0.1/health": return True, 200, b"ok", "", 2.0 if url.startswith("http://127.0.0.1:8100/android/screenshot?"): return True, 200, verify_local_install.PNG_SIG + b"image", "", 6.0 self.fail(f"unexpected bytes URL: {url}") def fake_http_json_post(url: str, payload: dict, *, timeout: float = 5.0): if url == "http://127.0.0.1:8100/android/action": self.assertEqual(payload["action"], "wake") self.assertEqual(payload["serial"], "127.0.0.1:5555") return True, 200, {"message": "Wake + swipe + home sent"}, "", 7.0 self.fail(f"unexpected POST URL: {url}") def fake_docker_exec(container: str, command: str, *, timeout: float = 8.0): if "supervisorctl status google-chrome" in command: return 0, "google-chrome RUNNING pid 8", "" if "supervisorctl status tampermonkey-provisioning" in command: return 0, "tampermonkey-provisioning RUNNING pid 9", "" if "tampermonkey-provisioning.json" in command: return 0, "", "" if "grep -R -a -l" in command: return 0, "/home/neko/.config/google-chrome/Default/Local Extension Settings", "" self.fail(f"unexpected docker exec: {container} {command}") args = argparse.Namespace( install_dir=str(install_dir), port="", timeout=5.0, skip_neko=False, skip_android=False, ) with patch.object(verify_local_install, "systemd_active", return_value=(True, "active")): with patch.object(verify_local_install, "http_json", side_effect=fake_http_json): with patch.object(verify_local_install, "http_bytes", side_effect=fake_http_bytes): with patch.object(verify_local_install, "http_json_post", side_effect=fake_http_json_post): with patch.object( verify_local_install, "docker_ps_names", return_value=(True, ["live-neko-1", "live-neko-2"], ""), ): with patch.object(verify_local_install, "docker_exec", side_effect=fake_docker_exec): report = verify_local_install.build_report(args) self.assertTrue(report["ready"]) ids = {item["id"] for item in report["checks"]} for check_id in ( "api_health", "deploy_check", "edge_health", "live_edge_watch_service", "neko_ip_watch_service", "youtube_pro_neko_capacity", "neko_1_browser", "neko_1_tm_payload", "neko_1_tm_storage", "android_devices", "android_screenshot", "android_action", ): self.assertIn(check_id, ids) def test_android_endpoint_without_devices_is_not_a_required_install_failure(self) -> None: checks: list[dict[str, object]] = [] with patch.object( verify_local_install, "http_json", return_value=(True, 200, {"available": False, "devices": []}, "", 3.0), ): verify_local_install.verify_android( checks, {"ENABLE_ANDROID_PANEL": "1", "ENABLE_REDROID": "1"}, port="8001", timeout=5.0, ) report = verify_local_install.summarize(checks) self.assertTrue(report["ready"], json.dumps(report, ensure_ascii=False)) self.assertFalse(next(item for item in checks if item["id"] == "android_device_presence")["required"]) def test_required_failure_returns_degraded_summary(self) -> None: checks: list[dict[str, object]] = [] verify_local_install.check(checks, "api_health", "FastAPI /health", False, detail="connection refused") report = verify_local_install.summarize(checks) self.assertFalse(report["ready"]) self.assertEqual(report["failing_checks"][0]["id"], "api_health") def test_youtube_pro_capacity_fails_when_channels_outgrow_neko_count(self) -> None: checks: list[dict[str, object]] = [] with patch.object( verify_local_install, "http_json", return_value=(True, 200, {"channels": [{"id": "a"}, {"id": "b"}, {"id": "c"}]}, "", 1.0), ): verify_local_install.verify_youtube_pro_capacity( checks, {"ENABLE_NEKO": "1", "NEKO_INSTANCE_COUNT": "2"}, port="8001", timeout=5.0, ) report = verify_local_install.summarize(checks) self.assertFalse(report["ready"]) self.assertEqual(report["failing_checks"][0]["id"], "youtube_pro_neko_capacity") def test_neko_required_count_matches_current_three_instance_compose_cap(self) -> None: self.assertEqual(verify_local_install.neko_required_count({"NEKO_INSTANCE_COUNT": "1"}), 2) self.assertEqual(verify_local_install.neko_required_count({"NEKO_INSTANCE_COUNT": "2"}), 2) self.assertEqual(verify_local_install.neko_required_count({"NEKO_INSTANCE_COUNT": "4"}), 4) self.assertEqual(verify_local_install.neko_required_count({"NEKO_INSTANCE_COUNT": "99"}), 64) if __name__ == "__main__": unittest.main()