shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: verify go services are enabled

Browse Source
This commit is contained in:
lingniu
2026-07-02 01:55:39 +08:00
parent f4d9c9ef5a
commit 5b7f386c5a
4 changed files with 44 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
tools/test_go_systemd_prod_smoke.py
View File

@@ -6,10 +6,10 @@ from tools import go_systemd_prod_smoke as smoke
class GoSystemdProdSmokeTest(unittest.TestCase):
def test_service_checks_require_all_go_units_active(self):
output = """
lingniu-go-gateway.service active
lingniu-go-history-writer.service active
lingniu-go-stat-writer.service active
lingniu-go-realtime-api.service active
lingniu-go-gateway.service active enabled
lingniu-go-history-writer.service active enabled
lingniu-go-stat-writer.service active enabled
lingniu-go-realtime-api.service active enabled
"""
checks = smoke.service_checks(output, smoke.DEFAULT_SERVICES)
@@ -18,8 +18,8 @@ lingniu-go-realtime-api.service active
def test_service_checks_fail_when_unit_is_missing_or_inactive(self):
output = """
lingniu-go-gateway.service active
lingniu-go-history-writer.service failed
lingniu-go-gateway.service active enabled
lingniu-go-history-writer.service failed enabled
"""
checks = smoke.service_checks(output, smoke.DEFAULT_SERVICES)
@@ -28,6 +28,20 @@ lingniu-go-history-writer.service failed
self.assertEqual(by_name["service.lingniu-go-history-writer.service"].status, "fail")
self.assertEqual(by_name["service.lingniu-go-stat-writer.service"].status, "fail")
def test_service_checks_fail_when_unit_is_not_enabled(self):
output = """
lingniu-go-gateway.service active disabled
lingniu-go-history-writer.service active enabled
lingniu-go-stat-writer.service active enabled
lingniu-go-realtime-api.service active enabled
"""
checks = smoke.service_checks(output, smoke.DEFAULT_SERVICES)
by_name = {check.name: check for check in checks}
self.assertEqual(by_name["service.lingniu-go-gateway.service"].status, "fail")
self.assertIn("enabled=disabled", by_name["service.lingniu-go-gateway.service"].message)
def test_port_checks_require_expected_go_processes(self):
output = """
LISTEN 0 4096 *:32960 *:* users:(("gateway",pid=10,fd=3))