-
Notifications
You must be signed in to change notification settings - Fork 159
/
hawkbit.py
205 lines (169 loc) · 6.57 KB
/
hawkbit.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
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
import attr
import requests as r
@attr.s(eq=False)
class HawkbitTestClient:
host = attr.ib(validator=attr.validators.instance_of(str))
port = attr.ib(validator=attr.validators.instance_of(str))
username = attr.ib(validator=attr.validators.instance_of(str))
password = attr.ib(validator=attr.validators.instance_of(str))
version = attr.ib(
default=1.0, validator=attr.validators.instance_of(float)
)
def __attrs_post_init__(self):
self.url = 'http://{host}:{port}/rest/v1/{endpoint}'
def add_target(self, target_id: str, token: str):
"""Add a target to the HawkBit Installation
Arguments:
- target_id(str): the (unique) device name of the target to add
- token(str): pre-shared key to authenticate the target
"""
testdata = [{
'controllerId': target_id,
'name': target_id,
'description': 'test',
'securityToken': token
}]
self.post_json('targets', testdata)
def delete_target(self, target_id: str):
"""Delete a target from the HawkBit Installation
Arguments:
- target_id(str): the (unique) device name of the target to delete
"""
self.delete('targets/{}'.format(target_id))
def add_swmodule(self, modulename: str):
testdata = [{
'name': modulename,
'version': str(self.version),
'type': 'os'
}]
self.module_id = self.post_json('softwaremodules', testdata)[0]['id']
return self.module_id
def delete_swmodule(self, module_id: str):
"""Delete a softwaremodule from the HawkBit Installation
Arguments:
- module_id(str): the ID given by hawkBit for the module
"""
self.delete('softwaremodules/{}'.format(module_id))
def add_distributionset(self, module_id, name=None):
testdata = [{
'name': name if name else 'Test_distribution {}'.format(self.version),
'description': 'Test distribution',
'version': str(self.version),
'modules': [{
'id': module_id
}],
'type': 'os'
}]
self.distribution_id = self.post_json('distributionsets',
testdata)[0]['id']
return self.distribution_id
def delete_distributionset(self, distset_id: str):
"""Delete a distrubitionset from the HawkBit Installation
Arguments:
- distset_id(str): the ID of the distribution set to delete
"""
self.delete('distributionsets/{}'.format(distset_id))
def add_artifact(self, module_id: str, filename: str):
endpoint = 'softwaremodules/{}/artifacts'.format(module_id)
return self.post_binary(endpoint, filename)['id']
def delete_artifact(self, module_id: str, artifact_id: str):
"""Delete an artifact from the HawkBit Installation
Arguments:
- artifact_id(str): the ID of the artifact to delete
"""
self.delete('softwaremodules/{}/artifacts/{}'.format(module_id, artifact_id))
def assign_target(self, distribution_id, target_id):
endpoint = 'distributionsets/{}/assignedTargets'.format(
distribution_id
)
testdata = [{'id': target_id}]
self.post_json(endpoint, testdata)
# Increment version to be able to flash over an already deployed distribution
self.version = self.version + 0.1
def add_rollout(self, name, distribution_id, groups):
testdata = {
'name': name,
'distributionSetId': distribution_id,
'targetFilterQuery': 'id==*',
'amountGroups': groups
}
self.rollout_id = self.post_json('rollouts', testdata)['id']
return self.rollout_id
def start_rollout(self, rollout_id):
endpoint = 'rollouts/{}/start'.format(
rollout_id
)
self.post(endpoint)
def post(self, endpoint: str):
req = r.post(
self.url.format(
endpoint=endpoint, host=self.host, port=self.port
),
auth=(self.username, self.password),
)
if req.status_code != 200 and req.status_code != 201:
raise HawkbitError(
'Wrong statuscode, got {} instead of 200/201'.
format(req.status_code)
)
def post_json(self, endpoint: str, data: dict):
headers = {'Content-Type': 'application/json;charset=UTF-8'}
req = r.post(
self.url.format(
endpoint=endpoint, host=self.host, port=self.port
),
headers=headers,
auth=(self.username, self.password),
json=data
)
if req.status_code != 200 and req.status_code != 201:
raise HawkbitError(
'Wrong statuscode, got {} instead of 200/201, with error {}'.
format(req.status_code, req.json())
)
return req.json()
def post_binary(self, endpoint: str, filename: str):
files = {'file': open(filename, 'rb')}
req = r.post(
self.url.format(
endpoint=endpoint, host=self.host, port=self.port
),
auth=(self.username, self.password),
files=files
)
if req.status_code != 201:
raise HawkbitError(
'Wrong statuscode, got {} instead of 201, with error {}'.
format(req.status_code, req.json())
)
return req.json()
def delete(self, endpoint: str):
req = r.delete(
self.url.format(
endpoint=endpoint, host=self.host, port=self.port
),
auth=(self.username, self.password),
)
if req.status_code != 200:
raise HawkbitError(
'Wrong statuscode, got {} instead of 200, with error {}'.
format(req.status_code, req.json())
)
def get_endpoint(self, endpoint: str):
headers = {'Content-Type': 'application/json;charset=UTF-8'}
req = r.get(
self.url.format(
endpoint=endpoint, host=self.host, port=self.port
),
headers=headers,
auth=(self.username, self.password)
)
if req.status_code != 200:
raise HawkbitError(
'Wrong statuscode, got {} instead of 200, with error {}'.
format(req.status_code, req.json())
)
return req.json()
@attr.s(eq=False)
class HawkbitError(Exception):
msg = attr.ib()