tests/test_deploy_windows_tray.py
Rama master · réplica de lectura del repositorio original.
from __future__ import annotations
import importlib.util
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from peach import __version__
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "deploy_windows_tray.py"
SPEC = importlib.util.spec_from_file_location("deploy_windows_tray", SCRIPT)
deploy_windows_tray = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = deploy_windows_tray
SPEC.loader.exec_module(deploy_windows_tray)
def git_runner(porcelain: str = "", commit: str = "abc12345def", branch: str = "master"):
answers = {
"status --porcelain": porcelain,
"rev-parse HEAD": commit,
"rev-parse --abbrev-ref HEAD": branch,
}
def run(command, **_kwargs):
return subprocess.CompletedProcess(
command, 0, stdout=answers[" ".join(command[1:])], stderr="",
)
return run
class DeployWindowsTrayTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
# 部署入口先 resolve 检出根和目标，断言也得拿 resolve 过的路径比。
self.root = Path(self.temp.name).resolve()
self.target = self.root / "dist" / "Peach" / "Peach.exe"
self.target.parent.mkdir(parents=True)
self.target.write_bytes(b"old")
self.staged = self.root / "staging" / "Peach.exe"
self.staged.parent.mkdir()
self.staged.write_bytes(b"new")
self.installer = mock.Mock()
self.installer.log_path = self.root / "sync.log"
self.installer.build_staged_tray.return_value = self.staged
self.installer.packaged_migrations_pass.return_value = True
self.restarted = mock.Mock(return_value=mock.Mock(
ok=True, message="托盘已正常重启并重新拥有 HTTP/HTTPS 子服务",
old_tray_pid=10, new_tray_pid=30, service_pids=(51, 52),
backup=str(self.target.with_name("Peach.pre-source-sync-20260906-120000.exe")),
))
def tearDown(self):
self.temp.cleanup()
def run_deploy(self, **overrides):
arguments = dict(
target=self.target, installer=self.installer, restart=self.restarted,
run=git_runner(),
identity=lambda **_kwargs: {"version": __version__, "build_commit": "abc12345def"},
)
arguments.update(overrides)
return deploy_windows_tray.deploy(self.root, **arguments)
def test_a_dirty_checkout_stops_before_anything_is_built_or_swapped(self):
outcome = self.run_deploy(run=git_runner(porcelain=" M src/peach/tray.py\n"))
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["step"], "checkout")
self.installer.build_staged_tray.assert_not_called()
self.restarted.assert_not_called()
def test_a_package_that_fails_the_migration_check_never_reaches_production(self):
self.installer.packaged_migrations_pass.return_value = False
outcome = self.run_deploy()
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["step"], "validate")
self.restarted.assert_not_called()
self.assertEqual(self.target.read_bytes(), b"old")
def test_a_given_package_is_reused_instead_of_building_again(self):
outcome = self.run_deploy(staged=self.staged)
self.assertTrue(outcome["ok"], outcome["message"])
self.installer.build_staged_tray.assert_not_called()
self.assertEqual(self.restarted.call_args.kwargs["swap_from"], self.staged)
self.assertEqual(outcome["commit"], "abc12345def")
self.assertEqual(outcome["version"], __version__)
def test_a_tray_built_from_another_commit_is_reported_as_a_failure(self):
"""认的是构建提交：同一个版本号下有很多个构建，只比版本号证明不了二进制换了。"""
outcome = self.run_deploy(
identity=lambda **_kwargs: {"version": __version__, "build_commit": "0ldc0mm1t"})
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["step"], "verify")
self.assertEqual(outcome["served_commit"], "0ldc0mm1t")
self.assertEqual(outcome["expected_version"], __version__)
self.assertEqual(outcome["backup"], self.restarted.return_value.backup)
def swapping_restart(self):
"""照真实那步的样子换文件：`os.replace` 是搬走，暂存包换完就不在了。
指纹只能在换之前取，这个替身把「换完再去读暂存包」直接变成 FileNotFoundError。
"""
def restart(target, *, swap_from, **_kwargs):
os.replace(swap_from, target)
return self.restarted.return_value
return restart
def test_a_venv_owned_deployment_is_verified_by_the_binary_that_landed(self):
"""回话的是源码进程时，认落地那个文件的指纹。
托盘按 `_peach_executable()` 的设计把两个子服务交回项目 venv，那个进程没有第二个
版本，`/healthz` 的 `build_commit` 结构上恒为 None。拿它当判据的话，二进制换得
再对也必然判失败——这条链路上唯一动了生产入口的那一步，反而永远报不出成功。
"""
outcome = self.run_deploy(
staged=self.staged, restart=self.swapping_restart(),
identity=lambda **_kwargs: {"version": __version__, "build_commit": None})
self.assertTrue(outcome["ok"], outcome["message"])
self.assertEqual(outcome["step"], "verify")
self.assertEqual(outcome["built_digest"], outcome["target_digest"])
self.assertEqual(self.target.read_bytes(), b"new")
def test_a_binary_that_never_landed_fails_even_though_the_tray_answers(self):
"""托盘回话不等于跑的是这一份：指纹对不上就是没换上。"""
outcome = self.run_deploy(
staged=self.staged,
identity=lambda **_kwargs: {"version": __version__, "build_commit": None})
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["step"], "verify")
self.assertNotEqual(outcome["built_digest"], outcome["target_digest"])
self.assertIn("指纹", outcome["message"])
self.assertEqual(outcome["backup"], self.restarted.return_value.backup)
def test_a_silent_health_endpoint_is_reported_as_a_failure(self):
outcome = self.run_deploy(identity=lambda **_kwargs: None)
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["step"], "verify")
self.assertIn("/healthz", outcome["message"])
def test_the_last_health_error_is_reported_verbatim(self):
error = "https://192.0.2.10/healthz ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED]"
outcome = self.run_deploy(identity=lambda **_kwargs: {"error": error})
self.assertFalse(outcome["ok"])
self.assertEqual(outcome["step"], "verify")
self.assertIn(error, outcome["message"])
class ServingIdentityTests(unittest.TestCase):
def test_the_verdict_comes_from_the_https_port_checked_against_the_project_ca(self):
specs = (
mock.Mock(name="http", health_url="http://127.0.0.1/healthz", verify=True),
mock.Mock(health_url="https://192.0.2.10/healthz", verify="/ca/peach.crt"),
)
specs[0].name = "http"
specs[1].name = "https"
response = mock.Mock(status_code=200)
response.json.return_value = {"ok": True, "version": "9.9.9", "build_commit": "c0ffee"}
with (
mock.patch.object(deploy_windows_tray, "build_service_specs",
return_value=specs),
mock.patch.object(deploy_windows_tray.httpx, "get",
return_value=response) as get,
):
self.assertEqual(deploy_windows_tray.serving_identity(),
{"version": "9.9.9", "build_commit": "c0ffee"})
self.assertEqual(get.call_args.args[0], "https://192.0.2.10/healthz")
self.assertEqual(get.call_args.kwargs["verify"], "/ca/peach.crt")
self.assertFalse(get.call_args.kwargs["trust_env"],
"系统代理会替生产口回话，探测必须直连")
def test_a_timeout_carries_the_last_raw_error(self):
spec = mock.Mock(health_url="https://192.0.2.10/healthz", verify="/ca/peach.crt")
spec.name = "https"
refused = deploy_windows_tray.httpx.ConnectError("[SSL: CERTIFICATE_VERIFY_FAILED]")
unready = mock.Mock(status_code=503)
for answer, expected in ((refused, "ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED]"),
(unready, "HTTP 503")):
with self.subTest(expected=expected), \
mock.patch.object(deploy_windows_tray, "build_service_specs", return_value=(spec,)), \
mock.patch.object(deploy_windows_tray.httpx, "get",
**({"side_effect": answer} if isinstance(answer, Exception)
else {"return_value": answer})):
served = deploy_windows_tray.serving_identity(timeout=0, sleep=mock.Mock())
self.assertEqual(served, {"error": f"https://192.0.2.10/healthz {expected}"})
if __name__ == "__main__":
unittest.main()