Files
gitlab-instance-0a899031_sh/d2ypp2/tests/test_neko_config.py
eric 1d89aa29e0 s
2026-05-19 12:52:04 +00:00

332 lines
19 KiB
Python

from __future__ import annotations
import base64
import importlib.util
import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
class NekoConfigTests(unittest.TestCase):
def test_browser_launch_cleanup_script_still_prunes_profile_bloat(self) -> None:
script_path = ROOT / "services" / "neko" / "browser-launch.sh"
self.assertTrue(os.access(script_path, os.X_OK))
script = script_path.read_text(encoding="utf-8")
self.assertIn("arch=\"$(uname -m", script)
self.assertIn("--disable-software-rasterizer", script)
self.assertIn("--use-gl=swiftshader", script)
self.assertIn("NEKO_BROWSER_RENDER_MODE", script)
self.assertIn(".local/share/icons/hicolor", script)
self.assertIn("--start-maximized", script)
self.assertIn("--disable-background-mode", script)
self.assertNotIn("--disable-background-networking", script)
self.assertNotIn("--disable-sync", script)
self.assertNotIn("--disable-extensions", script)
self.assertIn('ui["developer_mode"] = True', script)
self.assertIn('"Local State"', script)
self.assertIn('browser_class="google-chrome"', script)
self.assertIn('xdotool search --onlyvisible --class "$browser_class"', script)
self.assertIn('prune_path "$profile_dir/BrowserMetrics"', script)
self.assertIn('prune_path "$profile_dir/Crash Reports"', script)
self.assertIn('prune_path "$profile_dir/SingletonLock"', script)
self.assertIn('prune_path "$profile_dir/Default/Code Cache"', script)
def test_browser_policies_default_to_allowed_permissions(self) -> None:
policies = json.loads((ROOT / "services" / "neko" / "policies" / "policies.json").read_text(encoding="utf-8"))
self.assertEqual(policies["DefaultNotificationsSetting"], 1)
self.assertEqual(policies["DefaultGeolocationSetting"], 1)
self.assertTrue(policies["VideoCaptureAllowed"])
self.assertTrue(policies["AudioCaptureAllowed"])
self.assertEqual(policies["DeveloperToolsAvailability"], 1)
self.assertNotIn("ExtensionInstallAllowlist", policies)
tm_policy = policies["3rdparty"]["extensions"]["dhdgffkkebhmkfjojejmpbldmpobfkfo"]["jsonImport"][0]
self.assertEqual(tm_policy["url"], "http://127.0.0.1:18080/tampermonkey-provisioning.json")
self.assertTrue(tm_policy["hash"].startswith("1:"))
self.assertFalse(tm_policy["installAsSystemScripts"])
def test_google_chrome_supervisor_uses_direct_stable_command(self) -> None:
conf = (ROOT / "services" / "neko" / "google-chrome.conf").read_text(encoding="utf-8")
self.assertIn("command=/etc/neko/browser-launch.sh google-chrome", conf)
self.assertNotIn("--bwsi", conf)
self.assertIn("autorestart=true", conf)
self.assertIn("[program:tampermonkey-provisioning]", conf)
self.assertIn("127.0.0.1:18080", conf)
self.assertIn("[program:openbox]", conf)
self.assertIn("/usr/bin/openbox --config-file /etc/neko/openbox.xml", conf)
script = (ROOT / "services" / "neko" / "browser-launch.sh").read_text(encoding="utf-8")
self.assertIn('profile_dir="/home/neko/.config/google-chrome"', script)
self.assertIn('--user-data-dir="$profile_dir"', script)
def test_chromium_supervisor_uses_direct_stable_command(self) -> None:
conf = (ROOT / "services" / "neko" / "chromium.conf").read_text(encoding="utf-8")
self.assertIn("command=/etc/neko/browser-launch.sh chromium", conf)
self.assertNotIn("--bwsi", conf)
self.assertIn("autorestart=true", conf)
self.assertIn("[program:tampermonkey-provisioning]", conf)
self.assertIn("127.0.0.1:18080", conf)
self.assertIn("[program:openbox]", conf)
script = (ROOT / "services" / "neko" / "browser-launch.sh").read_text(encoding="utf-8")
self.assertIn('profile_dir="/home/neko/.config/chromium"', script)
self.assertIn('--user-data-dir="$profile_dir"', script)
def test_legacy_neko_compose_template_mounts_only_selected_supervisor_conf(self) -> None:
compose = (ROOT / "services" / "neko" / "docker-compose.yml").read_text(encoding="utf-8")
self.assertIn("./${NEKO_BROWSER_SUPERVISOR_CONF}:/etc/neko/supervisord/${NEKO_BROWSER_SUPERVISOR_CONF}:ro", compose)
self.assertIn("./browser-launch.sh:/etc/neko/browser-launch.sh:ro", compose)
self.assertIn("./provisioning:/var/www/provisioning:ro", compose)
self.assertIn('NEKO_MANAGED_POLICIES: "${NEKO_MANAGED_POLICIES:-1}"', compose)
self.assertIn("${NEKO_POLICY_DIR:-./policies}:/etc/chromium/policies/managed:ro", compose)
self.assertIn("${NEKO_POLICY_DIR:-./policies}:/etc/chromium-browser/policies/managed:ro", compose)
self.assertIn("${NEKO_POLICY_DIR:-./policies}:/etc/opt/chrome/policies/managed:ro", compose)
self.assertIn("./external-extensions:/opt/google/chrome/extensions:ro", compose)
self.assertIn("./external-extensions:/usr/share/chromium/extensions:ro", compose)
self.assertIn("./external-extensions:/usr/share/chromium-browser/extensions:ro", compose)
self.assertIn("./tampermonkey-seed:/etc/neko/tampermonkey-seed:ro", compose)
self.assertNotIn("./no-managed-policies:/etc/chromium/policies/managed", compose)
self.assertNotIn("./no-managed-policies:/etc/opt/chrome/policies/managed", compose)
self.assertIn("container_name: live-neko-1", compose)
self.assertIn("container_name: live-neko-2", compose)
self.assertIn("container_name: live-neko-3", compose)
self.assertIn("${NEKO_HTTP_PORT_1}:8080", compose)
self.assertIn("${NEKO_HTTP_PORT_2}:8080", compose)
self.assertIn("${NEKO_HTTP_PORT_3}:8080", compose)
self.assertNotIn("/etc/neko/supervisord/browser.conf:ro", compose)
def test_dynamic_neko_compose_renderer_supports_more_than_three_instances(self) -> None:
spec = importlib.util.spec_from_file_location(
"render_neko_compose",
ROOT / "tools" / "render_neko_compose.py",
)
self.assertIsNotNone(spec)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
with tempfile.TemporaryDirectory() as tmp:
with patch.dict(
os.environ,
{
"NEKO_IMAGE": "test-image",
"NEKO_PROFILE_HOST_DIR": f"{tmp}/profile",
"NEKO_HTTP_PORT_4": "9404",
"NEKO_WEBRTC_TCPMUX_4": "54304",
"NEKO_WEBRTC_UDPMUX_4": "54305",
"NEKO_WEBRTC_UDP_RANGE_4": "54400-54499",
"NEKO_PROFILE_HOST_DIR_4": f"{tmp}/custom-profile-4",
"NEKO_BROWSER_SUPERVISOR_CONF": "google-chrome.conf",
"NEKO_MANAGED_POLICIES": "1",
},
clear=False,
):
rendered = module.render(4)
self.assertIn("container_name: live-neko-4", rendered)
self.assertIn('"9404:8080"', rendered)
self.assertIn('"54304:54304/tcp"', rendered)
self.assertIn('"54305:54305/udp"', rendered)
self.assertIn('"54400-54499:54400-54499/udp"', rendered)
self.assertIn(f'"{tmp}/custom-profile-4:/home/neko/.config/google-chrome"', rendered)
self.assertIn('NEKO_MANAGED_POLICIES: "1"', rendered)
self.assertIn('"./policies:/etc/chromium/policies/managed:ro"', rendered)
self.assertIn('"./policies:/etc/chromium-browser/policies/managed:ro"', rendered)
self.assertIn('"./policies:/etc/opt/chrome/policies/managed:ro"', rendered)
self.assertIn('"./external-extensions:/opt/google/chrome/extensions:ro"', rendered)
self.assertIn('"./external-extensions:/usr/share/chromium/extensions:ro"', rendered)
self.assertIn('"./external-extensions:/usr/share/chromium-browser/extensions:ro"', rendered)
self.assertIn('"./tampermonkey-seed:/etc/neko/tampermonkey-seed:ro"', rendered)
with patch.dict(os.environ, {"NEKO_MANAGED_POLICIES": "0"}, clear=False):
personal_rendered = module.render(2)
self.assertIn('NEKO_MANAGED_POLICIES: "0"', personal_rendered)
self.assertIn('"./no-managed-policies:/etc/chromium/policies/managed:ro"', personal_rendered)
self.assertIn('"./no-managed-policies:/etc/chromium-browser/policies/managed:ro"', personal_rendered)
self.assertIn('"./no-managed-policies:/etc/opt/chrome/policies/managed:ro"', personal_rendered)
def test_neko_runtime_scripts_use_generated_compose(self) -> None:
service_ctl = (ROOT / "scripts" / "linux" / "service_ctl.sh").read_text(encoding="utf-8")
common = (ROOT / "scripts" / "linux" / "lib" / "common.sh").read_text(encoding="utf-8")
for script in (service_ctl, common):
self.assertIn("render_neko_compose_file()", script)
self.assertIn("tools/render_neko_compose.py", script)
self.assertIn("docker-compose.generated.yml", script)
self.assertIn("compose_file=\"$(render_neko_compose_file)\"", script)
self.assertIn("NEKO_MAX_INSTANCE_COUNT:-64", script)
self.assertIn("NEKO_MANAGED_POLICIES", script)
self.assertIn("copy_neko_tampermonkey_seed_dir()", script)
self.assertIn("find_neko_tampermonkey_seed_profile()", script)
self.assertIn("services/neko/tampermonkey-seed", script)
self.assertIn("sync_neko_tampermonkey_preferences()", script)
self.assertIn("should_replace_current", script)
self.assertIn('value.get("from_webstore") is not True', script)
self.assertIn('path.startswith("/")', script)
self.assertIn('value.get("location") == 8', script)
self.assertIn("neko_managed_policies_enabled()", script)
def test_tampermonkey_external_extension_seed_is_available_without_managed_policy(self) -> None:
external = json.loads(
(ROOT / "services" / "neko" / "external-extensions" / "dhdgffkkebhmkfjojejmpbldmpobfkfo.json").read_text(
encoding="utf-8"
)
)
self.assertEqual(external["external_update_url"], "https://clients2.google.com/service/update2/crx")
seed = ROOT / "services" / "neko" / "tampermonkey-seed" / "Default" / "Local Extension Settings" / "dhdgffkkebhmkfjojejmpbldmpobfkfo"
self.assertTrue(seed.is_dir())
seed_text = "\n".join(path.read_text(encoding="utf-8", errors="ignore") for path in seed.glob("*") if path.is_file())
self.assertIn("YouTube Studio Auto Dismiss", seed_text)
seed_extension = (
ROOT
/ "services"
/ "neko"
/ "tampermonkey-seed"
/ "Default"
/ "Extensions"
/ "dhdgffkkebhmkfjojejmpbldmpobfkfo"
/ "5.5.0_0"
/ "manifest.json"
)
self.assertTrue(seed_extension.is_file())
seed_preferences = json.loads(
(ROOT / "services" / "neko" / "tampermonkey-seed" / "Default" / "Preferences").read_text(encoding="utf-8")
)
self.assertEqual(seed_preferences["extensions"]["settings"]["dhdgffkkebhmkfjojejmpbldmpobfkfo"]["state"], 1)
launch = (ROOT / "services" / "neko" / "browser-launch.sh").read_text(encoding="utf-8")
self.assertIn("ensure_tampermonkey_seed", launch)
self.assertIn("sync_tampermonkey_preferences", launch)
self.assertIn("should_replace_current", launch)
self.assertIn('value.get("from_webstore") is not True', launch)
self.assertIn('path.startswith("/")', launch)
self.assertIn('value.get("location") == 8', launch)
self.assertNotIn("--load-extension=", launch)
self.assertNotIn("youtube-auto-dismiss-extension", launch)
self.assertIn("/etc/neko/tampermonkey-seed", launch)
self.assertIn("rm -rf \"$profile_dir/Default/Managed Extension Settings/$ext_id\"", launch)
def test_supervisor_uses_browser_launch_script(self) -> None:
chrome = (ROOT / "services" / "neko" / "google-chrome.conf").read_text(encoding="utf-8")
chromium = (ROOT / "services" / "neko" / "chromium.conf").read_text(encoding="utf-8")
self.assertIn("command=/etc/neko/browser-launch.sh google-chrome", chrome)
self.assertIn("command=/etc/neko/browser-launch.sh chromium", chromium)
self.assertNotIn("--start-maximized", chrome)
self.assertNotIn("--start-maximized", chromium)
def test_tampermonkey_provisioning_matches_policy_hash(self) -> None:
provisioning_path = ROOT / "services" / "neko" / "provisioning" / "tampermonkey-provisioning.json"
userscript_path = ROOT / "services" / "neko" / "provisioning" / "youtube-studio-auto-dismiss.user.js"
payload = json.loads(provisioning_path.read_text(encoding="utf-8"))
self.assertEqual(payload["version"], "1")
self.assertEqual(payload["settings"]["runtime_content_mode"], "content")
self.assertEqual(payload["scripts"][0]["file_url"], "https://greasyfork.org/scripts/557378-youtube-studio-auto-dismiss/code/YouTube%20Studio%20Auto%20Dismiss.user.js")
self.assertEqual(payload["scripts"][0]["options"], {})
decoded_source = base64.b64decode(payload["scripts"][0]["source"]).decode("utf-8")
self.assertEqual(decoded_source, userscript_path.read_text(encoding="utf-8"))
spec = importlib.util.spec_from_file_location(
"render_tampermonkey_provisioning",
ROOT / "tools" / "render_tampermonkey_provisioning.py",
)
self.assertIsNotNone(spec)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
policies = json.loads((ROOT / "services" / "neko" / "policies" / "policies.json").read_text(encoding="utf-8"))
tm_import = policies["3rdparty"]["extensions"]["dhdgffkkebhmkfjojejmpbldmpobfkfo"]["jsonImport"][0]
self.assertFalse(tm_import["installAsSystemScripts"])
expected_hash = tm_import["hash"]
self.assertEqual(expected_hash, f"1:{module.tm_hash(payload)}")
def test_neko_scripts_use_family_specific_profile_dirs(self) -> None:
service_ctl = (ROOT / "scripts" / "linux" / "service_ctl.sh").read_text(encoding="utf-8")
common = (ROOT / "scripts" / "linux" / "lib" / "common.sh").read_text(encoding="utf-8")
self.assertIn("resolve_neko_profile_dir()", service_ctl)
self.assertIn('${family}-profile-v2', service_ctl)
self.assertIn("resolve_neko_profile_dir()", common)
self.assertIn('${family}-profile-v2', common)
self.assertIn("resolve_neko_profile_dir_for_instance()", service_ctl)
self.assertIn("seed_neko_tampermonkey_profile()", service_ctl)
self.assertIn("Local Extension Settings/dhdgffkkebhmkfjojejmpbldmpobfkfo", service_ctl)
self.assertIn("NEKO_HTTP_PORT_2", service_ctl)
self.assertIn("NEKO_WEBRTC_UDP_RANGE_3", service_ctl)
self.assertIn("resolve_neko_profile_dir_for_instance()", common)
self.assertIn("seed_neko_tampermonkey_profile()", common)
self.assertIn("Local Extension Settings/dhdgffkkebhmkfjojejmpbldmpobfkfo", common)
self.assertIn("NEKO_HTTP_PORT_2", common)
self.assertIn("NEKO_WEBRTC_UDP_RANGE_3", common)
self.assertIn('eval "override=\\"\\${NEKO_PROFILE_HOST_DIR_${idx}:-}\\""', common)
def test_neko_common_profile_helper_honors_instance_override_beyond_three(self) -> None:
script = (
f"ROOT_DIR={ROOT!s}; "
f"source {ROOT / 'scripts' / 'linux' / 'lib' / 'common.sh'}; "
"NEKO_PROFILE_HOST_DIR=/var/lib/live-neko/profile; "
"NEKO_PROFILE_HOST_DIR_4=/srv/live/neko/channel-four; "
"resolve_neko_profile_dir_for_instance 4"
)
proc = subprocess.run(["bash", "-lc", script], capture_output=True, text=True, check=False)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertEqual(proc.stdout.strip(), "/srv/live/neko/channel-four")
def test_neko_common_arch_selection_covers_x86_and_arm(self) -> None:
script = (
f"ROOT_DIR={ROOT!s}; "
f"source {ROOT / 'scripts' / 'linux' / 'lib' / 'common.sh'}; "
"NEKO_IMAGE_X86=ghcr.io/m1k1o/neko/google-chrome:3.0.0; "
"NEKO_IMAGE_ARM64=ghcr.io/m1k1o/neko/chromium:3.0.0; "
"uname(){ printf '%s\\n' \"$FAKE_UNAME\"; }; "
"FAKE_UNAME=x86_64; resolve_neko_image; printf ' '; resolve_neko_supervisor_conf; printf '\\n'; "
"FAKE_UNAME=aarch64; resolve_neko_image; printf ' '; resolve_neko_supervisor_conf; printf '\\n'"
)
proc = subprocess.run(["bash", "-lc", script], capture_output=True, text=True, check=False)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertEqual(
proc.stdout.splitlines(),
[
"ghcr.io/m1k1o/neko/google-chrome:3.0.0 google-chrome.conf",
"ghcr.io/m1k1o/neko/chromium:3.0.0 chromium.conf",
],
)
def test_neko_install_defaults_match_runtime_screen_profile(self) -> None:
common = (ROOT / "scripts" / "linux" / "lib" / "common.sh").read_text(encoding="utf-8")
service_ctl = (ROOT / "scripts" / "linux" / "service_ctl.sh").read_text(encoding="utf-8")
self.assertIn('NEKO_DESKTOP_SCREEN="${NEKO_DESKTOP_SCREEN:-1280x720@15}"', common)
self.assertIn('NEKO_DESKTOP_SCREEN="${NEKO_DESKTOP_SCREEN:-1280x720@15}"', service_ctl)
self.assertNotIn('NEKO_DESKTOP_SCREEN="${NEKO_DESKTOP_SCREEN:-1920x1080@30}"', common)
def test_neko_ip_watcher_is_installed_and_restarts_on_drift(self) -> None:
common = (ROOT / "scripts" / "linux" / "lib" / "common.sh").read_text(encoding="utf-8")
watcher = (ROOT / "scripts" / "linux" / "neko_ip_watch.sh").read_text(encoding="utf-8")
upgrade = (ROOT / "scripts" / "linux" / "upgrade_live_console.sh").read_text(encoding="utf-8")
uninstall = (ROOT / "scripts" / "linux" / "uninstall_stack.sh").read_text(encoding="utf-8")
stack_env = (ROOT / "config" / "system-stack.env.example").read_text(encoding="utf-8")
self.assertIn("install_neko_ip_watch_service()", common)
self.assertIn("live-neko-ip-watch.service", common)
self.assertIn("ExecStart=/bin/bash", common)
self.assertIn("scripts/linux/neko_ip_watch.sh", common)
self.assertIn("detect_preferred_lan_ip", watcher)
self.assertIn('bash "$SERVICE_CTL" neko restart', watcher)
self.assertIn("NEKO_IP_WATCH_INTERVAL", watcher)
self.assertIn("install_neko_ip_watch_service", upgrade)
self.assertIn("live-neko-ip-watch.service", uninstall)
self.assertIn("NEKO_IP_WATCH_INTERVAL=15", stack_env)
self.assertIn("NEKO_MANAGED_POLICIES=1", stack_env)
def test_local_browser_prefers_chrome_on_x86_and_chromium_elsewhere(self) -> None:
script = (ROOT / "scripts" / "linux" / "launch_browser.sh").read_text(encoding="utf-8")
self.assertIn("x86_64|amd64", script)
x86_branch = script.split("x86_64|amd64)", 1)[1].split(";;", 1)[0]
other_branch = script.split("*)", 1)[1].split(";;", 1)[0]
self.assertLess(x86_branch.find("google-chrome"), x86_branch.find("chromium"))
self.assertLess(other_branch.find("chromium"), other_branch.find("google-chrome"))
if __name__ == "__main__":
unittest.main()