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 STD_1.1.3 version of the simulator
25 #Constants for the test case
26 INTERFACE_VERSION="STD_1.1.3"
29 SERVER_URL="http://"+HOST_IP+":"+PORT_NUMBER+"/"
32 # Env TESTS_BASE_PATH is set when executed via tox.ini
33 # If basic test is executed from cmd line, that env var is not needed
34 if 'TESTS_BASE_PATH' in os.environ:
35 cwd=os.environ['TESTS_BASE_PATH']+"/"+INTERFACE_VERSION+"/"
36 TESTDATA=cwd+"/../../test/"+INTERFACE_VERSION+"/jsonfiles/"
38 #Env var to setup api version and host logging
39 os.environ['APIPATH'] = cwd+"/../../api/"+INTERFACE_VERSION
40 os.environ['REMOTE_HOSTS_LOGGING'] = "ON"
42 # Paths need to run the sim, including needed source file dirs
43 sys.path.append(os.path.abspath(cwd+'../../src/common'))
44 sys.path.append(os.path.abspath(cwd+'../../test/common'))
45 sys.path.append(os.path.abspath(cwd+'../../src/'+INTERFACE_VERSION))
46 os.chdir(cwd+"../../src/"+INTERFACE_VERSION)
50 from compare_json import compare
54 with app.app.test_client() as c:
57 def test_apis(client):
59 # header for json payload
61 "Content-Type" : "application/json"
64 # Simulator hello world
65 response=client.get(SERVER_URL)
66 assert response.status_code == 200
68 # Check used and implemented interfaces
69 response=client.get(SERVER_URL+'container_interfaces')
70 assert response.status_code == 200
71 assert response.data == b"Current interface: STD_1.1.3 All supported A1 interface yamls in this container: ['OSC_2.1.0', 'STD_1.1.3']"
73 # Reset simulator instances
74 response=client.post(SERVER_URL+'deleteinstances')
75 assert response.status_code == 200
77 # Reset simulator, all
78 response=client.post(SERVER_URL+'deleteall')
79 assert response.status_code == 200
81 # API: Get policy instances, shall be empty
83 response=client.get(SERVER_URL+'A1-P/v1/policies')
84 assert response.status_code == 200
85 result=json.loads(response.data)
86 res=compare(data_policy_get, result)
89 #API: Create policy instance pi1
102 response=client.put(SERVER_URL+'A1-P/v1/policies/pi1', headers=header, data=json.dumps(data_pi1))
103 assert response.status_code == 201
104 result=json.loads(response.data)
105 res=compare(data_pi1, result)
108 #API: Update policy instance pi1
121 response=client.put(SERVER_URL+'A1-P/v1/policies/pi1', headers=header, data=json.dumps(data_pi1_updated))
122 assert response.status_code == 200
123 result=json.loads(response.data)
124 res=compare(data_pi1_updated, result)
127 #API: Create policy instance pi9, bad json
128 response=client.put(SERVER_URL+'A1-P/v1/policies/pi9', headers=header, data="test")
129 assert response.status_code == 400
131 # API: Get policy instances, shall contain pi1
132 data_policy_get = [ "pi1" ]
133 response=client.get(SERVER_URL+'A1-P/v1/policies')
134 assert response.status_code == 200
135 result=json.loads(response.data)
136 res=compare(data_policy_get, result)
139 # API: Create policy instance pi2 (copy of pi1). Shall fail
140 data_create_errror_pi1 = {
141 "title" : "The policy json already exists.",
145 response=client.put(SERVER_URL+'A1-P/v1/policies/pi2', headers=header, data=json.dumps(data_pi1_updated))
146 assert response.status_code == 400
147 result=json.loads(response.data)
148 res=compare(data_create_errror_pi1, result)
151 # Set force response code 409. ==="
152 response=client.post(SERVER_URL+'forceresponse?code=409')
153 assert response.status_code == 200
155 # API: Get policy instances, shall fail
157 "title" : "Conflict",
159 "detail" : "Request could not be processed in the current state of the resource"
161 response=client.get(SERVER_URL+'A1-P/v1/policies')
162 assert response.status_code == 409
163 result=json.loads(response.data)
164 res=compare(data_get_errror, result)
167 # Reset force response
168 response=client.post(SERVER_URL+'forceresponse')
169 assert response.status_code == 200
170 assert response.data == b"Force response code: None set for one single A1 response"
173 ### Repeating the above two test for code coverage
175 # Set force response code 400
176 response=client.post(SERVER_URL+'forceresponse?code=400')
177 assert response.status_code == 200
179 #API: Create policy instance pi3, shall fail
189 "title" : "Bad request",
191 "detail" : "Object in payload not properly formulated or not related to the method"
193 response=client.put(SERVER_URL+'A1-P/v1/policies/pi3', headers=header, data=json.dumps(data_pi3))
194 assert response.status_code == 400
195 result=json.loads(response.data)
196 res=compare(data_get_errror, result)
199 # Set force response code 404
200 response=client.post(SERVER_URL+'forceresponse?code=404')
201 assert response.status_code == 200
203 # API: Get policy instance pi1, shall fail =="
205 "title" : "Not found",
207 "detail" : "No resource found at the URI"
209 response=client.get(SERVER_URL+'A1-P/v1/policies/pi1')
210 assert response.status_code == 404
211 result=json.loads(response.data)
212 res=compare(data_get_errror, result)
215 # Set force response code 405
216 response=client.post(SERVER_URL+'forceresponse?code=405')
217 assert response.status_code == 200
219 # API: Delete policy instances pi1, shall fail =="
221 "title" : "Method not allowed",
223 "detail" : "Method not allowed for the URI"
225 response=client.delete(SERVER_URL+'A1-P/v1/policies/pi1')
226 assert response.status_code == 405
227 result=json.loads(response.data)
228 res=compare(data_get_errror, result)
231 # Set force response code 429
232 response=client.post(SERVER_URL+'forceresponse?code=429')
233 assert response.status_code == 200
235 # API: Get policy status pi3, shall fail =="
237 "title" : "Too many requests",
239 "detail" : "Too many requests have been sent in a given amount of time"
241 response=client.get(SERVER_URL+'A1-P/v1/policies/pi3/status')
242 assert response.status_code == 429
243 result=json.loads(response.data)
244 res=compare(data_get_errror, result)
247 # Set force response code 507
248 response=client.post(SERVER_URL+'forceresponse?code=507')
249 assert response.status_code == 200
251 # API: Get policy instances, shall fail
253 "title" : "Insufficient storage",
255 "detail" : "The method could not be performed on the resource because the provider is unable to store the representation needed to successfully complete the request"
257 response=client.get(SERVER_URL+'A1-P/v1/policies')
258 assert response.status_code == 507
259 result=json.loads(response.data)
260 res=compare(data_get_errror, result)
263 # Set force response code 503. ==="
264 response=client.post(SERVER_URL+'forceresponse?code=503')
265 assert response.status_code == 200
267 # API: Get policy instances, shall fail
269 "title" : "Service unavailable",
271 "detail" : "The provider is currently unable to handle the request due to a temporary overload"
273 response=client.get(SERVER_URL+'A1-P/v1/policies')
274 assert response.status_code == 503
275 result=json.loads(response.data)
276 res=compare(data_get_errror, result)
279 # Set force response code 555. ==="
280 response=client.post(SERVER_URL+'forceresponse?code=555')
281 assert response.status_code == 200
283 # API: Get policy instances, shall fail
287 "detail" : "Not implemented response code"
289 response=client.get(SERVER_URL+'A1-P/v1/policies')
290 assert response.status_code == 555
291 result=json.loads(response.data)
292 res=compare(data_get_errror, result)
296 ### End of repeated test
300 # API: Get policy status
301 data_policy_status = {
302 "enforceStatus" : "UNDEFINED"
304 response=client.get(SERVER_URL+'A1-P/v1/policies/pi1/status')
305 assert response.status_code == 200
306 result=json.loads(response.data)
307 res=compare(data_policy_status, result)
310 # API: Create policy instance pi2
323 response=client.put(SERVER_URL+'A1-P/v1/policies/pi2', headers=header, data=json.dumps(data_pi2))
324 assert response.status_code == 201
325 result=json.loads(response.data)
326 res=compare(data_pi2, result)
330 # API: Update policy instance pi2
331 # Reuse same policy data
332 response=client.put(SERVER_URL+'A1-P/v1/policies/pi2?notificationDestination=http://'+HOST_IP+':'+PORT_NUMBER+'/statustest', headers=header, data=json.dumps(data_pi2))
333 assert response.status_code == 200
334 result=json.loads(response.data)
335 res=compare(data_pi2, result)
338 # API: Get policy instances, shall contain pi1 and pi2
339 data_policy_get = [ "pi1", "pi2" ]
340 response=client.get(SERVER_URL+'A1-P/v1/policies')
341 assert response.status_code == 200
342 result=json.loads(response.data)
343 res=compare(data_policy_get, result)
347 response=client.post(SERVER_URL+'forcedelay?delay=10')
348 assert response.status_code == 200
349 assert response.data == b"Force delay: 10 sec set for all A1 responses"
351 # API: Get policy instances, shall contain pi1 and pi2 and delayed 10 sec
352 data_policy_get = [ "pi1", "pi2" ]
353 response=client.get(SERVER_URL+'A1-P/v1/policies')
354 assert response.status_code == 200
355 result=json.loads(response.data)
356 res=compare(data_policy_get, result)
360 response=client.post(SERVER_URL+'forcedelay')
361 assert response.status_code == 200
362 assert response.data == b"Force delay: None sec set for all A1 responses"
364 # API: GET policy instance pi1
365 response=client.get(SERVER_URL+'A1-P/v1/policies/pi1')
366 assert response.status_code == 200
367 result=json.loads(response.data)
368 res=compare(data_pi1_updated, result)
371 # API: GET policy instance pi2
372 response=client.get(SERVER_URL+'A1-P/v1/policies/pi2')
373 assert response.status_code == 200
374 result=json.loads(response.data)
375 res=compare(data_pi2, result)
378 # API: GET policy instance pi9, shall fail
379 response=client.get(SERVER_URL+'A1-P/v1/policies/pi9')
380 assert response.status_code == 404
382 # API: DELETE policy instance pi1
383 response=client.delete(SERVER_URL+'A1-P/v1/policies/pi1')
384 assert response.status_code == 204
386 # API: DELETE policy instance pi9, shall fail
387 response=client.delete(SERVER_URL+'A1-P/v1/policies/pi9')
388 assert response.status_code == 404
390 # API: Get policy status for pi1, shall fail
392 "title" : "The policy identity does not exist.",
394 "detail" : "There is no existing policy instance with the identity: pi1",
397 response=client.get(SERVER_URL+'A1-P/v1/policies/pi1/status')
398 assert response.status_code == 404
399 result=json.loads(response.data)
400 res=compare(data_get_errror, result)
403 # Set status for policy instance pi2
404 response=client.put(SERVER_URL+'status?policyid=pi2&status=OK')
405 assert response.status_code == 200
407 # API: Get policy status for pi2
409 "enforceStatus" : "OK"
411 response=client.get(SERVER_URL+'A1-P/v1/policies/pi2/status')
412 assert response.status_code == 200
413 result=json.loads(response.data)
414 res=compare(data_get_status, result)
417 # Set status for policy instance pi2
418 response=client.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason')
419 assert response.status_code == 200
421 # API: Get policy status for pi2
423 "enforceStatus" : "NOTOK",
424 "enforceReason" : "notok_reason"
426 response=client.get(SERVER_URL+'A1-P/v1/policies/pi2/status')
427 assert response.status_code == 200
428 result=json.loads(response.data)
429 res=compare(data_get_status, result)
432 #Found no way to test these functions
433 #'sendstatus' will send a http request that will fail
434 #since no server will receive the call
435 #These function is instead tested when running the bash script in the 'test' dir
436 # # Send status for pi2
437 # response=client.post(SERVER_URL+'sendstatus?policyid=pi2')
438 # assert response.status_code == 200
439 # result=json.loads(response.data)
440 # res=compare(data_get_status, result)
443 # # Send status, shall fail
444 # response=client.post(SERVER_URL+'sendstatus')
445 # assert response.status_code == 400
447 # # Send status pi9, shall fail
448 # response=client.post(SERVER_URL+'sendstatus?policyid=pi9')
449 # assert response.status_code == 404
451 # Set status for policy instance , shall fail
452 response=client.put(SERVER_URL+'status')
453 assert response.status_code == 400
455 # Set status for policy instance pi9, shall fail
456 response=client.put(SERVER_URL+'status?policyid=pi9')
457 assert response.status_code == 404
459 # Set status for policy instance pi2, shall fail
460 response=client.put(SERVER_URL+'status?policyid=pi2')
461 assert response.status_code == 400
464 # Get counter: intstance
465 response=client.get(SERVER_URL+'counter/num_instances')
466 assert response.status_code == 200
467 assert response.data == b"1"
469 # Get counter: types (shall be 0)
470 response=client.get(SERVER_URL+'counter/num_types')
471 assert response.status_code == 200
472 assert response.data == b"0"
474 # Get counter: interface
475 response=client.get(SERVER_URL+'counter/interface')
476 assert response.status_code == 200
477 assert response.data == b"STD_1.1.3"
479 # Get counter: remote hosts
480 response=client.get(SERVER_URL+'counter/remote_hosts')
481 assert response.status_code == 200
483 # Get counter: test, shall fail
484 response=client.get(SERVER_URL+'counter/test')
485 assert response.status_code == 404