eric/gitlab-instance-0a899031_sh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

s

Browse Source
This commit is contained in:
eric
2026-05-19 12:52:04 +00:00
parent bae55ec52c
commit 1d89aa29e0
27 changed files with 251 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
d2ypp2/tests/test_neko_capacity.py
View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from subprocess import CompletedProcess
from unittest.mock import patch
from src.neko_capacity import (
@@ -36,7 +37,7 @@ class NekoCapacityTests(unittest.TestCase):
self.assertTrue(changed)
self.assertEqual(env.read_text(encoding="utf-8"), "HOSTNAME_ALIAS=live\nNEKO_INSTANCE_COUNT=3\nPORT=8001\n")
def test_ensure_capacity_increases_env_and_restarts_neko_in_background(self) -> None:
def test_ensure_capacity_increases_env_and_restarts_neko_synchronously(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
env = root / "config" / "system-stack.env"
@@ -46,17 +47,76 @@ class NekoCapacityTests(unittest.TestCase):
service_ctl.parent.mkdir(parents=True)
service_ctl.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
with patch("src.neko_capacity.subprocess.Popen") as popen:
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
with patch("src.neko_capacity._compose_has_neko_service", side_effect=[False, True]):
with patch(
"src.neko_capacity.subprocess.run",
return_value=CompletedProcess(args=["service_ctl"], returncode=0),
) as run:
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
self.assertTrue(result["ok"])
self.assertTrue(result["changed"])
self.assertTrue(result["restart_required"])
self.assertTrue(result["restarted"])
self.assertTrue(result["compose_ready"])
self.assertIn("NEKO_INSTANCE_COUNT=3", env.read_text(encoding="utf-8"))
popen.assert_called_once()
self.assertEqual(popen.call_args.args[0][-2:], ["neko", "restart"])
run.assert_called_once()
self.assertEqual(run.call_args.args[0][-2:], ["neko", "restart"])
def test_ensure_capacity_repairs_stale_generated_compose_even_when_env_count_is_high_enough(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
env = root / "config" / "system-stack.env"
env.parent.mkdir()
env.write_text("NEKO_INSTANCE_COUNT=3\n", encoding="utf-8")
service_ctl = root / "scripts" / "linux" / "service_ctl.sh"
service_ctl.parent.mkdir(parents=True)
service_ctl.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
with patch("src.neko_capacity._compose_has_neko_service", side_effect=[False, True]):
with patch(
"src.neko_capacity.subprocess.run",
return_value=CompletedProcess(args=["service_ctl"], returncode=0),
) as run:
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
self.assertTrue(result["ok"])
self.assertFalse(result["changed"])
self.assertTrue(result["restart_required"])
self.assertTrue(result["restarted"])
self.assertIn("compose_missing_desired_service", result["reason"])
run.assert_called_once()
def test_ensure_capacity_reports_restart_success_but_missing_compose_as_not_ok(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
env = root / "config" / "system-stack.env"
env.parent.mkdir()
env.write_text("NEKO_INSTANCE_COUNT=2\n", encoding="utf-8")
service_ctl = root / "scripts" / "linux" / "service_ctl.sh"
service_ctl.parent.mkdir(parents=True)
service_ctl.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
with patch("src.neko_capacity._compose_has_neko_service", return_value=False):
with patch(
"src.neko_capacity.subprocess.run",
return_value=CompletedProcess(args=["service_ctl"], returncode=0),
):
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
self.assertFalse(result["ok"])
self.assertTrue(result["restarted"])
self.assertFalse(result["compose_ready"])
self.assertIn("compose_missing_desired_service_after_restart", result["reason"])
def test_ensure_capacity_keeps_saved_env_when_restart_spawn_fails(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
@@ -68,15 +128,43 @@ class NekoCapacityTests(unittest.TestCase):
service_ctl.parent.mkdir(parents=True)
service_ctl.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
with patch("src.neko_capacity.subprocess.Popen", side_effect=OSError("fork failed")):
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
with patch("src.neko_capacity._compose_has_neko_service", return_value=False):
with patch("src.neko_capacity.subprocess.run", side_effect=OSError("fork failed")):
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
self.assertTrue(result["changed"])
self.assertFalse(result["restarted"])
self.assertIn("restart_spawn_failed", result["reason"])
self.assertFalse(result["ok"])
self.assertIn("restart_failed", result["reason"])
self.assertIn("NEKO_INSTANCE_COUNT=3", env.read_text(encoding="utf-8"))
def test_ensure_capacity_keeps_saved_env_when_restart_exits_nonzero(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
env = root / "config" / "system-stack.env"
env.parent.mkdir()
env.write_text("NEKO_INSTANCE_COUNT=2\n", encoding="utf-8")
service_ctl = root / "scripts" / "linux" / "service_ctl.sh"
service_ctl.parent.mkdir(parents=True)
service_ctl.write_text("#!/usr/bin/env bash\n", encoding="utf-8")
with patch("src.neko_capacity._compose_has_neko_service", return_value=False):
with patch(
"src.neko_capacity.subprocess.run",
return_value=CompletedProcess(args=["service_ctl"], returncode=17),
):
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}, {"id": "lane2"}, {"id": "lane3"}],
)
self.assertTrue(result["changed"])
self.assertFalse(result["restarted"])
self.assertFalse(result["ok"])
self.assertIn("restart_exit_17", result["reason"])
self.assertIn("NEKO_INSTANCE_COUNT=3", env.read_text(encoding="utf-8"))
def test_ensure_capacity_does_not_downscale_manual_larger_count(self) -> None:
@@ -86,7 +174,12 @@ class NekoCapacityTests(unittest.TestCase):
env.parent.mkdir()
env.write_text("NEKO_INSTANCE_COUNT=3\n", encoding="utf-8")
result = ensure_neko_capacity_for_youtube_channels(root, [{"id": "lane1"}], restart=False)
with patch("src.neko_capacity._compose_has_neko_service", return_value=True):
result = ensure_neko_capacity_for_youtube_channels(
root,
[{"id": "lane1"}],
restart=False,
)
self.assertFalse(result["changed"])
self.assertEqual(env.read_text(encoding="utf-8"), "NEKO_INSTANCE_COUNT=3\n")

36
d2ypp2/tests/test_neko_config.py
View File

@@ -23,6 +23,7 @@ class NekoConfigTests(unittest.TestCase):
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)
@@ -80,16 +81,16 @@ class NekoConfigTests(unittest.TestCase):
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:-0}"', compose)
self.assertIn("./no-managed-policies:/etc/chromium/policies/managed:ro", compose)
self.assertIn("./no-managed-policies:/etc/chromium-browser/policies/managed:ro", compose)
self.assertIn("./no-managed-policies:/etc/opt/chrome/policies/managed: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("./policies:/etc/chromium/policies/managed", compose)
self.assertNotIn("./policies:/etc/opt/chrome/policies/managed", 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)
@@ -120,6 +121,7 @@ class NekoConfigTests(unittest.TestCase):
"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,
):
@@ -131,22 +133,21 @@ class NekoConfigTests(unittest.TestCase):
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: "0"', rendered)
self.assertIn('"./no-managed-policies:/etc/chromium/policies/managed:ro"', rendered)
self.assertIn('"./no-managed-policies:/etc/chromium-browser/policies/managed:ro"', rendered)
self.assertIn('"./no-managed-policies:/etc/opt/chrome/policies/managed:ro"', 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)
self.assertNotIn('"./policies:/etc/chromium/policies/managed:ro"', rendered)
self.assertNotIn('"./policies:/etc/opt/chrome/policies/managed:ro"', rendered)
with patch.dict(os.environ, {"NEKO_MANAGED_POLICIES": "1"}, clear=False):
managed_rendered = module.render(2)
self.assertIn('"./policies:/etc/chromium/policies/managed:ro"', managed_rendered)
self.assertIn('"./policies:/etc/chromium-browser/policies/managed:ro"', managed_rendered)
self.assertIn('"./policies:/etc/opt/chrome/policies/managed:ro"', managed_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")
@@ -315,6 +316,7 @@ class NekoConfigTests(unittest.TestCase):
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")

3
d2ypp2/tests/test_verify_local_install.py
View File

@@ -45,6 +45,7 @@ class VerifyLocalInstallTests(unittest.TestCase):
"ENABLE_LIVE_EDGE_PROXY=1",
"ENABLE_NEKO=1",
"NEKO_INSTANCE_COUNT=2",
"NEKO_MANAGED_POLICIES=1",
"ENABLE_ANDROID_PANEL=1",
"ENABLE_REDROID=1",
"",
@@ -96,7 +97,7 @@ class VerifyLocalInstallTests(unittest.TestCase):
if "Developer mode preference not found" in command:
return 0, "/home/neko/.config/google-chrome/Default/Preferences: developer_mode=true\n", ""
if "policies/managed" in command and "-name '*.json'" in command:
return 0, "", ""
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: