1 # ============LICENSE_START===============================================
2 # Copyright (C) 2020 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 OSC_2.1.0 version of the simulator
20 from unittest import TestCase
28 #Constants for the test case
29 INTERFACE_VERSION="OSC_2.1.0"
32 SERVER_URL="http://"+HOST_IP+":"+PORT_NUMBER+"/"
35 # Env TESTS_BASE_PATH is set when executed via tox.ini
36 # If basic test is executed from cmd line, that env var is not needed
37 if 'TESTS_BASE_PATH' in os.environ:
38 cwd=os.environ['TESTS_BASE_PATH']+"/"+INTERFACE_VERSION+"/"
39 TESTDATA=cwd+"/../../test/"+INTERFACE_VERSION+"/jsonfiles/"
41 #Env var to setup version and host logging
42 os.environ['APIPATH'] = cwd+"/../../api/"+INTERFACE_VERSION
43 os.environ['REMOTE_HOSTS_LOGGING'] = "ON"
45 # Paths need to run the sim, including needed source file dirs
46 sys.path.append(os.path.abspath(cwd+'../../src/common'))
47 sys.path.append(os.path.abspath(cwd+'../../test/common'))
48 sys.path.append(os.path.abspath(cwd+'../../src/'+INTERFACE_VERSION))
49 os.chdir(cwd+"../../src/"+INTERFACE_VERSION)
52 from compare_json import compare
54 class TestCase1(TestCase):
58 def startServer(self):
59 self.app.run(port=PORT_NUMBER, host=HOST_IP, threaded=True)
62 self.server=threading.Thread(target=self.startServer, args=())
63 self.server.daemon = True
66 # executed after each test
68 self.server.killed=True
72 # Header for json payload
74 "Content-Type" : "application/json"
77 # Simulator hello world
78 response=requests.get(SERVER_URL)
79 self.assertEqual(response.status_code, 200)
81 # Check used and implemented interfaces
82 response=requests.get(SERVER_URL+'container_interfaces')
83 self.assertEqual(response.status_code, 200)
84 self.assertEqual(response.text, "Current interface: OSC_2.1.0 All supported A1 interface yamls in this container: ['OSC_2.1.0', 'STD_1.1.3']")
86 # Reset simulator instances
87 response=requests.post(SERVER_URL+'deleteinstances')
88 self.assertEqual(response.status_code, 200)
90 # Reset simulator, all
91 response=requests.post(SERVER_URL+'deleteall')
92 self.assertEqual(response.status_code, 200)
95 response=requests.get(SERVER_URL+'a1-p/healthcheck')
96 self.assertEqual(response.status_code, 200)
98 # API: Get policy types, shall be empty
99 data_policytypes_get = [ ]
100 response=requests.get(SERVER_URL+'a1-p/policytypes')
101 self.assertEqual(response.status_code, 200)
102 result=json.loads(response.text)
103 res=compare(data_policytypes_get, result)
104 self.assertEqual(res, True)
106 # API: Delete a policy type, shall fail
107 response=requests.delete(SERVER_URL+'a1-p/policytypes/1')
108 self.assertEqual(response.status_code, 404)
110 # API: Get policy instances for type 1, shall fail
111 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies')
112 self.assertEqual(response.status_code, 404)
114 # API: Put a policy type: 1
115 with open(TESTDATA+'pt1.json') as json_file:
116 policytype_1 = json.load(json_file)
117 response=requests.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
118 self.assertEqual(response.status_code, 201)
120 # API: Put a policy type: 1 again
121 with open(TESTDATA+'pt1.json') as json_file:
122 policytype_1 = json.load(json_file)
123 response=requests.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
124 self.assertEqual(response.status_code, 201)
126 # API: Delete a policy type
127 response=requests.delete(SERVER_URL+'a1-p/policytypes/1')
128 self.assertEqual(response.status_code, 204)
130 # API: Get policy type ids, shall be empty
131 data_policytypes_get = [ ]
132 response=requests.get(SERVER_URL+'a1-p/policytypes')
133 self.assertEqual(response.status_code, 200)
134 result=json.loads(response.text)
135 res=compare(data_policytypes_get, result)
136 self.assertEqual(res, True)
138 # API: Put a policy type: 1
139 with open(TESTDATA+'pt1.json') as json_file:
140 policytype_1 = json.load(json_file)
141 response=requests.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
142 self.assertEqual(response.status_code, 201)
144 # API: Get policy type ids, shall contain '1'
145 data_policytypes_get = [ 1 ]
146 response=requests.get(SERVER_URL+'a1-p/policytypes')
147 self.assertEqual(response.status_code, 200)
148 result=json.loads(response.text)
149 res=compare(data_policytypes_get, result)
150 self.assertEqual(res, True)
152 # API: Get instances for type 1, shall be empty
153 data_policies_get = [ ]
154 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies')
155 self.assertEqual(response.status_code, 200)
156 result=json.loads(response.text)
157 res=compare(data_policies_get, result)
158 self.assertEqual(res, True)
160 # API: Create policy instance pi1 of type: 1
161 with open(TESTDATA+'pi1.json') as json_file:
162 policy_1 = json.load(json_file)
163 response=requests.put(SERVER_URL+'a1-p/policytypes/1/policies/pi1', headers=header, data=json.dumps(policy_1))
164 self.assertEqual(response.status_code, 202)
166 # API: Get policy instance pi1 of type: 1
167 with open(TESTDATA+'pi1.json') as json_file:
168 policy_1 = json.load(json_file)
169 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
170 self.assertEqual(response.status_code, 200)
171 result=json.loads(response.text)
172 res=compare(policy_1, result)
173 self.assertEqual(res, True)
175 # API: Update policy instance pi1 of type: 1
176 with open(TESTDATA+'pi1.json') as json_file:
177 policy_1 = json.load(json_file)
178 response=requests.put(SERVER_URL+'a1-p/policytypes/1/policies/pi1', headers=header, data=json.dumps(policy_1))
179 self.assertEqual(response.status_code, 202)
181 # API: Update policy type: 1, shall fail
182 with open(TESTDATA+'pt1.json') as json_file:
183 policytype_1 = json.load(json_file)
184 response=requests.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
185 self.assertEqual(response.status_code, 400)
187 # API: Get instances for type 1, shall contain 'pi1'
188 data_policies_get = [ "pi1" ]
189 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies')
190 self.assertEqual(response.status_code, 200)
191 result=json.loads(response.text)
192 res=compare(data_policies_get, result)
193 self.assertEqual(res, True)
195 # API: Create policy instance pi2 (copy of pi1) of type: 1. Shall fail
196 with open(TESTDATA+'pi1.json') as json_file:
197 policy_2 = json.load(json_file)
198 response=requests.put(SERVER_URL+'a1-p/policytypes/1/policies/pi2', headers=header, data=json.dumps(policy_2))
199 self.assertEqual(response.status_code, 400)
201 # Set force response code 401
202 response=requests.post(SERVER_URL+'forceresponse?code=401')
203 self.assertEqual(response.status_code, 200)
205 # API: Get policy type 1. Shall fail with forced code
206 response=requests.get(SERVER_URL+'a1-p/policytypes/1')
207 self.assertEqual(response.status_code, 401)
209 # API: Get policy status
211 "instance_status" : "NOT IN EFFECT",
212 "has_been_deleted" : "false",
213 "created_at" : "????"
215 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
216 self.assertEqual(response.status_code, 200)
217 result=json.loads(response.text)
218 res=compare(policy_status, result)
219 self.assertEqual(res, True)
221 # Load a policy type: 2
222 with open(TESTDATA+'pt2.json') as json_file:
223 policytype_2 = json.load(json_file)
224 response=requests.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
225 self.assertEqual(response.status_code, 201)
226 self.assertEqual(response.text, "Policy type 2 is OK.")
228 # Load a policy type: 2, again
229 with open(TESTDATA+'pt2.json') as json_file:
230 policytype_2 = json.load(json_file)
231 response=requests.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
232 self.assertEqual(response.status_code, 200)
233 self.assertEqual(response.text, "Policy type 2 is OK.")
235 # API: Get policy type ids, shall contain '1' and '2'
236 data_policytypes_get = [ 1,2 ]
237 response=requests.get(SERVER_URL+'a1-p/policytypes')
238 self.assertEqual(response.status_code, 200)
239 result=json.loads(response.text)
240 res=compare(data_policytypes_get, result)
241 self.assertEqual(res, True)
243 # Get policy type ids, shall contain type 1 and 2 =="
244 data_policytypes_get = [ "1","2" ]
245 response=requests.get(SERVER_URL+'policytypes')
246 self.assertEqual(response.status_code, 200)
247 result=json.loads(response.text)
248 res=compare(data_policytypes_get, result)
249 self.assertEqual(res, True)
251 # API: Get policy type 2
252 with open(TESTDATA+'pt2.json') as json_file:
253 policytype_2 = json.load(json_file)
254 response=requests.get(SERVER_URL+'a1-p/policytypes/2')
255 self.assertEqual(response.status_code, 200)
256 result=json.loads(response.text)
257 res=compare(policytype_2, result)
258 self.assertEqual(res, True)
260 # Delete a policy type
261 response=requests.delete(SERVER_URL+'policytype?id=2')
262 self.assertEqual(response.status_code, 204)
264 # API: Get policy type ids, shall contain '1'
265 data_policytypes_get = [ 1]
266 response=requests.get(SERVER_URL+'a1-p/policytypes')
267 self.assertEqual(response.status_code, 200)
268 result=json.loads(response.text)
269 res=compare(data_policytypes_get, result)
270 self.assertEqual(res, True)
272 # Load a policy type: 2
273 with open(TESTDATA+'pt2.json') as json_file:
274 policytype_2 = json.load(json_file)
275 response=requests.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
276 self.assertEqual(response.status_code, 201)
277 self.assertEqual(response.text, "Policy type 2 is OK.")
279 # API: Get policy type 2
280 with open(TESTDATA+'pt2.json') as json_file:
281 policytype_2 = json.load(json_file)
282 response=requests.get(SERVER_URL+'a1-p/policytypes/2')
283 self.assertEqual(response.status_code, 200)
284 result=json.loads(response.text)
285 res=compare(policytype_2, result)
286 self.assertEqual(res, True)
288 # API: Get instances for type 2, shall be empty
289 data_policies_get = [ ]
290 response=requests.get(SERVER_URL+'a1-p/policytypes/2/policies')
291 self.assertEqual(response.status_code, 200)
292 result=json.loads(response.text)
293 res=compare(data_policies_get, result)
294 self.assertEqual(res, True)
296 # API: Create policy instance pi1 of type: 2, shall fail
297 with open(TESTDATA+'pi1.json') as json_file:
298 policy_1 = json.load(json_file)
299 response=requests.put(SERVER_URL+'a1-p/policytypes/2/policies/pi1', headers=header, data=json.dumps(policy_1))
300 self.assertEqual(response.status_code, 400)
302 # API: Create policy instance pi2 of type: 2. Missing param, shall fail
303 with open(TESTDATA+'pi2_missing_param.json') as json_file:
304 policy_2 = json.load(json_file)
305 response=requests.put(SERVER_URL+'a1-p/policytypes/2/policies/pi1', headers=header, data=json.dumps(policy_2))
306 self.assertEqual(response.status_code, 400)
308 # API: Create policy instance pi2 of type: 2
309 with open(TESTDATA+'pi2.json') as json_file:
310 policy_2 = json.load(json_file)
311 response=requests.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
312 self.assertEqual(response.status_code, 202)
314 with open(TESTDATA+'pi2.json') as json_file:
315 policy_2 = json.load(json_file)
316 response=requests.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
317 self.assertEqual(response.status_code, 202)
319 # API: Get instances for type 1, shall contain pi1
320 data_policies_get = [ "pi1" ]
321 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies')
322 self.assertEqual(response.status_code, 200)
323 result=json.loads(response.text)
324 res=compare(data_policies_get, result)
325 self.assertEqual(res, True)
327 # API: Get instances for type 2, shall contain pi2
328 data_policies_get = ["pi2" ]
329 response=requests.get(SERVER_URL+'a1-p/policytypes/2/policies')
330 self.assertEqual(response.status_code, 200)
331 result=json.loads(response.text)
332 res=compare(data_policies_get, result)
333 self.assertEqual(res, True)
335 # API: Create policy instance pi11 (copy of pi1) of type: 1. Shall fail
336 with open(TESTDATA+'pi1.json') as json_file:
337 policy_1 = json.load(json_file)
338 response=requests.put(SERVER_URL+'a1-p/policytypes/1/policies/pi11', headers=header, data=json.dumps(policy_1))
339 self.assertEqual(response.status_code, 400)
341 # Set force response code 409. ==="
342 response=requests.post(SERVER_URL+'forceresponse?code=401')
343 self.assertEqual(response.status_code, 200)
345 # API: Get policy status for pi1, shall fail
346 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
347 self.assertEqual(response.status_code, 401)
350 response=requests.post(SERVER_URL+'forcedelay?delay=10')
351 self.assertEqual(response.status_code, 200)
352 self.assertEqual(response.text, "Force delay: 10 sec set for all A1 responses")
354 # API: Get policy status for pi1. Shall delay 10 sec
356 "instance_status" : "NOT IN EFFECT",
357 "has_been_deleted" : "false",
358 "created_at" : "????"
360 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
361 self.assertEqual(response.status_code, 200)
362 result=json.loads(response.text)
363 res=compare(policy_status, result)
364 self.assertEqual(res, True)
367 response=requests.post(SERVER_URL+'forcedelay')
368 self.assertEqual(response.status_code, 200)
369 self.assertEqual(response.text, "Force delay: None sec set for all A1 responses")
372 response=requests.put(SERVER_URL+'status?policyid=pi1&status=IN%20EFFECT')
373 self.assertEqual(response.status_code, 200)
375 # API: Get policy status for pi1
377 "instance_status" : "IN EFFECT",
378 "has_been_deleted" : "false",
379 "created_at" : "????"
381 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
382 self.assertEqual(response.status_code, 200)
383 result=json.loads(response.text)
384 res=compare(policy_status, result)
385 self.assertEqual(res, True)
388 response=requests.put(SERVER_URL+'status?policyid=pi1&status=IN%20EFFECT&deleted=true&created_at=2020-03-30%2012:00:00')
389 self.assertEqual(response.status_code, 200)
391 # API: Get policy status for pi1
393 "instance_status" : "IN EFFECT",
394 "has_been_deleted" : "true",
395 "created_at" : "????"
397 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
398 self.assertEqual(response.status_code, 200)
399 result=json.loads(response.text)
400 res=compare(policy_status, result)
401 self.assertEqual(res, True)
403 # Get counter: intstance
404 response=requests.get(SERVER_URL+'counter/num_instances')
405 self.assertEqual(response.status_code, 200)
406 self.assertEqual(response.text, "2")
408 # Get counter: types (shall be 2)
409 response=requests.get(SERVER_URL+'counter/num_types')
410 self.assertEqual(response.status_code, 200)
411 self.assertEqual(response.text, "2")
413 # Get counter: interface
414 response=requests.get(SERVER_URL+'counter/interface')
415 self.assertEqual(response.status_code, 200)
416 self.assertEqual(response.text, "OSC_2.1.0")
418 # Get counter: remote hosts
419 response=requests.get(SERVER_URL+'counter/remote_hosts')
420 self.assertEqual(response.status_code, 200)
422 # Get counter: test, shall fail
423 response=requests.get(SERVER_URL+'counter/test')
424 self.assertEqual(response.status_code, 404)
426 # API: DELETE policy instance pi1
427 response=requests.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
428 self.assertEqual(response.status_code, 202)
430 # API: Get instances for type 1, shall be empty
431 data_policies_get = [ ]
432 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies')
433 self.assertEqual(response.status_code, 200)
434 result=json.loads(response.text)
435 res=compare(data_policies_get, result)
436 self.assertEqual(res, True)
438 # API: Get instances for type 2, shall contain pi2
439 data_policies_get = ["pi2" ]
440 response=requests.get(SERVER_URL+'a1-p/policytypes/2/policies')
441 self.assertEqual(response.status_code, 200)
442 result=json.loads(response.text)
443 res=compare(data_policies_get, result)
444 self.assertEqual(res, True)
446 # Get counter: instances
447 response=requests.get(SERVER_URL+'counter/num_instances')
448 self.assertEqual(response.status_code, 200)
449 self.assertEqual(response.text, "1")
452 ### Tests to increase code coverage
454 # Set force response code 500
455 response=requests.post(SERVER_URL+'forceresponse?code=500')
456 self.assertEqual(response.status_code, 200)
459 response=requests.get(SERVER_URL+'a1-p/healthcheck')
460 self.assertEqual(response.status_code, 500)
462 # Set force response code 501
463 response=requests.post(SERVER_URL+'forceresponse?code=501')
464 self.assertEqual(response.status_code, 200)
466 # API: Get policy types
467 data_policytypes_get = [ ]
468 response=requests.get(SERVER_URL+'a1-p/policytypes')
469 self.assertEqual(response.status_code, 501)
471 # Set force response code 502
472 response=requests.post(SERVER_URL+'forceresponse?code=502')
473 self.assertEqual(response.status_code, 200)
475 # API: Delete a policy type, shall fail
476 response=requests.delete(SERVER_URL+'a1-p/policytypes/55')
477 self.assertEqual(response.status_code, 502)
479 # Set force response code 503. ==="
480 response=requests.post(SERVER_URL+'forceresponse?code=503')
481 self.assertEqual(response.status_code, 200)
483 with open(TESTDATA+'pi1.json') as json_file:
484 policy_1 = json.load(json_file)
485 response=requests.put(SERVER_URL+'a1-p/policytypes/1/policies/pi11', headers=header, data=json.dumps(policy_1))
486 self.assertEqual(response.status_code, 503)
488 # Set force response code 504
489 response=requests.post(SERVER_URL+'forceresponse?code=504')
490 self.assertEqual(response.status_code, 200)
492 # API: Get instances for type 1, shall fail
493 data_policies_get = [ ]
494 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies')
495 self.assertEqual(response.status_code, 504)
497 # Set force response code 505. ==="
498 response=requests.post(SERVER_URL+'forceresponse?code=505')
499 self.assertEqual(response.status_code, 200)
501 # API: delete instance pi1, shall fail
502 response=requests.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
503 self.assertEqual(response.status_code, 505)
505 # API: Delete a policy type having instances, shall fail
506 response=requests.delete(SERVER_URL+'a1-p/policytypes/2')
507 self.assertEqual(response.status_code, 400)
509 # API: delete instance pi1 in type 5, shall fail
510 response=requests.delete(SERVER_URL+'a1-p/policytypes/5/policies/pi1')
511 self.assertEqual(response.status_code, 404)
513 # API: delete instance pi99 in type 1, shall fail
514 response=requests.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi99')
515 self.assertEqual(response.status_code, 404)
517 # API: Create policy instance pi80 of type: 5
518 with open(TESTDATA+'pi1.json') as json_file:
519 policy_80 = json.load(json_file)
520 response=requests.put(SERVER_URL+'a1-p/policytypes/5/policies/pi80', headers=header, data=json.dumps(policy_80))
521 self.assertEqual(response.status_code, 404)
523 # API: Get policy type
524 data_policytypes_get = [ ]
525 response=requests.get(SERVER_URL+'a1-p/policytypes/55')
526 self.assertEqual(response.status_code, 404)
528 # API: Get status, bad type - shall fail
529 response=requests.get(SERVER_URL+'a1-p/policytypes/99/policies/pi1/status')
530 self.assertEqual(response.status_code, 404)
532 # API: Get status, bad instance - shall fail
533 response=requests.get(SERVER_URL+'a1-p/policytypes/1/policies/pi111/status')
534 self.assertEqual(response.status_code, 404)
536 # Load policy type, no type in url - shall faill
537 with open(TESTDATA+'pt2.json') as json_file:
538 policytype_2 = json.load(json_file)
539 response=requests.put(SERVER_URL+'policytype', headers=header, data=json.dumps(policytype_2))
540 self.assertEqual(response.status_code, 400)
542 # Load policy type - duplicatee - shall faill
543 with open(TESTDATA+'pt1.json') as json_file:
544 policytype_1 = json.load(json_file)
545 response=requests.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_1))
546 self.assertEqual(response.status_code, 400)