1 # ============LICENSE_START===============================================
2 # Copyright (C) 2021 Nordix Foundation. All rights reserved.
3 # ========================================================================
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 # ============LICENSE_END=================================================
18 # This test case test the STD_2.0.0 version of the simulator
24 INTERFACE_VERSION="STD_2.0.0"
26 from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client
28 #Setup env and import paths
29 setup_env(INTERFACE_VERSION)
31 from compare_json import compare
33 def test_apis(client):
35 testdata=get_testdata_dir()
37 # Header for json payload
39 "Content-Type" : "application/json"
42 # Simulator hello world
43 response=client.get(SERVER_URL)
44 assert response.status_code == 200
46 # Check used and implemented interfaces
47 response=client.get(SERVER_URL+'container_interfaces')
48 assert response.status_code == 200
49 assert response.data == b"Current interface: STD_2.0.0 All supported A1 interface yamls in this container: ['OSC_2.1.0', 'STD_1.1.3', 'STD_2.0.0']"
51 # Reset simulator instances
52 response=client.post(SERVER_URL+'deleteinstances')
53 assert response.status_code == 200
55 # Reset simulator, all
56 response=client.post(SERVER_URL+'deleteall')
57 assert response.status_code == 200
59 # Get counter: interface
60 response=client.get(SERVER_URL+'counter/interface')
61 assert response.status_code == 200
62 assert response.data == b"STD_2.0.0"
64 # Get counter: remote hosts
65 response=client.get(SERVER_URL+'counter/remote_hosts')
66 assert response.status_code == 200
68 # Get counter: intstance
69 response=client.get(SERVER_URL+'counter/num_instances')
70 assert response.status_code == 200
71 assert response.data == b"0"
74 response=client.get(SERVER_URL+'counter/num_types')
75 assert response.status_code == 200
76 assert response.data == b"0"
78 # API: Get policy type, shall be empty
80 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
81 assert response.status_code == 200
82 result=json.loads(response.data)
83 res=compare(data_response, result)
86 # API: Get policy instances for type 1, type not found
87 data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "1"}
88 response=client.get(SERVER_URL+'A1-P/v2/policytypes/1/policies')
89 assert response.status_code == 404
90 result=json.loads(response.data)
91 res=compare(data_response, result)
94 # API: Get policy instances, type not found
95 data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "test"}
96 response=client.get(SERVER_URL+'A1-P/v2/policytypes/test/policies')
97 assert response.status_code == 404
98 result=json.loads(response.data)
99 res=compare(data_response, result)
102 # Put a policy type: STD_1
103 with open(testdata+'std_1.json') as json_file:
104 data_response = b"Policy type STD_1 is OK."
105 json_payload=json.load(json_file)
106 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
107 assert response.status_code == 201
108 assert data_response == response.data
110 # Put a policy type: STD_1, again
111 with open(testdata+'std_1.json') as json_file:
112 data_response = b"Policy type STD_1 is OK."
113 json_payload=json.load(json_file)
114 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
115 assert response.status_code == 200
116 assert data_response == response.data
118 # API: Get policy type ids, shall contain type STD_1
119 data_response = [ "STD_1" ]
120 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
121 assert response.status_code == 200
122 result=json.loads(response.data)
123 res=compare(data_response, result)
126 # Delete a policy type: STD_1
128 response=client.delete(SERVER_URL+'policytype?id=STD_1')
129 assert response.status_code == 204
130 assert data_response == response.data
132 # API: Get policy type ids, shall be empty
134 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
135 assert response.status_code == 200
136 result=json.loads(response.data)
137 res=compare(data_response, result)
140 # Put a policy type: STD_1
141 with open(testdata+'std_1.json') as json_file:
142 data_response = b"Policy type STD_1 is OK."
143 json_payload=json.load(json_file)
144 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
145 assert response.status_code == 201
146 assert data_response == response.data
148 # API: Get policy type ids, shall contain type STD_1
149 data_response = [ "STD_1" ]
150 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
151 assert response.status_code == 200
152 result=json.loads(response.data)
153 res=compare(data_response, result)
156 # Get counter: types (shall be 1)
157 response=client.get(SERVER_URL+'counter/num_types')
158 assert response.status_code == 200
159 assert response.data == b"1"
161 # API: Get policy type: STD_1
162 with open(testdata+'std_1.json') as json_file:
163 data_response = json.load(json_file)
164 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1')
165 assert response.status_code == 200
166 result=json.loads(response.data)
167 res=compare(data_response, result)
170 # API: API: Get policy instances, shall be empty
172 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
173 assert response.status_code == 200
174 result=json.loads(response.data)
175 res=compare(data_response, result)
178 # API: Create policy instance pi1 of type: STD_1
179 with open(testdata+'pi1.json') as json_file:
180 json_payload=json.load(json_file)
181 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
182 assert response.status_code == 201
183 result=json.loads(response.data)
184 res=compare(json_payload, result)
187 # API: API: Get policy instance pi1 of type: STD_1
188 with open(testdata+'pi1.json') as json_file:
189 data_response = json.load(json_file)
190 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
191 assert response.status_code == 200
192 result=json.loads(response.data)
193 res=compare(data_response, result)
196 # API: Update policy instance pi1 of type: STD_1
197 with open(testdata+'pi1.json') as json_file:
198 json_payload=json.load(json_file)
199 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
200 assert response.status_code == 200
201 result=json.loads(response.data)
202 res=compare(json_payload, result)
205 # API: Update policy instance pi1 of type: STD_1
206 with open(testdata+'pi1_updated.json') as json_file:
207 json_payload=json.load(json_file)
208 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
209 assert response.status_code == 200
210 result=json.loads(response.data)
211 res=compare(json_payload, result)
214 # # API: Duplicate policy instance pi2 of type: STD_1 - and delete it
216 with open(testdata+'pi1_updated.json') as json_file:
217 json_payload=json.load(json_file)
218 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
219 assert response.status_code == 201
220 result=json.loads(response.data)
221 res=compare(json_payload, result)
224 response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
225 assert response.status_code == 204
226 assert data_response == response.data
229 # API: Get policy instances, shall contain pi1
230 data_response = ["pi1"]
231 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
232 assert response.status_code == 200
233 result=json.loads(response.data)
234 res=compare(data_response, result)
237 # Get counter: intstance
238 response=client.get(SERVER_URL+'counter/num_instances')
239 assert response.status_code == 200
240 assert response.data == b"1"
243 response=client.get(SERVER_URL+'counter/num_types')
244 assert response.status_code == 200
245 assert response.data == b"1"
247 # Set force response code 409. ==="
248 response=client.post(SERVER_URL+'forceresponse?code=409')
249 assert response.status_code == 200
251 # API: Get policy instances, shall fail
252 data_response = {"title" : "Conflict", "status" : 409, "detail" : "Request could not be processed in the current state of the resource"}
253 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
254 assert response.status_code == 409
255 result=json.loads(response.data)
256 res=compare(data_response, result)
259 # API: API: Get policy status
260 data_response = {"enforceStatus" : "", "enforceReason" : ""}
261 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1/status')
262 assert response.status_code == 200
263 result=json.loads(response.data)
264 res=compare(data_response, result)
267 # API: Create policy instance pi2 of type: STD_1
268 with open(testdata+'pi2.json') as json_file:
269 json_payload=json.load(json_file)
270 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
271 assert response.status_code == 201
272 result=json.loads(response.data)
273 res=compare(json_payload, result)
276 # API: Update policy instance pi2 of type: STD_1
277 with open(testdata+'pi2.json') as json_file:
278 json_payload=json.load(json_file)
279 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
280 assert response.status_code == 200
281 result=json.loads(response.data)
282 res=compare(json_payload, result)
285 # API: Get policy instances, shall contain pi1 and pi2
286 data_response = ["pi1","pi2"]
287 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
288 assert response.status_code == 200
289 result=json.loads(response.data)
290 res=compare(data_response, result)
293 # Get counter: intstance
294 response=client.get(SERVER_URL+'counter/num_instances')
295 assert response.status_code == 200
296 assert response.data == b"2"
299 response=client.get(SERVER_URL+'counter/num_types')
300 assert response.status_code == 200
301 assert response.data == b"1"
304 response=client.post(SERVER_URL+'forcedelay?delay=10')
305 assert response.status_code == 200
306 assert response.data == b"Force delay: 10 sec set for all A1 responses"
311 # API: Get policy instances, shall contain pi1 and pi2 and delayed 10 sec
312 data_response = ["pi1","pi2"]
313 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
314 assert response.status_code == 200
315 result=json.loads(response.data)
316 res=compare(data_response, result)
321 assert (end-start) > 9
324 response=client.post(SERVER_URL+'forcedelay')
325 assert response.status_code == 200
326 assert response.data == b"Force delay: None sec set for all A1 responses"
328 # API: API: Get policy instance pi1 of type: STD_1
329 with open(testdata+'pi1_updated.json') as json_file:
330 data_response = json.load(json_file)
331 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
332 assert response.status_code == 200
333 result=json.loads(response.data)
334 res=compare(data_response, result)
337 # API: API: Get policy instance pi2 of type: STD_1
338 with open(testdata+'pi2.json') as json_file:
339 data_response = json.load(json_file)
340 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
341 assert response.status_code == 200
342 result=json.loads(response.data)
343 res=compare(data_response, result)
346 # API: DELETE policy instance pi1
348 response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
349 assert response.status_code == 204
350 assert data_response == response.data
352 # API: Get policy instances, shall contain pi2
353 data_response = ["pi2"]
354 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
355 assert response.status_code == 200
356 result=json.loads(response.data)
357 res=compare(data_response, result)
360 # API: API: Get policy status
361 data_response = {"enforceStatus" : "", "enforceReason" : ""}
362 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
363 assert response.status_code == 200
364 result=json.loads(response.data)
365 res=compare(data_response, result)
368 # Set status for policy instance pi2
369 response=client.put(SERVER_URL+'status?policyid=pi2&status=OK')
370 assert response.status_code == 200
372 # API: API: Get policy status
373 data_response = {"enforceStatus" : "OK"}
374 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
375 assert response.status_code == 200
376 result=json.loads(response.data)
377 res=compare(data_response, result)
381 # Set status for policy instance pi2
382 response=client.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason')
383 assert response.status_code == 200
385 # API: API: Get policy status
386 data_response = {"enforceStatus" : "NOTOK", "enforceReason" : "notok_reason"}
387 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
388 assert response.status_code == 200
389 result=json.loads(response.data)
390 res=compare(data_response, result)
394 # #Found no way to test these functions
395 # #'sendstatus' will send a http request that will fail
396 # #since no server will receive the call
397 # #These function is instead tested when running the bash script in the 'test' dir
398 # # # Send status for pi2
399 # # response=client.post(SERVER_URL+'sendstatus?policyid=pi2')
400 # # assert response.status_code == 200
401 # # result=json.loads(response.data)
402 # # res=compare(data_get_status, result)
403 # # assert res == True
405 # # # Send status, shall fail
406 # # response=client.post(SERVER_URL+'sendstatus')
407 # # assert response.status_code == 400
409 # Get counter: data_delivery
410 response=client.get(SERVER_URL+'counter/datadelivery')
411 assert response.status_code == 200
412 assert response.data == b"0"
416 response=client.post(SERVER_URL+'datadelivery', headers=header, data=json.dumps(json_payload))
417 assert response.status_code == 200
419 # Get counter: data_delivery
420 response=client.get(SERVER_URL+'counter/datadelivery')
421 assert response.status_code == 200
422 assert response.data == b"1"
424 # Get counter: interface
425 response=client.get(SERVER_URL+'counter/interface')
426 assert response.status_code == 200
427 assert response.data == b"STD_2.0.0"
429 # Get counter: remote hosts
430 response=client.get(SERVER_URL+'counter/remote_hosts')
431 assert response.status_code == 200
433 # Get counter: intstance
434 response=client.get(SERVER_URL+'counter/num_instances')
435 assert response.status_code == 200
436 assert response.data == b"1"
439 response=client.get(SERVER_URL+'counter/num_types')
440 assert response.status_code == 200
441 assert response.data == b"1"