This repository has been archived by the owner on Sep 15, 2021. It is now read-only.
/
test_request_devices.py
117 lines (96 loc) · 4.75 KB
/
test_request_devices.py
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
import unittest
from mozharness.mozilla.testing.mozpool import MozpoolConflictException
from mozpoolmisc import TEST_ASSIGNEE, TEST_B2GBASE, BaseMozpoolTest
class RequestDevice(BaseMozpoolTest):
def _force_state(self, device, assignee, pxe_config, duration, desired_state):
dd = self.mph.query_device_details(device)
old_state = dd['state']
self.mph.device_state_change(device, assignee, pxe_config, duration, old_state, desired_state)
dd = self.mph.query_device_details(device)
new_state = dd['state']
self.assertEquals(desired_state, new_state)
def _test_query_all_requests(self):
""" Get a list of request. Expects at least one request to be active.
"""
# query all requests
response = self.mph.query_all_requests()
self.assertIsNotNone(response)
self.assertGreaterEqual(1, len(response))
for request in response['requests']:
for request_key in request.keys():
self.assertIn(request_key, ['assigned_device', 'assignee', 'boot_config', 'device_state',
'expires', 'id', 'imaging_server', 'requested_device', 'state'])
self.assertNotIn(request['state'], ['closed'])
response = self.mph.query_all_requests(include_closed_requests=True)
self.assertIsNotNone(response)
self.assertGreaterEqual(2, len(response))
for request in response['requests']:
for request_key in request.keys():
self.assertIn(request_key, ['assigned_device', 'assignee', 'boot_config', 'device_state',
'expires', 'id', 'imaging_server', 'requested_device', 'state'])
def _test_renew_request(self, request_url, old_expires):
""" Reset the lifetime of a request.
"""
self.mph.renew_request(request_url, 12)
response = self.mph.query_request_details(request_url)
expires = response['expires']
self.assertIsNotNone(expires)
self.assertNotEquals(expires, old_expires)
def _test_query_request_details(self, request_url):
""" Get request details. Expects at least one request to be active.
"""
response = self.mph.query_request_details(request_url)
self.assertIsNotNone(response)
for k in response.keys():
self.assertIn(k, ['assigned_device','assignee','boot_config','expires','id','requested_device','url'])
old_expires = response['expires']
self._test_renew_request(request_url, old_expires)
def _test_close_request(self, request_url):
""" Returns the device to the pool and deletes the request.
"""
self.mph.close_request(request_url)
response = self.mph.query_request_status(request_url)
self.assertEqual(response['state'], 'closed')
def test_request_any_device(self):
"""If 'any' device was requested, always returns 200 OK, since it will
retry a few times if no devices are free. If a specific device is requested
but is already assigned, returns 409 Conflict; otherwise, returns 200 OK.
"""
self.setup()
device = 'any'
duration = 15
image = 'b2g'
pxe_config = None
device_blob = self.mph.request_device(device, TEST_ASSIGNEE, image, duration, b2gbase=TEST_B2GBASE, pxe_config=pxe_config)
self.assertIsNotNone(device_blob)
request_url = device_blob['request']['url']
self.assertIsNotNone(request_url)
self.assertIn('http', request_url)
self._test_query_all_requests()
self._test_query_request_details(request_url)
self._test_close_request(request_url)
def test_request_conflicting_device(self):
self.setup()
device = 'any'
duration = 10
image = 'b2g'
pxe_config = None
device_blob = self.mph.request_device(device, TEST_ASSIGNEE, image, duration, b2gbase=TEST_B2GBASE, pxe_config=pxe_config)
self.assertIsNotNone(device_blob)
assigned_device = device_blob['request']['assigned_device']
device_blob['request']['url']
# try and request the same device
with self.assertRaises(MozpoolConflictException):
self.mph.request_device(assigned_device, TEST_ASSIGNEE, image, duration, b2gbase=TEST_B2GBASE, pxe_config=pxe_config)
def test_query_request_status(self):
self.setup()
device = 'any'
duration = 10
image = 'b2g'
pxe_config = None
device_blob = self.mph.request_device(device, TEST_ASSIGNEE, image, duration, b2gbase=TEST_B2GBASE, pxe_config=pxe_config)
request_url = device_blob['request']['url']
response = self.mph.query_request_status(request_url)
self.assertIsNotNone(response)
if __name__ == '__main__':
unittest.main()