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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
|
from urllib.parse import urlparse
import pytest
from labgrid.resource import NetworkPowerPort
from labgrid.driver.powerdriver import ExternalPowerDriver, ManualPowerDriver, NetworkPowerDriver
class TestManualPowerDriver:
def test_create(self, target):
d = ManualPowerDriver(target, 'foo-board')
assert (isinstance(d, ManualPowerDriver))
def test_on(self, target, mocker):
m = mocker.patch('builtins.input')
d = ManualPowerDriver(target, 'foo-board')
target.activate(d)
d.on()
m.assert_called_once_with(
"Turn the target Test ON and press enter"
)
def test_off(self, target, mocker):
m = mocker.patch('builtins.input')
d = ManualPowerDriver(target, 'foo-board')
target.activate(d)
d.off()
m.assert_called_once_with(
"Turn the target Test OFF and press enter"
)
def test_cycle(self, target, mocker):
m = mocker.patch('builtins.input')
d = ManualPowerDriver(target, 'foo-board')
target.activate(d)
d.cycle()
m.assert_called_once_with("CYCLE the target Test and press enter")
class TestExternalPowerDriver:
def test_create(self, target):
d = ExternalPowerDriver(
target, 'power', cmd_on='set -1 foo-board', cmd_off='set -0 foo-board'
)
assert (isinstance(d, ExternalPowerDriver))
def test_on(self, target, mocker):
pty = mocker.patch('pty.openpty')
mocker.patch('fcntl.fcntl')
fdopen = mocker.patch('os.fdopen')
close = mocker.patch('os.close')
popen = mocker.patch('subprocess.Popen')
select = mocker.patch('select.select')
fd_mock = mocker.MagicMock()
instance_mock = mocker.MagicMock()
select_mock = mocker.MagicMock()
popen.return_value = instance_mock
fdopen.return_value = fd_mock
pty.return_value = (instance_mock, 2)
fd_mock.read.return_value = b'Done\nDone'
instance_mock.returncode = 0
select.return_value = ([], [], [])
d = ExternalPowerDriver(
target, 'power', cmd_on='set -1 foo-board', cmd_off='set -0 foo-board'
)
target.activate(d)
d.on()
popen.assert_called_once_with(['set', '-1', 'foo-board'], bufsize=0,
stderr=2, stdout=2)
def test_off(self, target, mocker):
pty = mocker.patch('pty.openpty')
mocker.patch('fcntl.fcntl')
fdopen = mocker.patch('os.fdopen')
close = mocker.patch('os.close')
popen = mocker.patch('subprocess.Popen')
select = mocker.patch('select.select')
fd_mock = mocker.MagicMock()
instance_mock = mocker.MagicMock()
select_mock = mocker.MagicMock()
popen.return_value = instance_mock
fdopen.return_value = fd_mock
pty.return_value = (instance_mock, 2)
fd_mock.read.return_value = b'Done\nDone'
instance_mock.returncode = 0
select.return_value = ([], [], [])
d = ExternalPowerDriver(
target, 'power', cmd_on='set -1 foo-board', cmd_off='set -0 foo-board'
)
target.activate(d)
d.off()
popen.assert_called_once_with(['set', '-0', 'foo-board'], bufsize=0,
stderr=2, stdout=2)
def test_cycle(self, target, mocker):
pty = mocker.patch('pty.openpty')
mocker.patch('fcntl.fcntl')
fdopen = mocker.patch('os.fdopen')
close = mocker.patch('os.close')
popen = mocker.patch('subprocess.Popen')
select = mocker.patch('select.select')
fd_mock = mocker.MagicMock()
instance_mock = mocker.MagicMock()
select_mock = mocker.MagicMock()
popen.return_value = instance_mock
fdopen.return_value = fd_mock
pty.return_value = (instance_mock, 2)
fd_mock.read.return_value = b'Done\nDone'
instance_mock.returncode = 0
select.return_value = ([], [], [])
m_sleep = mocker.patch('time.sleep')
d = ExternalPowerDriver(
target, 'power', cmd_on='set -1 foo-board', cmd_off='set -0 foo-board'
)
target.activate(d)
d.cycle()
assert popen.call_args_list == [
mocker.call(['set', '-0', 'foo-board'], bufsize=0, stderr=2,
stdout=2),
mocker.call(['set', '-1', 'foo-board'], bufsize=0, stderr=2,
stdout=2),
]
m_sleep.assert_called_once_with(2.0)
def test_cycle_explicit(self, target, mocker):
pty = mocker.patch('pty.openpty')
mocker.patch('fcntl.fcntl')
fdopen = mocker.patch('os.fdopen')
close = mocker.patch('os.close')
popen = mocker.patch('subprocess.Popen')
select = mocker.patch('select.select')
fd_mock = mocker.MagicMock()
instance_mock = mocker.MagicMock()
select_mock = mocker.MagicMock()
popen.return_value = instance_mock
fdopen.return_value = fd_mock
pty.return_value = (instance_mock, 2)
fd_mock.read.return_value = b'Done\nDone'
instance_mock.returncode = 0
select.return_value = ([], [], [])
m_sleep = mocker.patch('time.sleep')
d = ExternalPowerDriver(
target, 'power',
cmd_on='set -1 foo-board',
cmd_off='set -0 foo-board',
cmd_cycle='set -c foo-board',
)
target.activate(d)
d.cycle()
popen.assert_called_once_with(['set', '-c', 'foo-board'], bufsize=0,
stderr=2, stdout=2)
m_sleep.assert_not_called()
class TestNetworkPowerDriver:
def test_create(self, target):
r = NetworkPowerPort(target, 'power', model='netio', host='dummy', index='1')
d = NetworkPowerDriver(target, 'power')
assert isinstance(d, NetworkPowerDriver)
@pytest.mark.parametrize('backend', ('rest', 'simplerest'))
@pytest.mark.parametrize(
'host',
(
'http://example.com/{index}',
'https://example.com/{index}',
'http://example.com:1234/{index}',
'https://example.com:1234/{index}',
'http://user:pass@example.com:1234/{index}',
'https://user:pass@example.com:1234/{index}',
)
)
def test_create_backend_with_url_in_host(self, target, mocker, backend, host):
get = mocker.patch('requests.get')
get.return_value.text = '1'
mocker.patch('requests.put')
index = '1'
NetworkPowerPort(target, 'power', model=backend, host=host, index=index)
d = NetworkPowerDriver(target, 'power')
assert isinstance(d, NetworkPowerDriver)
target.activate(d)
d.cycle()
assert d.get() is True
# the called URL should be similar to the one configured in the resource, but with
# index and explicit port
expected_host = host.format(index=index)
url = urlparse(expected_host)
if url.port is None:
implicit_port = 443 if url.scheme == 'https' else 80
expected_host = expected_host.replace(url.netloc, f'{url.netloc}:{implicit_port}')
get.assert_called_with(expected_host)
@pytest.mark.parametrize(
'host',
(
'http://example.com',
'https://example.com',
)
)
def test_create_shelly_gen1_backend_with_url_in_host(self, target, mocker, host):
get = mocker.patch('requests.get')
get.return_value.text = '{"ison": true}'
mocker.patch('requests.post')
index = '0'
NetworkPowerPort(target, 'power', model='shelly_gen1', host=host, index=index)
d = NetworkPowerDriver(target, 'power')
target.activate(d)
d.cycle()
assert d.get() is True
# the called URL should be similar to the one configured in the resource, but with
# index and explicit port
expected_host = f"{host}/relay/{index}"
url = urlparse(expected_host)
if url.port is None:
implicit_port = 443 if url.scheme == "https" else 80
expected_host = expected_host.replace(
url.netloc, f"{url.netloc}:{implicit_port}"
)
get.assert_called_with(expected_host)
def test_create_ubus_backend(self, target, mocker):
post = mocker.patch("requests.post")
post.return_value.json.return_value = {
"jsonrpc": "2.0",
"id": 1,
"result": [
0,
{
"firmware": "v80.1",
"budget": 77.000000,
"consumption": 1.700000,
"ports": {
"lan1": {
"priority": 0,
"mode": "PoE+",
"status": "Delivering power",
"consumption": 1.700000,
}
},
},
],
}
index = "1"
NetworkPowerPort(
target, "power", model="ubus", host="http://example.com/ubus", index=index
)
d = NetworkPowerDriver(target, "power")
target.activate(d)
d.cycle()
assert d.get() is True
def test_import_backends(self):
import labgrid.driver.power
import labgrid.driver.power.apc
import labgrid.driver.power.digipower
import labgrid.driver.power.digitalloggers_http
import labgrid.driver.power.digitalloggers_restapi
import labgrid.driver.power.eth008
import labgrid.driver.power.gude
import labgrid.driver.power.gude24
import labgrid.driver.power.netio
import labgrid.driver.power.netio_kshell
import labgrid.driver.power.rest
import labgrid.driver.power.sentry
import labgrid.driver.power.eg_pms2_network
import labgrid.driver.power.shelly_gen1
import labgrid.driver.power.ubus
def test_import_backend_eaton(self):
pytest.importorskip("pysnmp")
import labgrid.driver.power.eaton
def test_import_backend_tplink(self):
pytest.importorskip("kasa")
import labgrid.driver.power.tplink
def test_import_backend_siglent(self):
pytest.importorskip("vxi11")
import labgrid.driver.power.siglent
def test_import_backend_poe_mib(self):
pytest.importorskip("pysnmp")
import labgrid.driver.power.poe_mib
|
