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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import NoReturn
from testinfra.modules.base import Module
STATUS = [
"STOPPED",
"STARTING",
"RUNNING",
"BACKOFF",
"STOPPING",
"EXITED",
"FATAL",
"UNKNOWN",
]
def supervisor_not_running() -> NoReturn:
raise RuntimeError("Cannot get supervisor status. Is supervisor running ?")
class Supervisor(Module):
"""Test supervisor managed services
>>> gunicorn = host.supervisor("gunicorn")
>>> gunicorn.status
'RUNNING'
>>> gunicorn.is_running
True
>>> gunicorn.pid
4242
The path where supervisorctl and its configuration file reside can be specified.
>>> gunicorn = host.supervisor("gunicorn", "/usr/bin/supervisorctl", "/etc/supervisor/supervisord.conf")
>>> gunicorn.status
'RUNNING'
"""
def __init__(
self,
name,
supervisorctl_path="supervisorctl",
supervisorctl_conf=None,
_attrs_cache=None,
):
self.name = name
self.supervisorctl_path = supervisorctl_path
self.supervisorctl_conf = supervisorctl_conf
self._attrs_cache = _attrs_cache
super().__init__()
@staticmethod
def _parse_status(line):
splitted = line.split()
name = splitted[0]
status = splitted[1]
# some old supervisorctl versions exit status is 0 even if it cannot
# connect to supervisord socket and output the error to stdout. So we
# check that parsed status is a known status.
if status not in STATUS:
supervisor_not_running()
pid = int(splitted[3].removesuffix(",")) if status == "RUNNING" else None
return {"name": name, "status": status, "pid": pid}
@property
def _attrs(self):
if self._attrs_cache is None:
if self.supervisorctl_conf:
out = self.run_expect(
[0, 3, 4],
"%s -c %s status %s",
self.supervisorctl_path,
self.supervisorctl_conf,
self.name,
)
else:
out = self.run_expect(
[0, 3, 4], "%s status %s", self.supervisorctl_path, self.name
)
if out.rc == 4:
supervisor_not_running()
line = out.stdout.rstrip("\r\n")
attrs = self._parse_status(line)
assert attrs["name"] == self.name
self._attrs_cache = attrs
return self._attrs_cache
@property
def is_running(self):
"""Return True if managed service is in status RUNNING"""
return self.status == "RUNNING"
@property
def status(self):
"""Return the status of the managed service
Status can be STOPPED, STARTING, RUNNING, BACKOFF, STOPPING,
EXITED, FATAL, UNKNOWN.
See http://supervisord.org/subprocess.html#process-states
"""
return self._attrs["status"]
@property
def pid(self):
"""Return the pid (as int) of the managed service"""
return self._attrs["pid"]
@classmethod
def get_services(
cls,
supervisorctl_path="supervisorctl",
supervisorctl_conf=None,
):
"""Get a list of services running under supervisor
>>> host.supervisor.get_services()
[<Supervisor(name="gunicorn", status="RUNNING", pid=4232)>
<Supervisor(name="celery", status="FATAL", pid=None)>]
The path where supervisorctl and its configuration file reside can be specified.
>>> host.supervisor.get_services("/usr/bin/supervisorctl", "/etc/supervisor/supervisord.conf")
[<Supervisor(name="gunicorn", status="RUNNING", pid=4232)>
<Supervisor(name="celery", status="FATAL", pid=None)>]
"""
services = []
if supervisorctl_conf:
out = cls.check_output(
"%s -c %s status", supervisorctl_path, supervisorctl_conf
)
else:
out = cls.check_output("%s status", supervisorctl_path)
for line in out.splitlines():
attrs = cls._parse_status(line)
service = cls(
attrs["name"],
supervisorctl_path=supervisorctl_path,
supervisorctl_conf=supervisorctl_conf,
_attrs_cache=attrs,
)
services.append(service)
return services
def __repr__(self):
return f"<Supervisor(name={self.name}, status={self.status}, pid={self.pid})>"
|
