5b8be30b323d0d739a67f9da6cab889af650a76b
[sim/a1-interface.git] / near-rt-ric-simulator / tests / OSC_2.1.0 / test_osc_2_1_0.py
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
7 #
8 #       http://www.apache.org/licenses/LICENSE-2.0
9 #
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=================================================
16 #
17
18 # This test case test the OSC_2.1.0 version of the simulator
19
20 from unittest import TestCase
21 import sys
22 import os
23 import threading
24 import requests
25 import json
26
27
28 #Constants for the test case
29 INTERFACE_VERSION="OSC_2.1.0"
30 PORT_NUMBER=2222
31 HOST_IP="127.0.0.1"
32 SERVER_URL="http://127.0.0.1:"+str(PORT_NUMBER)+"/"
33
34 cwd=os.getcwd()+"/"
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/"
40
41 #Env var to setup version and host logging
42 os.environ['APIPATH'] = cwd+"/../../api/"+INTERFACE_VERSION
43 os.environ['REMOTE_HOSTS_LOGGING'] = "ON"
44
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)
50
51 import main
52 from compare_json import compare
53
54 class TestCase1(TestCase):
55     from main import app
56     server = None
57
58     def startServer(self):
59         self.app.run(port=PORT_NUMBER, host=HOST_IP, threaded=True)
60
61     def setUp(self):
62         self.server=threading.Thread(target=self.startServer, args=())
63         self.server.daemon = True
64         self.server.start()
65
66     # executed after each test
67     def tearDown(self):
68         self.server.killed=True
69
70     def test_apis(self):
71
72         # Header for json payload
73         header = {
74             "Content-Type" : "application/json"
75         }
76
77         # Simulator hello world
78         response=requests.get(SERVER_URL)
79         self.assertEqual(response.status_code, 200)
80
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']")
85
86         # Reset simulator instances
87         response=requests.post(SERVER_URL+'deleteinstances')
88         self.assertEqual(response.status_code, 200)
89
90         # Reset simulator, all
91         response=requests.post(SERVER_URL+'deleteall')
92         self.assertEqual(response.status_code, 200)
93
94         # API: Healthcheck
95         response=requests.get(SERVER_URL+'a1-p/healthcheck')
96         self.assertEqual(response.status_code, 200)
97
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)
105
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)
109
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)
113
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)
119
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)
125
126         # API: Delete a policy type
127         response=requests.delete(SERVER_URL+'a1-p/policytypes/1')
128         self.assertEqual(response.status_code, 204)
129
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)
137
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)
143
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)
151
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)
159
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)
165
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)
174
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)
180
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)
186
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)
194
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)
200
201         # Set force response code 401
202         response=requests.post(SERVER_URL+'forceresponse?code=401')
203         self.assertEqual(response.status_code, 200)
204
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)
208
209         # API: Get policy status
210         policy_status = {
211             "instance_status" : "NOT IN EFFECT",
212             "has_been_deleted" : "false",
213             "created_at" : "????"
214         }
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)
220
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.")
227
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.")
234
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)
242
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)
250
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)
259
260         # Delete a policy type
261         response=requests.delete(SERVER_URL+'policytype?id=2')
262         self.assertEqual(response.status_code, 204)
263
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)
271
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.")
278
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)
287
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)
295
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)
301
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)
307
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)
313
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)
318
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)
326
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)
334
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)
340
341         # Set force response code 409. ==="
342         response=requests.post(SERVER_URL+'forceresponse?code=401')
343         self.assertEqual(response.status_code, 200)
344
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)
348
349         # Set force delay 10
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")
353
354         # API: Get policy status for pi1. Shall delay 10 sec
355         policy_status = {
356             "instance_status" : "NOT IN EFFECT",
357             "has_been_deleted" : "false",
358             "created_at" : "????"
359         }
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)
365
366         # Reset force delay
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")
370
371         #  Set status for pi1
372         response=requests.put(SERVER_URL+'status?policyid=pi1&status=IN%20EFFECT')
373         self.assertEqual(response.status_code, 200)
374
375         # API: Get policy status for pi1
376         policy_status = {
377             "instance_status" : "IN EFFECT",
378             "has_been_deleted" : "false",
379             "created_at" : "????"
380         }
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)
386
387         #  Set status for pi1
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)
390
391         # API: Get policy status for pi1
392         policy_status = {
393             "instance_status" : "IN EFFECT",
394             "has_been_deleted" : "true",
395             "created_at" : "????"
396         }
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)
402
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")
407
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")
412
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")
417
418         # Get counter: remote hosts
419         response=requests.get(SERVER_URL+'counter/remote_hosts')
420         self.assertEqual(response.status_code, 200)
421
422         # Get counter: test, shall fail
423         response=requests.get(SERVER_URL+'counter/test')
424         self.assertEqual(response.status_code, 404)
425
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)
429
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)
437
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)
445
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")
450
451
452         ### Tests to increase code coverage
453
454         # Set force response code 500
455         response=requests.post(SERVER_URL+'forceresponse?code=500')
456         self.assertEqual(response.status_code, 200)
457
458         # API: Healthcheck
459         response=requests.get(SERVER_URL+'a1-p/healthcheck')
460         self.assertEqual(response.status_code, 500)
461
462         # Set force response code 501
463         response=requests.post(SERVER_URL+'forceresponse?code=501')
464         self.assertEqual(response.status_code, 200)
465
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)
470
471         # Set force response code 502
472         response=requests.post(SERVER_URL+'forceresponse?code=502')
473         self.assertEqual(response.status_code, 200)
474
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)
478
479         # Set force response code 503. ==="
480         response=requests.post(SERVER_URL+'forceresponse?code=503')
481         self.assertEqual(response.status_code, 200)
482
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)
487
488         # Set force response code 504
489         response=requests.post(SERVER_URL+'forceresponse?code=504')
490         self.assertEqual(response.status_code, 200)
491
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)
496
497         # Set force response code 505. ==="
498         response=requests.post(SERVER_URL+'forceresponse?code=505')
499         self.assertEqual(response.status_code, 200)
500
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)
504
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)
508
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)
512
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)
516
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)
522
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)
527
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)
531
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)
535
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)
541
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)