1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
#!/usr/bin/env python3
import subprocess
import unittest
SERVICE_PMCD = "pmcd.service"
SERVICE_PMLOGGER = "pmlogger.service"
PORTS_PMCD = [44321]
PORTS_PMLOGGER = [4330]
PROP_ACTIVESTATE = "ActiveState"
PROP_SUBSTATE = "SubState"
STATE_RUNNING = "running"
STATE_ACTIVE = "active"
def format_args(service, property):
return ["systemctl", "show", "-p" + property, service]
def search_output(output, startsWith):
for line in output.splitlines():
if line.startswith(startsWith + "="):
return line[(len(startsWith)+1):]
def check_port(port):
import socket
s = socket.socket()
address = '127.0.0.1'
try:
s.connect((address, port))
except:
return False
finally:
s.close()
return True
class TestDaemons(unittest.TestCase):
def check_property(self, service, property, expected):
output = subprocess.check_output(
format_args(service, property)).decode("utf-8")
self.assertEqual(search_output(output, property), expected)
def test_pmcd(self):
self.check_property(SERVICE_PMCD, PROP_ACTIVESTATE, STATE_ACTIVE)
self.check_property(SERVICE_PMCD, PROP_SUBSTATE, STATE_RUNNING)
for port in PORTS_PMCD:
self.assertTrue(check_port(port))
def test_pmlogger(self):
self.check_property(SERVICE_PMLOGGER, PROP_ACTIVESTATE, STATE_ACTIVE)
self.check_property(SERVICE_PMLOGGER, PROP_SUBSTATE, STATE_RUNNING)
for port in PORTS_PMLOGGER:
self.assertTrue(check_port(port))
if __name__ == "__main__":
import sys
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
verbosity=2))
|
