84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
import unittest
|
|
|
|
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 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)
|
|
|
|
self.assertTrue(all(check.status == "pass" for check in checks))
|
|
|
|
def test_service_checks_fail_when_unit_is_missing_or_inactive(self):
|
|
output = """
|
|
lingniu-go-gateway.service active enabled
|
|
lingniu-go-history-writer.service failed 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-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))
|
|
LISTEN 0 4096 *:808 *:* users:(("gateway",pid=10,fd=4))
|
|
LISTEN 0 4096 *:20210 *:* users:(("realtime-api",pid=11,fd=3))
|
|
"""
|
|
|
|
checks = smoke.port_checks(output, smoke.DEFAULT_PORTS)
|
|
|
|
self.assertTrue(all(check.status == "pass" for check in checks))
|
|
|
|
def test_port_checks_fail_when_java_owns_ingest_port(self):
|
|
output = """
|
|
LISTEN 0 4096 *:808 *:* users:(("java",pid=20,fd=4))
|
|
LISTEN 0 4096 *:32960 *:* users:(("gateway",pid=10,fd=3))
|
|
LISTEN 0 4096 *:20210 *:* users:(("realtime-api",pid=11,fd=3))
|
|
"""
|
|
|
|
checks = smoke.port_checks(output, smoke.DEFAULT_PORTS)
|
|
by_name = {check.name: check for check in checks}
|
|
|
|
self.assertEqual(by_name["port.808"].status, "fail")
|
|
self.assertIn("java", by_name["port.808"].message)
|
|
|
|
def test_spool_check_passes_when_no_pending_files(self):
|
|
check = smoke.spool_check("files=0 bytes=208896 recent=0")
|
|
|
|
self.assertEqual(check.status, "pass")
|
|
self.assertIn("files=0", check.message)
|
|
|
|
def test_spool_check_fails_when_pending_or_recent_files_exist(self):
|
|
for output in ["files=2 bytes=300 recent=0", "files=0 bytes=208896 recent=1"]:
|
|
check = smoke.spool_check(output)
|
|
|
|
self.assertEqual(check.status, "fail")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|