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
|
import os
import random
import shelve
import testtools
import time
import requests
from requests_oauthlib import OAuth1
from config import MAAS_URL
# TODO: Get this from a config file/options
MAAS_API = MAAS_URL + '/api/2.0/'
APP_JSON_HEADER = {'Accept': 'application/json'}
def get_api_key():
return os.environ.get('MAAS_API_KEY', None)
class APITestCase(testtools.TestCase):
def setUp(self):
super(APITestCase, self).setUp()
self.shelf = shelve.open('tests.db')
api_key = get_api_key()
if api_key is not None:
self.consumer_key, self.key, self.secret = api_key.split(":")
self.oauth = OAuth1(
self.consumer_key,
resource_owner_key=self.key,
resource_owner_secret=self.secret,
signature_method='PLAINTEXT',
)
def _get_machines_status(self, machines):
machines_uri = MAAS_API + "machines/"
response = requests.get(
machines_uri, auth=self.oauth, headers=APP_JSON_HEADER
)
if response.ok:
return [machine['status'] for machine in response.json()
if machines is None or machine['system_id'] in machines]
else:
return None
def _wait_machines_in_state(self, state, machines=None):
while True:
statuses = self._get_machines_status(machines)
if statuses is not None:
if (len(statuses) > 0
and all([status == state for status in statuses])):
return True
time.sleep(30)
else:
return False # couldn't get statuses
def _wait_power_state(self, state, system_id):
uri = MAAS_API + "machines/{system_id}/?op=query_power_state".format(
system_id=system_id
)
while True:
response = requests.get(
uri, auth=self.oauth, headers=APP_JSON_HEADER
)
if response.ok and response.json()['state'] == state:
return True
time.sleep(30)
def random_mac(self):
return "52:54:00:{:02x}:{:02x}:{:02x}".format(
random.randint(0, 255),
random.randint(0, 255),
random.randint(0, 255))
|