1 # ============LICENSE_START===============================================
2 # Copyright (C) 2021-2023 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 tests the OSC_2.1.0 version of the simulator
21 # Version of simulator
22 INTERFACE_VERSION="OSC_2.1.0"
26 import multiprocessing
27 from unittest_setup import SERVER_URL, PORT_NUMBER, setup_env, get_testdata_dir, client
28 from unittest_setup import run_flask_app
30 # Setup env and import paths
31 setup_env(INTERFACE_VERSION)
33 from compare_json import compare
34 from models.enforceStatus import EnforceStatus
36 def test_enforce_reason(client):
38 Test that we can set a valid enforce status and reason, and that we reject invalid cases.
40 enforceStatus = EnforceStatus()
42 enforceStatus.enforce_status = 'NOT_ENFORCED'
43 enforceStatus.enforce_reason = 'SCOPE_NOT_APPLICABLE'
44 enforce_dict = enforceStatus.to_dict()
45 assert enforce_dict['enforceStatus'] == 'NOT_ENFORCED'
46 assert enforce_dict['enforceReason'] == 'SCOPE_NOT_APPLICABLE'
48 enforceStatus.enforce_status = 'ENFORCED'
49 enforceStatus.enforce_reason = 'STATEMENT_NOT_APPLICABLE'
50 enforce_dict = enforceStatus.to_dict()
51 assert enforce_dict['enforceStatus'] == 'ENFORCED'
52 assert enforce_dict['enforceReason'] == 'STATEMENT_NOT_APPLICABLE'
54 enforceStatus.enforce_reason = 'OTHER_REASON'
55 enforce_dict = enforceStatus.to_dict()
56 assert enforce_dict['enforceReason'] == 'OTHER_REASON'
58 enforce_status = enforceStatus.enforce_status
59 assert str(enforce_status) == 'ENFORCED'
61 enforce_reason = enforceStatus.enforce_reason
62 assert str(enforce_reason) == 'OTHER_REASON'
64 with pytest.raises(ValueError):
65 enforceStatus.enforce_status = 'ERROR'
67 with pytest.raises(ValueError):
68 enforceStatus.enforce_reason = 'ERROR'
71 def test_apis(client):
73 testdata=get_testdata_dir()
75 # Simulator hello world
76 response=client.get(SERVER_URL)
77 assert response.status_code == 200
79 # Check used and implemented interfaces
80 response=client.get(SERVER_URL+'container_interfaces')
81 assert response.status_code == 200
82 assert response.data == b"Current interface: OSC_2.1.0 All supported A1 interface yamls in this container: ['OSC_2.1.0', 'STD_1.1.3', 'STD_2.0.0']"
84 # Reset simulator instances
85 response=client.post(SERVER_URL+'deleteinstances')
86 assert response.status_code == 200
88 # Reset simulator, all
89 response=client.post(SERVER_URL+'deleteall')
90 assert response.status_code == 200
93 response=client.get(SERVER_URL+'a1-p/healthcheck')
94 assert response.status_code == 200
96 # API: Get policy types, shall be empty
97 data_policytypes_get = [ ]
98 response=client.get(SERVER_URL+'a1-p/policytypes')
99 assert response.status_code == 200
100 result=json.loads(response.data)
101 res=compare(data_policytypes_get, result)
104 # API: Delete a policy type, shall fail
105 response=client.delete(SERVER_URL+'a1-p/policytypes/1')
106 assert response.status_code == 404
108 # API: Get policy instances for type 1, shall fail
109 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
110 assert response.status_code == 404
112 # Header for json payload
114 "Content-Type" : "application/json"
117 # API: Put a policy type: 1
118 with open(testdata+'pt1.json') as json_file:
119 policytype_1 = json.load(json_file)
120 response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
121 assert response.status_code == 201
123 # API: Put a policy type: 1 again
124 with open(testdata+'pt1.json') as json_file:
125 policytype_1 = json.load(json_file)
126 response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
127 assert response.status_code == 201
129 # API: Delete a policy type
130 response=client.delete(SERVER_URL+'a1-p/policytypes/1')
131 assert response.status_code == 204
133 # API: Get policy type ids, shall be empty
134 data_policytypes_get = [ ]
135 response=client.get(SERVER_URL+'a1-p/policytypes')
136 assert response.status_code == 200
137 result=json.loads(response.data)
138 res=compare(data_policytypes_get, result)
141 # API: Put a policy type: 1
142 with open(testdata+'pt1.json') as json_file:
143 policytype_1 = json.load(json_file)
144 response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
145 assert response.status_code == 201
147 # API: Get policy type ids, shall contain '1'
148 data_policytypes_get = [ 1 ]
149 response=client.get(SERVER_URL+'a1-p/policytypes')
150 assert response.status_code == 200
151 result=json.loads(response.data)
152 res=compare(data_policytypes_get, result)
155 # API: Get instances for type 1, shall be empty
156 data_policies_get = [ ]
157 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
158 assert response.status_code == 200
159 result=json.loads(response.data)
160 res=compare(data_policies_get, result)
163 # API: Create policy instance pi1 of type: 1
164 with open(testdata+'pi1.json') as json_file:
165 policy_1 = json.load(json_file)
166 response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi1', headers=header, data=json.dumps(policy_1))
167 assert response.status_code == 202
169 # API: Get policy instance pi1 of type: 1
170 with open(testdata+'pi1.json') as json_file:
171 policy_1 = json.load(json_file)
172 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
173 assert response.status_code == 200
174 result=json.loads(response.data)
175 res=compare(policy_1, result)
178 # API: Update policy instance pi1 of type: 1
179 with open(testdata+'pi1.json') as json_file:
180 policy_1 = json.load(json_file)
181 response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi1', headers=header, data=json.dumps(policy_1))
182 assert response.status_code == 202
184 # API: Update policy type: 1, shall fail
185 with open(testdata+'pt1.json') as json_file:
186 policytype_1 = json.load(json_file)
187 response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
188 assert response.status_code == 400
190 # API: Get instances for type 1, shall contain 'pi1'
191 data_policies_get = [ "pi1" ]
192 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
193 assert response.status_code == 200
194 result=json.loads(response.data)
195 res=compare(data_policies_get, result)
198 # API: Create policy instance pi2 (copy of pi1) of type: 1.
199 with open(testdata+'pi1.json') as json_file:
200 policy_2 = json.load(json_file)
201 response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi2', headers=header, data=json.dumps(policy_2))
202 assert response.status_code == 202
204 # API: DELETE policy instance pi1
205 response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi2')
206 assert response.status_code == 202
208 # Set force response code 401
209 response=client.post(SERVER_URL+'forceresponse?code=401')
210 assert response.status_code == 200
212 # API: Get policy type 1. Shall fail with forced code
213 response=client.get(SERVER_URL+'a1-p/policytypes/1')
214 assert response.status_code == 401
216 # API: Get policy status
218 "enforceStatus" : "NOT_ENFORCED",
219 "enforceReason" : "OTHER_REASON",
221 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
222 assert response.status_code == 200
223 result=json.loads(response.data)
224 res=compare(policy_status, result)
227 # Load a policy type: 2
228 with open(testdata+'pt2.json') as json_file:
229 policytype_2 = json.load(json_file)
230 response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
231 assert response.status_code == 201
232 assert response.data == b"Policy type 2 is OK."
234 # Load a policy type: 2, again
235 with open(testdata+'pt2.json') as json_file:
236 policytype_2 = json.load(json_file)
237 response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
238 assert response.status_code == 200
239 assert response.data == b"Policy type 2 is OK."
241 # API: Get policy type ids, shall contain '1' and '2'
242 data_policytypes_get = [ 1,2 ]
243 response=client.get(SERVER_URL+'a1-p/policytypes')
244 assert response.status_code == 200
245 result=json.loads(response.data)
246 res=compare(data_policytypes_get, result)
249 # Get policy type ids, shall contain type 1 and 2 =="
250 data_policytypes_get = [ "1","2" ]
251 response=client.get(SERVER_URL+'policytypes')
252 assert response.status_code == 200
253 result=json.loads(response.data)
254 res=compare(data_policytypes_get, result)
257 # API: Get policy type 2
258 with open(testdata+'pt2.json') as json_file:
259 policytype_2 = json.load(json_file)
260 response=client.get(SERVER_URL+'a1-p/policytypes/2')
261 assert response.status_code == 200
262 result=json.loads(response.data)
263 res=compare(policytype_2, result)
266 # Delete a policy type
267 response=client.delete(SERVER_URL+'policytype?id=2')
268 assert response.status_code == 204
270 # API: Get policy type ids, shall contain '1'
271 data_policytypes_get = [ 1]
272 response=client.get(SERVER_URL+'a1-p/policytypes')
273 assert response.status_code == 200
274 result=json.loads(response.data)
275 res=compare(data_policytypes_get, result)
278 # Load a policy type: 2
279 with open(testdata+'pt2.json') as json_file:
280 policytype_2 = json.load(json_file)
281 response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
282 assert response.status_code == 201
283 assert response.data == b"Policy type 2 is OK."
285 # API: Get policy type 2
286 with open(testdata+'pt2.json') as json_file:
287 policytype_2 = json.load(json_file)
288 response=client.get(SERVER_URL+'a1-p/policytypes/2')
289 assert response.status_code == 200
290 result=json.loads(response.data)
291 res=compare(policytype_2, result)
294 # API: Get instances for type 2, shall be empty
295 data_policies_get = [ ]
296 response=client.get(SERVER_URL+'a1-p/policytypes/2/policies')
297 assert response.status_code == 200
298 result=json.loads(response.data)
299 res=compare(data_policies_get, result)
302 # API: Create policy instance pi1 of type: 2, shall fail
303 with open(testdata+'pi1.json') as json_file:
304 policy_1 = json.load(json_file)
305 response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi1', headers=header, data=json.dumps(policy_1))
306 assert response.status_code == 400
308 # API: Create policy instance pi2 of type: 2. Missing param, shall fail
309 with open(testdata+'pi2_missing_param.json') as json_file:
310 policy_2 = json.load(json_file)
311 response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
312 assert response.status_code == 400
314 # API: Create policy instance pi2 of type: 2
315 with open(testdata+'pi2.json') as json_file:
316 policy_2 = json.load(json_file)
317 response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
318 assert response.status_code == 202
320 with open(testdata+'pi2.json') as json_file:
321 policy_2 = json.load(json_file)
322 response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
323 assert response.status_code == 202
325 # API: Get instances for type 1, shall contain pi1
326 data_policies_get = [ "pi1" ]
327 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
328 assert response.status_code == 200
329 result=json.loads(response.data)
330 res=compare(data_policies_get, result)
333 # API: Get instances for type 2, shall contain pi2
334 data_policies_get = ["pi2" ]
335 response=client.get(SERVER_URL+'a1-p/policytypes/2/policies')
336 assert response.status_code == 200
337 result=json.loads(response.data)
338 res=compare(data_policies_get, result)
341 # Set force response code 409. ==="
342 response=client.post(SERVER_URL+'forceresponse?code=401')
343 assert response.status_code == 200
345 # API: Get policy status for pi1, shall fail
346 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
347 assert response.status_code == 401
350 response=client.post(SERVER_URL+'forcedelay?delay=10')
351 assert response.status_code == 200
352 assert response.data == b"Force delay: 10 sec set for all A1 responses"
354 # API: Get policy status for pi1. Shall delay 10 sec
356 "enforceStatus" : "NOT_ENFORCED",
357 "enforceReason" : "OTHER_REASON",
359 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
360 assert response.status_code == 200
361 result=json.loads(response.data)
362 res=compare(policy_status, result)
366 response=client.post(SERVER_URL+'forcedelay')
367 assert response.status_code == 200
368 assert response.data == b"Force delay: None sec set for all A1 responses"
371 response=client.put(SERVER_URL+'status?policyid=pi1&status=ENFORCED')
372 assert response.status_code == 200
374 # API: Get policy status for pi1
376 "enforceStatus" : "ENFORCED",
377 "enforceReason" : None,
379 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
380 assert response.status_code == 200
381 result=json.loads(response.data)
382 res=compare(policy_status, result)
386 response=client.put(SERVER_URL+'status?policyid=pi1&status=NOT_ENFORCED&reason=SCOPE_NOT_APPLICABLE')
387 assert response.status_code == 200
389 # API: Get policy status for pi1
391 "enforceStatus" : "NOT_ENFORCED",
392 "enforceReason" : "SCOPE_NOT_APPLICABLE",
394 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
395 assert response.status_code == 200
396 result=json.loads(response.data)
397 res=compare(policy_status, result)
400 # Get counter: num_instances
401 response=client.get(SERVER_URL+'counter/num_instances')
402 assert response.status_code == 200
403 assert response.data == b"2"
405 # Get counter: types (shall be 2)
406 response=client.get(SERVER_URL+'counter/num_types')
407 assert response.status_code == 200
408 assert response.data == b"2"
410 # Get counter: interface
411 response=client.get(SERVER_URL+'counter/interface')
412 assert response.status_code == 200
413 assert response.data == b"OSC_2.1.0"
415 # Get counter: remote hosts
416 response=client.get(SERVER_URL+'counter/remote_hosts')
417 assert response.status_code == 200
419 # Get counter: test, shall fail
420 response=client.get(SERVER_URL+'counter/test')
421 assert response.status_code == 404
423 # API: DELETE policy instance pi1
424 response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
425 assert response.status_code == 202
427 # API: Get instances for type 1, shall be empty
428 data_policies_get = [ ]
429 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
430 assert response.status_code == 200
431 result=json.loads(response.data)
432 res=compare(data_policies_get, result)
435 # API: Get instances for type 2, shall contain pi2
436 data_policies_get = ["pi2" ]
437 response=client.get(SERVER_URL+'a1-p/policytypes/2/policies')
438 assert response.status_code == 200
439 result=json.loads(response.data)
440 res=compare(data_policies_get, result)
443 # Get counter: instances
444 response=client.get(SERVER_URL+'counter/num_instances')
445 assert response.status_code == 200
446 assert response.data == b"1"
449 ### Tests to increase code coverage
451 # Set force response code 500
452 response=client.post(SERVER_URL+'forceresponse?code=500')
453 assert response.status_code == 200
456 response=client.get(SERVER_URL+'a1-p/healthcheck')
457 assert response.status_code == 500
459 # Set force response code 501
460 response=client.post(SERVER_URL+'forceresponse?code=501')
461 assert response.status_code == 200
463 # API: Get policy types
464 data_policytypes_get = [ ]
465 response=client.get(SERVER_URL+'a1-p/policytypes')
466 assert response.status_code == 501
468 # Set force response code 502
469 response=client.post(SERVER_URL+'forceresponse?code=502')
470 assert response.status_code == 200
472 # API: Delete a policy type, shall fail
473 response=client.delete(SERVER_URL+'a1-p/policytypes/55')
474 assert response.status_code == 502
476 # Set force response code 503. ==="
477 response=client.post(SERVER_URL+'forceresponse?code=503')
478 assert response.status_code == 200
480 with open(testdata+'pi1.json') as json_file:
481 policy_1 = json.load(json_file)
482 response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi11', headers=header, data=json.dumps(policy_1))
483 assert response.status_code == 503
485 # Set force response code 504
486 response=client.post(SERVER_URL+'forceresponse?code=504')
487 assert response.status_code == 200
489 # API: Get instances for type 1, shall fail
490 data_policies_get = [ ]
491 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
492 assert response.status_code == 504
494 # Set force response code 505. ==="
495 response=client.post(SERVER_URL+'forceresponse?code=505')
496 assert response.status_code == 200
498 # API: delete instance pi1, shall fail
499 response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
500 assert response.status_code == 505
502 # API: Delete a policy type having instances, shall fail
503 response=client.delete(SERVER_URL+'a1-p/policytypes/2')
504 assert response.status_code == 400
506 # API: delete instance pi1 in type 5, shall fail
507 response=client.delete(SERVER_URL+'a1-p/policytypes/5/policies/pi1')
508 assert response.status_code == 404
510 # API: delete instance pi99 in type 1, shall fail
511 response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi99')
512 assert response.status_code == 404
514 # API: Create policy instance pi80 of type: 5
515 with open(testdata+'pi1.json') as json_file:
516 policy_80 = json.load(json_file)
517 response=client.put(SERVER_URL+'a1-p/policytypes/5/policies/pi80', headers=header, data=json.dumps(policy_80))
518 assert response.status_code == 404
520 # API: Get policy type
521 data_policytypes_get = [ ]
522 response=client.get(SERVER_URL+'a1-p/policytypes/55')
523 assert response.status_code == 404
525 # API: Get status, bad type - shall fail
526 response=client.get(SERVER_URL+'a1-p/policytypes/99/policies/pi1/status')
527 assert response.status_code == 404
529 # API: Get status, bad instance - shall fail
530 response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi111/status')
531 assert response.status_code == 404
533 # Load policy type, no type in url - shall fail
534 with open(testdata+'pt2.json') as json_file:
535 policytype_2 = json.load(json_file)
536 response=client.put(SERVER_URL+'policytype', headers=header, data=json.dumps(policytype_2))
537 assert response.status_code == 400
539 # Load policy type - duplicatee - shall fail
540 with open(testdata+'pt1.json') as json_file:
541 policytype_1 = json.load(json_file)
542 response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_1))
543 assert response.status_code == 400
545 # Get counter: data_delivery
546 response=client.get(SERVER_URL+'counter/datadelivery')
547 assert response.status_code == 200
548 assert response.data == b"0"
550 # Send data to data-delivery with empty payload
552 response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
553 assert response.status_code == 400
555 # Send invalid data to data-delivery
560 response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
561 assert response.status_code == 404
563 # Send data to data-delivery with valid job
568 response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
569 assert response.status_code == 200
571 # Send data to data-delivery with valid job
574 "payload":"another payload"
576 response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
577 assert response.status_code == 200
579 # Get counter: data_delivery
580 response=client.get(SERVER_URL+'counter/datadelivery')
581 assert response.status_code == 200
582 assert response.data == b"2"
584 def test_notificationDestination(client):
585 test_data = get_testdata_dir() + 'pi2.json'
586 # Header for json payload
587 header = { "Content-Type" : "application/json" }
589 # === API: Update policy instance pi2 of type: 2 ==="
590 with open(test_data) as json_file:
591 payload = json.load(json_file)
592 response = client.put(SERVER_URL+"a1-p/policytypes/2/policies/pi2?notificationDestination=http://localhost:8086/statustest", headers=header, data=json.dumps(payload))
594 assert response.status_code == 202
595 result = response.data
599 def test_sendstatus(client):
600 # Create a new thread to run the Flask app in parallel on a different port so that we can call the callback.
601 proc = multiprocessing.Process(target=run_flask_app, args=())
604 test_data = get_testdata_dir() + 'pi2.json'
605 header = { "Content-Type" : "application/json" }
607 # === Send status for pi2===
608 with open(test_data) as json_file:
609 payload = json.load(json_file)
610 response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(payload))
612 assert response.status_code == 201
613 result = response.data
614 assert result == b"OK"
616 # Send status, negative test with missing parameter
617 response = client.post(SERVER_URL+'sendstatus', headers=header, data="")
618 assert response.status_code == 400
620 # Send status pi9, negative test for policy id not found
621 response = client.post(SERVER_URL+'sendstatus?policyid=pi9', headers=header, data="")
622 assert response.status_code == 404