252 lines
12 KiB
Python
252 lines
12 KiB
Python
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 "policies/managed" in command and "-name '*.json'" in command:
|
|
return 0, "", ""
|
|
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_personal_policy_mode",
|
|
"neko_1_tm_payload",
|
|
"neko_1_tm_storage",
|
|
"android_devices",
|
|
"android_screenshot",
|
|
"android_action",
|
|
):
|
|
self.assertIn(check_id, ids)
|
|
|
|
def test_neko_personal_policy_mode_rejects_managed_policy_json(self) -> None:
|
|
checks: list[dict[str, object]] = []
|
|
|
|
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 "policies/managed" in command and "-name '*.json'" in command:
|
|
return 0, "/etc/opt/chrome/policies/managed/policies.json\n", ""
|
|
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 1, "", ""
|
|
self.fail(f"unexpected docker exec: {container} {command}")
|
|
|
|
with patch.object(verify_local_install, "docker_ps_names", return_value=(True, ["live-neko-1", "live-neko-2"], "")):
|
|
with patch.object(verify_local_install, "systemd_active", return_value=(True, "active")):
|
|
with patch.object(verify_local_install, "docker_exec", side_effect=fake_docker_exec):
|
|
verify_local_install.verify_neko(
|
|
checks,
|
|
{"ENABLE_NEKO": "1", "NEKO_INSTANCE_COUNT": "2", "NEKO_MANAGED_POLICIES": "0"},
|
|
timeout=5.0,
|
|
)
|
|
|
|
report = verify_local_install.summarize(checks)
|
|
self.assertFalse(report["ready"])
|
|
self.assertEqual(report["failing_checks"][0]["id"], "neko_1_personal_policy_mode")
|
|
tm_storage = next(item for item in checks if item["id"] == "neko_1_tm_storage")
|
|
self.assertTrue(tm_storage["required"])
|
|
|
|
def test_neko_managed_policy_mode_allows_policy_json_and_requires_tm_storage(self) -> None:
|
|
checks: list[dict[str, object]] = []
|
|
|
|
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 "policies/managed" in command and "-name '*.json'" in command:
|
|
return 0, "/etc/opt/chrome/policies/managed/policies.json\n", ""
|
|
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 1, "", ""
|
|
self.fail(f"unexpected docker exec: {container} {command}")
|
|
|
|
with patch.object(verify_local_install, "docker_ps_names", return_value=(True, ["live-neko-1", "live-neko-2"], "")):
|
|
with patch.object(verify_local_install, "systemd_active", return_value=(True, "active")):
|
|
with patch.object(verify_local_install, "docker_exec", side_effect=fake_docker_exec):
|
|
verify_local_install.verify_neko(
|
|
checks,
|
|
{"ENABLE_NEKO": "1", "NEKO_INSTANCE_COUNT": "2", "NEKO_MANAGED_POLICIES": "1"},
|
|
timeout=5.0,
|
|
)
|
|
|
|
report = verify_local_install.summarize(checks)
|
|
self.assertFalse(report["ready"])
|
|
self.assertEqual(report["failing_checks"][0]["id"], "neko_1_tm_storage")
|
|
|
|
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()
|