64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
import unittest
|
|
|
|
from tools import go_native_deploy as deploy
|
|
|
|
|
|
class GoNativeDeployTest(unittest.TestCase):
|
|
def test_build_commands_cover_four_production_binaries(self):
|
|
commands = deploy.build_commands("/repo/go/vehicle-gateway", "/tmp/out")
|
|
|
|
self.assertEqual([command.output_name for command in commands], [
|
|
"gateway",
|
|
"history-writer",
|
|
"stat-writer",
|
|
"realtime-api",
|
|
])
|
|
for command in commands:
|
|
self.assertEqual(command.argv[:3], ["go", "build", "-trimpath"])
|
|
self.assertIn("GOOS=linux", command.env)
|
|
self.assertIn("GOARCH=amd64", command.env)
|
|
self.assertIn("CGO_ENABLED=0", command.env)
|
|
|
|
def test_remote_switch_command_creates_release_and_restarts_all_services(self):
|
|
command = deploy.remote_switch_command("409f55b", "/opt/lingniu-go-native", "/tmp/lingniu-go-native-409f55b.tar.gz")
|
|
|
|
self.assertIn("/opt/lingniu-go-native/releases/409f55b", command)
|
|
self.assertIn("ln -sfn", command)
|
|
self.assertIn("systemctl restart lingniu-go-gateway", command)
|
|
self.assertIn("systemctl restart lingniu-go-history-writer", command)
|
|
self.assertIn("systemctl restart lingniu-go-stat-writer", command)
|
|
self.assertIn("systemctl restart lingniu-go-realtime-api", command)
|
|
|
|
def test_acceptance_command_uses_deployed_hosts_and_date(self):
|
|
command = deploy.acceptance_command(
|
|
app_host="115.29.187.205",
|
|
kafka_host="114.55.58.251",
|
|
date="2026-07-02",
|
|
timeout=20,
|
|
)
|
|
|
|
self.assertEqual(command[:2], ["python3", "tools/go_prod_acceptance.py"])
|
|
self.assertIn("--app-host", command)
|
|
self.assertIn("115.29.187.205", command)
|
|
self.assertIn("--kafka-host", command)
|
|
self.assertIn("114.55.58.251", command)
|
|
self.assertIn("--date", command)
|
|
self.assertIn("2026-07-02", command)
|
|
|
|
def test_parse_spool_status_reads_file_and_recent_counts(self):
|
|
status = deploy.parse_spool_status("files=42 recent=3")
|
|
|
|
self.assertEqual(status.files, 42)
|
|
self.assertEqual(status.recent, 3)
|
|
|
|
def test_spool_status_command_targets_gateway_spool_directory(self):
|
|
command = deploy.spool_status_command("/opt/lingniu-go-native")
|
|
|
|
self.assertIn("/opt/lingniu-go-native/spool/gateway", command)
|
|
self.assertIn("files=", command)
|
|
self.assertIn("recent=", command)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|