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 tests the STD_2.0.0 version of the simulator.
22 import multiprocessing
23 from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client
24 from unittest_setup import run_flask_app
26 # Setup env and import paths
28 # Version of simulator
29 INTERFACE_VERSION="STD_2.0.0"
31 setup_env(INTERFACE_VERSION)
32 from compare_json import compare
35 def test_apis(client):
37 testdata=get_testdata_dir()
39 # Header for json payload
41 "Content-Type" : "application/json"
44 # Simulator hello world
45 response=client.get(SERVER_URL)
46 assert response.status_code == 200
48 # Check used and implemented interfaces
49 response=client.get(SERVER_URL+'container_interfaces')
50 assert response.status_code == 200
51 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']"
53 # Reset simulator instances
54 response=client.post(SERVER_URL+'deleteinstances')
55 assert response.status_code == 200
57 # Reset simulator, all
58 response=client.post(SERVER_URL+'deleteall')
59 assert response.status_code == 200
61 # Get counter: interface
62 response=client.get(SERVER_URL+'counter/interface')
63 assert response.status_code == 200
64 assert response.data == b"STD_2.0.0"
66 # Get counter: remote hosts
67 response=client.get(SERVER_URL+'counter/remote_hosts')
68 assert response.status_code == 200
70 # Get counter: intstance
71 response=client.get(SERVER_URL+'counter/num_instances')
72 assert response.status_code == 200
73 assert response.data == b"0"
76 response=client.get(SERVER_URL+'counter/num_types')
77 assert response.status_code == 200
78 assert response.data == b"0"
80 # API: Get policy type, shall be empty
82 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
83 assert response.status_code == 200
84 result=json.loads(response.data)
85 res=compare(data_response, result)
88 # API: Get policy instances for type 1, type not found
89 data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "1"}
90 response=client.get(SERVER_URL+'A1-P/v2/policytypes/1/policies')
91 assert response.status_code == 404
92 result=json.loads(response.data)
93 res=compare(data_response, result)
96 # API: Get policy instances, type not found
97 data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "test"}
98 response=client.get(SERVER_URL+'A1-P/v2/policytypes/test/policies')
99 assert response.status_code == 404
100 result=json.loads(response.data)
101 res=compare(data_response, result)
104 # Put a policy type: STD_1
105 with open(testdata+'std_1.json') as json_file:
106 data_response = b"Policy type STD_1 is OK."
107 json_payload=json.load(json_file)
108 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
109 assert response.status_code == 201
110 assert data_response == response.data
112 # Put a policy type: STD_1, again
113 with open(testdata+'std_1.json') as json_file:
114 data_response = b"Policy type STD_1 is OK."
115 json_payload=json.load(json_file)
116 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
117 assert response.status_code == 200
118 assert data_response == response.data
120 # API: Get policy type ids, shall contain type STD_1
121 data_response = [ "STD_1" ]
122 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
123 assert response.status_code == 200
124 result=json.loads(response.data)
125 res=compare(data_response, result)
128 # Delete a policy type: STD_1
130 response=client.delete(SERVER_URL+'policytype?id=STD_1')
131 assert response.status_code == 204
132 assert data_response == response.data
134 # API: Get policy type ids, shall be empty
136 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
137 assert response.status_code == 200
138 result=json.loads(response.data)
139 res=compare(data_response, result)
142 # Put a policy type: STD_1
143 with open(testdata+'std_1.json') as json_file:
144 data_response = b"Policy type STD_1 is OK."
145 json_payload=json.load(json_file)
146 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
147 assert response.status_code == 201
148 assert data_response == response.data
150 # API: Get policy type ids, shall contain type STD_1
151 data_response = [ "STD_1" ]
152 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
153 assert response.status_code == 200
154 result=json.loads(response.data)
155 res=compare(data_response, result)
158 # Get counter: types (shall be 1)
159 response=client.get(SERVER_URL+'counter/num_types')
160 assert response.status_code == 200
161 assert response.data == b"1"
163 # API: Get policy type: STD_1
164 with open(testdata+'std_1.json') as json_file:
165 data_response = json.load(json_file)
166 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1')
167 assert response.status_code == 200
168 result=json.loads(response.data)
169 res=compare(data_response, result)
172 # API: API: Get policy instances, shall be empty
174 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
175 assert response.status_code == 200
176 result=json.loads(response.data)
177 res=compare(data_response, result)
180 # API: Create policy instance pi1 of type: STD_1
181 with open(testdata+'pi1.json') as json_file:
182 json_payload=json.load(json_file)
183 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
184 assert response.status_code == 201
185 result=json.loads(response.data)
186 res=compare(json_payload, result)
189 # API: API: Get policy instance pi1 of type: STD_1
190 with open(testdata+'pi1.json') as json_file:
191 data_response = json.load(json_file)
192 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
193 assert response.status_code == 200
194 result=json.loads(response.data)
195 res=compare(data_response, result)
198 # API: Update policy instance pi1 of type: STD_1
199 with open(testdata+'pi1.json') as json_file:
200 json_payload=json.load(json_file)
201 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
202 assert response.status_code == 200
203 result=json.loads(response.data)
204 res=compare(json_payload, result)
207 # API: Update policy instance pi1 of type: STD_1
208 with open(testdata+'pi1_updated.json') as json_file:
209 json_payload=json.load(json_file)
210 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
211 assert response.status_code == 200
212 result=json.loads(response.data)
213 res=compare(json_payload, result)
216 # # API: Duplicate policy instance pi2 of type: STD_1 - and delete it
218 with open(testdata+'pi1_updated.json') as json_file:
219 json_payload=json.load(json_file)
220 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
221 assert response.status_code == 201
222 result=json.loads(response.data)
223 res=compare(json_payload, result)
226 response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
227 assert response.status_code == 204
228 assert data_response == response.data
231 # API: Get policy instances, shall contain pi1
232 data_response = ["pi1"]
233 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
234 assert response.status_code == 200
235 result=json.loads(response.data)
236 res=compare(data_response, result)
239 # Get counter: intstance
240 response=client.get(SERVER_URL+'counter/num_instances')
241 assert response.status_code == 200
242 assert response.data == b"1"
245 response=client.get(SERVER_URL+'counter/num_types')
246 assert response.status_code == 200
247 assert response.data == b"1"
249 # Set force response code 409. ==="
250 response=client.post(SERVER_URL+'forceresponse?code=409')
251 assert response.status_code == 200
253 # API: Get policy instances, shall fail
254 data_response = {"title" : "Conflict", "status" : 409, "detail" : "Request could not be processed in the current state of the resource"}
255 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
256 assert response.status_code == 409
257 result=json.loads(response.data)
258 res=compare(data_response, result)
261 # API: API: Get policy status
262 data_response = {"enforceStatus" : "", "enforceReason" : ""}
263 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1/status')
264 assert response.status_code == 200
265 result=json.loads(response.data)
266 res=compare(data_response, result)
269 # API: Create policy instance pi2 of type: STD_1
270 with open(testdata+'pi2.json') as json_file:
271 json_payload=json.load(json_file)
272 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
273 assert response.status_code == 201
274 result=json.loads(response.data)
275 res=compare(json_payload, result)
278 # API: Update policy instance pi2 of type: STD_1
279 with open(testdata+'pi2.json') as json_file:
280 json_payload=json.load(json_file)
281 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
282 assert response.status_code == 200
283 result=json.loads(response.data)
284 res=compare(json_payload, result)
287 # API: Get policy instances, shall contain pi1 and pi2
288 data_response = ["pi1","pi2"]
289 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
290 assert response.status_code == 200
291 result=json.loads(response.data)
292 res=compare(data_response, result)
295 # Get counter: intstance
296 response=client.get(SERVER_URL+'counter/num_instances')
297 assert response.status_code == 200
298 assert response.data == b"2"
301 response=client.get(SERVER_URL+'counter/num_types')
302 assert response.status_code == 200
303 assert response.data == b"1"
306 response=client.post(SERVER_URL+'forcedelay?delay=10')
307 assert response.status_code == 200
308 assert response.data == b"Force delay: 10 sec set for all A1 responses"
313 # API: Get policy instances, shall contain pi1 and pi2 and delayed 10 sec
314 data_response = ["pi1","pi2"]
315 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
316 assert response.status_code == 200
317 result=json.loads(response.data)
318 res=compare(data_response, result)
323 assert (end-start) > 9
326 response=client.post(SERVER_URL+'forcedelay')
327 assert response.status_code == 200
328 assert response.data == b"Force delay: None sec set for all A1 responses"
330 # API: API: Get policy instance pi1 of type: STD_1
331 with open(testdata+'pi1_updated.json') as json_file:
332 data_response = json.load(json_file)
333 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
334 assert response.status_code == 200
335 result=json.loads(response.data)
336 res=compare(data_response, result)
339 # API: API: Get policy instance pi2 of type: STD_1
340 with open(testdata+'pi2.json') as json_file:
341 data_response = json.load(json_file)
342 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
343 assert response.status_code == 200
344 result=json.loads(response.data)
345 res=compare(data_response, result)
348 # API: DELETE policy instance pi1
350 response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
351 assert response.status_code == 204
352 assert data_response == response.data
354 # API: Get policy instances, shall contain pi2
355 data_response = ["pi2"]
356 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
357 assert response.status_code == 200
358 result=json.loads(response.data)
359 res=compare(data_response, result)
362 # API: API: Get policy status
363 data_response = {"enforceStatus" : "", "enforceReason" : ""}
364 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
365 assert response.status_code == 200
366 result=json.loads(response.data)
367 res=compare(data_response, result)
370 # Set status for policy instance pi2
371 response=client.put(SERVER_URL+'status?policyid=pi2&status=OK')
372 assert response.status_code == 200
374 # API: API: Get policy status
375 data_response = {"enforceStatus" : "OK"}
376 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
377 assert response.status_code == 200
378 result=json.loads(response.data)
379 res=compare(data_response, result)
383 # Set status for policy instance pi2
384 response=client.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason')
385 assert response.status_code == 200
387 # API: API: Get policy status
388 data_response = {"enforceStatus" : "NOTOK", "enforceReason" : "notok_reason"}
389 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
390 assert response.status_code == 200
391 result=json.loads(response.data)
392 res=compare(data_response, result)
395 # Get counter: data_delivery
396 response=client.get(SERVER_URL+'counter/datadelivery')
397 assert response.status_code == 200
398 assert response.data == b"0"
402 response=client.post(SERVER_URL+'datadelivery', headers=header, data=json.dumps(json_payload))
403 assert response.status_code == 200
405 # Get counter: data_delivery
406 response=client.get(SERVER_URL+'counter/datadelivery')
407 assert response.status_code == 200
408 assert response.data == b"1"
410 # Get counter: interface
411 response=client.get(SERVER_URL+'counter/interface')
412 assert response.status_code == 200
413 assert response.data == b"STD_2.0.0"
415 # Get counter: remote hosts
416 response=client.get(SERVER_URL+'counter/remote_hosts')
417 assert response.status_code == 200
419 # Get counter: intstance
420 response=client.get(SERVER_URL+'counter/num_instances')
421 assert response.status_code == 200
422 assert response.data == b"1"
425 response=client.get(SERVER_URL+'counter/num_types')
426 assert response.status_code == 200
427 assert response.data == b"1"
429 def test_notificationDestination(client):
430 test_data = get_testdata_dir() + 'pi2.json'
431 # Header for json payload
432 header = { "Content-Type" : "application/json" }
434 # === API: Update policy instance pi2 of type: 2 ==="
435 with open(test_data) as json_file:
436 payload = json.load(json_file)
437 response = client.put(SERVER_URL+"A1-P/v2/policytypes/STD_1/policies/pi2?notificationDestination=http://localhost:8085/statustest", headers=header, data=json.dumps(payload))
438 assert response.status_code == 200
439 result = json.loads(response.data)
440 assert compare(payload, result) == True
442 def test_sendstatus(client):
443 # Create a new thread to run the Flask app in parallel on a different port so that we can call the callback.
444 proc = multiprocessing.Process(target=run_flask_app, args=())
447 test_data = get_testdata_dir() + 'pi2.json'
448 header = { "Content-Type" : "application/json" }
450 # === Send status for pi2===
451 with open(test_data) as json_file:
452 payload = json.load(json_file)
453 response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(payload))
455 assert response.status_code == 204
456 result = response.data
459 # Send status, negative test with missing parameter
460 response = client.post(SERVER_URL+'sendstatus', headers=header, data="")
461 assert response.status_code == 400
463 # Send status pi9, negative test for policy id not found
464 response = client.post(SERVER_URL+'sendstatus?policyid=pi9', headers=header, data="")
465 assert response.status_code == 404