208 lines
7.7 KiB
Python
208 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import subprocess
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
USER_INSTALLER = (
|
|
PROJECT_ROOT / "deploy" / "scripts" / "install-wechat-health-watch.sh"
|
|
)
|
|
PRODUCTION_INSTALLER = (
|
|
PROJECT_ROOT / "deploy" / "scripts" / "deploy-jwebot-production.sh"
|
|
)
|
|
BOOTSTRAP = PROJECT_ROOT / "deploy" / "scripts" / "bootstrap-jwebot.sh"
|
|
LEGACY_INSTALLER = (
|
|
PROJECT_ROOT / "deploy" / "scripts" / "install-wechat-local-ocr.sh"
|
|
)
|
|
LEGACY_UPGRADER = (
|
|
PROJECT_ROOT / "deploy" / "scripts" / "upgrade-wechat-local-ocr.sh"
|
|
)
|
|
ADAPTIVE_UNIT = (
|
|
PROJECT_ROOT
|
|
/ "deploy"
|
|
/ "systemd"
|
|
/ "user"
|
|
/ "wechat-ocr-adaptive-recorder.service"
|
|
)
|
|
PACKAGE_GATE = PROJECT_ROOT / "tools" / "check-packages.sh"
|
|
LOCAL_GATE = PROJECT_ROOT / "tools" / "check-local.sh"
|
|
|
|
|
|
class DeploymentScriptTests(unittest.TestCase):
|
|
def test_installers_are_executable_and_parse_with_their_shells(self):
|
|
for path, shell in (
|
|
(BOOTSTRAP, "bash"),
|
|
(USER_INSTALLER, "sh"),
|
|
(PRODUCTION_INSTALLER, "bash"),
|
|
):
|
|
self.assertTrue(os.access(path, os.X_OK), path)
|
|
result = subprocess.run(
|
|
[shell, "-n", str(path)],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_user_installer_owns_the_complete_adaptive_runtime(self):
|
|
source = USER_INSTALLER.read_text(encoding="utf-8")
|
|
for required in (
|
|
"wechat_local_ocr.py",
|
|
"wechat_health_watch.py",
|
|
"wechat_adaptive_recorder.py",
|
|
"wechat_adaptive_gui_bridge.py",
|
|
"jwebot_edge.py",
|
|
"jwebot_edge_delivery.py",
|
|
"jwebot_config",
|
|
"jwebot_sdk",
|
|
"jwebot_kernel",
|
|
"jwebot_official_extensions",
|
|
"data/cs_autochess",
|
|
"official_autochess_data_source_dir",
|
|
):
|
|
self.assertIn(required, source)
|
|
|
|
def test_user_installer_has_rollback_and_stability_gates(self):
|
|
source = USER_INSTALLER.read_text(encoding="utf-8")
|
|
for required in (
|
|
"systemd-user-state.tsv",
|
|
"restore_unit_states",
|
|
"legacy system OCR is active",
|
|
'{"op":"health"}',
|
|
"EDGE_HEALTH_TIMEOUT_SECONDS:-45",
|
|
"wait_for_edge_health",
|
|
"ADAPTIVE_READY_TIMEOUT_SECONDS:-90",
|
|
"wait_for_adaptive_ready",
|
|
"adaptive-start.marker",
|
|
'value.get("initialized") is True',
|
|
'captured_at.timestamp() >= int(sys.argv[2])',
|
|
"NRestarts",
|
|
'test "$restarts_before" = "$restarts_after"',
|
|
"[JWeBot] %s",
|
|
'jwebot_log "edge starting"',
|
|
'jwebot_log "edge active health=ok"',
|
|
'jwebot_log "adaptive starting"',
|
|
'jwebot_log "adaptive active"',
|
|
'jwebot_log "adaptive ready baseline=captured"',
|
|
'jwebot_log "user runtime ready"',
|
|
):
|
|
self.assertIn(required, source)
|
|
self.assertLess(
|
|
source.index('jwebot_log "adaptive active"'),
|
|
source.index('jwebot_log "adaptive ready baseline=captured"'),
|
|
)
|
|
self.assertLess(
|
|
source.index('jwebot_log "adaptive ready baseline=captured"'),
|
|
source.index('jwebot_log "user runtime ready"'),
|
|
)
|
|
|
|
def test_root_installer_enforces_the_production_cutover_contract(self):
|
|
source = PRODUCTION_INSTALLER.read_text(encoding="utf-8")
|
|
for required in (
|
|
"production deploy must run as root",
|
|
"assert_queues_empty",
|
|
"assert_wechat_ready",
|
|
'[[ "${title}" == "Weixin" ]]',
|
|
"WECHAT_READY_TIMEOUT_SECONDS:-30",
|
|
"/usr/sbin/runuser",
|
|
"/usr/bin/xdotool search --onlyvisible --class wechat",
|
|
"/usr/bin/xdotool search --onlyvisible --name '^Weixin$'",
|
|
'jwebot_log "WeChat readiness starting timeout=',
|
|
'source.backup(target)',
|
|
"restore_enable_state",
|
|
"restore_active_state",
|
|
"install-wechat-health-watch.sh",
|
|
"systemctl disable --now",
|
|
"wechat-local-ocr-recorder.service",
|
|
"wechat-local-ocr-responder.service",
|
|
"sha256sum -c SHA256SUMS",
|
|
"[JWeBot] %s",
|
|
'jwebot_log "preflight starting"',
|
|
'jwebot_log "runtime install ready"',
|
|
'jwebot_log "edge active"',
|
|
'jwebot_log "adaptive active"',
|
|
'jwebot_log "deployment ready version=',
|
|
'cd "${target_home}"',
|
|
):
|
|
self.assertIn(required, source)
|
|
self.assertEqual(
|
|
source.count("runuser"),
|
|
source.count("/usr/sbin/runuser"),
|
|
)
|
|
self.assertEqual(
|
|
source.count("xdotool"),
|
|
source.count("/usr/bin/xdotool"),
|
|
)
|
|
|
|
def test_production_reports_bounded_progress(self):
|
|
source = PRODUCTION_INSTALLER.read_text(encoding="utf-8")
|
|
self.assertIn("jwebot_progress()", source)
|
|
self.assertIn('jwebot_progress 1 8 "preflight"', source)
|
|
self.assertIn('jwebot_progress 8 8 "deployment ready"', source)
|
|
self.assertNotIn("pixel_art", source)
|
|
|
|
def test_production_preflight_verifies_rendered_runtime_units(self):
|
|
source = PRODUCTION_INSTALLER.read_text(encoding="utf-8")
|
|
render = 'render_root_configuration "${preflight_rendered_dir}"'
|
|
verify = (
|
|
'systemd-analyze verify \\\n'
|
|
' "${preflight_rendered_dir}/wechat-local-ocr-recorder.service"'
|
|
)
|
|
self.assertIn("render_root_configuration()", source)
|
|
self.assertIn("preflight_rendered_dir=$(mktemp -d)", source)
|
|
self.assertIn(render, source)
|
|
self.assertIn(verify, source)
|
|
self.assertLess(source.index(render), source.index(verify))
|
|
self.assertNotIn(
|
|
'systemd-analyze verify \\\n'
|
|
' "${source_dir}/deploy/systemd/system/'
|
|
'wechat-local-ocr-recorder.service"',
|
|
source,
|
|
)
|
|
|
|
def test_bootstrap_has_prepare_verify_and_deploy_stages(self):
|
|
source = BOOTSTRAP.read_text(encoding="utf-8")
|
|
for required in (
|
|
"prepare()",
|
|
"verify()",
|
|
"deploy()",
|
|
"requirements-runtime.txt",
|
|
"jwebot_config",
|
|
"tools/check-secrets.py",
|
|
"xdotool search --onlyvisible --class wechat",
|
|
"deploy-jwebot-production.sh",
|
|
"progress 6 6",
|
|
):
|
|
self.assertIn(required, source)
|
|
|
|
def test_legacy_maintenance_never_enables_services_implicitly(self):
|
|
installer = LEGACY_INSTALLER.read_text(encoding="utf-8")
|
|
upgrader = LEGACY_UPGRADER.read_text(encoding="utf-8")
|
|
self.assertIn("ENABLE_LEGACY_OCR:-0", installer)
|
|
self.assertIn("restore_service_state", installer)
|
|
self.assertIn("restore_service_state", upgrader)
|
|
self.assertNotIn(
|
|
'systemctl restart "$recorder_service" "$responder_service"',
|
|
upgrader,
|
|
)
|
|
|
|
def test_adaptive_unit_bounds_real_crash_restarts(self):
|
|
unit = ADAPTIVE_UNIT.read_text(encoding="utf-8")
|
|
self.assertIn("Restart=on-failure", unit)
|
|
self.assertIn("StartLimitIntervalSec=300", unit)
|
|
self.assertIn("StartLimitBurst=6", unit)
|
|
|
|
def test_release_gates_are_repeatable_without_ripgrep(self):
|
|
package_gate = PACKAGE_GATE.read_text(encoding="utf-8")
|
|
local_gate = LOCAL_GATE.read_text(encoding="utf-8")
|
|
self.assertIn("PYTHONPYCACHEPREFIX", package_gate)
|
|
self.assertIn("command -v rg", package_gate)
|
|
self.assertIn("command -v rg", local_gate)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|