hyapp-server/deploy/tencent-tat/tests/test_hyapp_tat_deploy.py
2026-05-13 16:07:25 +08:00

111 lines
4.3 KiB
Python

from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from deploy.tencent_tat import hyapp_tat_deploy as deploy
class TatDeployPlanTests(unittest.TestCase):
def load_inventory(self) -> dict:
return deploy.load_inventory("deploy/tencent-tat/inventory.prod.example.json")
def test_inventory_contains_current_cvm_topology(self) -> None:
inventory = self.load_inventory()
self.assertEqual(inventory["hosts"]["gateway-1"]["instance_id"], "ins-aeyaoj40")
self.assertEqual(inventory["hosts"]["gateway-2"]["private_ip"], "10.2.2.17")
self.assertEqual(inventory["services"]["wallet-service"]["hosts"], ["pay-1", "pay-2"])
self.assertEqual(inventory["services"]["game-service"]["target_port"], 13008)
self.assertIn("TencentDB for MySQL", inventory["managed_dependencies"]["mysql"])
def test_plan_is_service_first_and_keeps_clb_enabled(self) -> None:
inventory = self.load_inventory()
plan = deploy.build_plan(
inventory,
action="deploy",
service_names=["gateway-service", "wallet-service"],
raw_hosts="",
tag="20260513",
overrides={},
no_clb=False,
)
self.assertEqual(
[(step["service"], step["host"]) for step in plan],
[
("gateway-service", "gateway-1"),
("gateway-service", "gateway-2"),
("wallet-service", "pay-1"),
("wallet-service", "pay-2"),
],
)
self.assertTrue(all(step["clb_enabled"] for step in plan))
self.assertEqual(plan[0]["image"], "10.2.1.3:18082/hyapp/gateway-service:20260513")
def test_host_filter_must_belong_to_service(self) -> None:
inventory = self.load_inventory()
with self.assertRaisesRegex(RuntimeError, "not assigned"):
deploy.build_plan(
inventory,
action="restart",
service_names=["gateway-service"],
raw_hosts="pay-1",
tag="",
overrides={},
no_clb=False,
)
def test_remote_deploy_script_only_touches_hyapp_unit_and_env(self) -> None:
inventory = self.load_inventory()
service = inventory["services"]["gateway-service"]
script = deploy.remote_restart_script(service, image="registry/gateway:tag", health_timeout_seconds=30)
self.assertIn("systemctl restart \"$UNIT\"", script)
self.assertIn("ENV_FILE=/etc/hyapp/gateway-service/docker.env", script)
self.assertIn("docker pull \"$IMAGE\"", script)
self.assertNotIn("docker compose", script)
self.assertNotIn("java", script.lower())
def test_discover_plan_does_not_require_image_tag(self) -> None:
inventory = self.load_inventory()
plan = deploy.build_plan(
inventory,
action="discover",
service_names=["gateway-service"],
raw_hosts="",
tag="",
overrides={},
no_clb=False,
)
self.assertEqual([step["action"] for step in plan], ["discover", "discover"])
self.assertEqual([step["image"] for step in plan], ["", ""])
def test_real_clb_ids_disable_auto_discovery_but_placeholders_enable_it(self) -> None:
inventory = self.load_inventory()
step = deploy.build_plan(
inventory,
action="restart",
service_names=["gateway-service"],
raw_hosts="gateway-1",
tag="",
overrides={},
no_clb=False,
)[0]
self.assertFalse(deploy.should_auto_discover_clb(step))
step["clb_load_balancer_id"] = "lb-REPLACE_PUBLIC_GATEWAY"
self.assertTrue(deploy.should_auto_discover_clb(step))
def test_validate_inventory_rejects_unknown_service_host(self) -> None:
inventory = self.load_inventory()
inventory["services"]["gateway-service"]["hosts"] = ["missing-host"]
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "inventory.json"
path.write_text(json.dumps(inventory), encoding="utf-8")
with self.assertRaisesRegex(RuntimeError, "unknown host"):
deploy.load_inventory(str(path))
if __name__ == "__main__":
unittest.main()