113 lines
4.5 KiB
Python
113 lines
4.5 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.assertEqual(inventory["services"]["notice-service"]["hosts"], ["new-app-activity-1", "new-app-activity-2"])
|
|
self.assertEqual(inventory["services"]["notice-service"]["target_port"], 13009)
|
|
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()
|