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.
23 import multiprocessing
24 from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client
25 from unittest_setup import run_flask_app
27 # Setup env and import paths
29 # Version of simulator
30 INTERFACE_VERSION="STD_2.0.0"
32 setup_env(INTERFACE_VERSION)
33 from compare_json import compare
35 from models.enforceStatus import EnforceStatus
37 def test_enforce_reason(client):
39 Test that we can set a valid enforce status and reason, and that we reject invalid cases.
41 enforceStatus = EnforceStatus()
43 enforceStatus.enforce_status = 'NOT_ENFORCED'
44 enforceStatus.enforce_reason = 'SCOPE_NOT_APPLICABLE'
45 enforce_dict = enforceStatus.to_dict()
46 assert enforce_dict['enforceStatus'] == 'NOT_ENFORCED'
47 assert enforce_dict['enforceReason'] == 'SCOPE_NOT_APPLICABLE'
49 enforceStatus.enforce_status = 'ENFORCED'
50 enforceStatus.enforce_reason = 'STATEMENT_NOT_APPLICABLE'
51 enforce_dict = enforceStatus.to_dict()
52 assert enforce_dict['enforceStatus'] == 'ENFORCED'
53 assert enforce_dict['enforceReason'] == 'STATEMENT_NOT_APPLICABLE'
55 enforceStatus.enforce_reason = 'OTHER_REASON'
56 enforce_dict = enforceStatus.to_dict()
57 assert enforce_dict['enforceReason'] == 'OTHER_REASON'
59 enforce_status = enforceStatus.enforce_status
60 assert str(enforce_status) == 'ENFORCED'
62 enforce_reason = enforceStatus.enforce_reason
63 assert str(enforce_reason) == 'OTHER_REASON'
65 with pytest.raises(ValueError):
66 enforceStatus.enforce_status = 'ERROR'
68 with pytest.raises(ValueError):
69 enforceStatus.enforce_reason = 'ERROR'
72 def test_apis(client):
74 testdata=get_testdata_dir()
76 # Header for json payload
78 "Content-Type" : "application/json"
81 # Simulator hello world
82 response=client.get(SERVER_URL)
83 assert response.status_code == 200
85 # Check used and implemented interfaces
86 response=client.get(SERVER_URL+'container_interfaces')
87 assert response.status_code == 200
88 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']"
90 # Reset simulator instances
91 response=client.post(SERVER_URL+'deleteinstances')
92 assert response.status_code == 200
94 # Reset simulator, all
95 response=client.post(SERVER_URL+'deleteall')
96 assert response.status_code == 200
98 # Get counter: interface
99 response=client.get(SERVER_URL+'counter/interface')
100 assert response.status_code == 200
101 assert response.data == b"STD_2.0.0"
103 # Get counter: remote hosts
104 response=client.get(SERVER_URL+'counter/remote_hosts')
105 assert response.status_code == 200
107 # Get counter: intstance
108 response=client.get(SERVER_URL+'counter/num_instances')
109 assert response.status_code == 200
110 assert response.data == b"0"
113 response=client.get(SERVER_URL+'counter/num_types')
114 assert response.status_code == 200
115 assert response.data == b"0"
117 # API: Get policy type, shall be empty
119 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
120 assert response.status_code == 200
121 result=json.loads(response.data)
122 res=compare(data_response, result)
125 # API: Get policy instances for type 1, type not found
126 data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "1"}
127 response=client.get(SERVER_URL+'A1-P/v2/policytypes/1/policies')
128 assert response.status_code == 404
129 result=json.loads(response.data)
130 res=compare(data_response, result)
133 # API: Get policy instances, type not found
134 data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "test"}
135 response=client.get(SERVER_URL+'A1-P/v2/policytypes/test/policies')
136 assert response.status_code == 404
137 result=json.loads(response.data)
138 res=compare(data_response, result)
141 # Put a policy type: STD_1
142 with open(testdata+'std_1.json') as json_file:
143 data_response = b"Policy type STD_1 is OK."
144 json_payload=json.load(json_file)
145 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
146 assert response.status_code == 201
147 assert data_response == response.data
149 # Put a policy type: STD_1, again
150 with open(testdata+'std_1.json') as json_file:
151 data_response = b"Policy type STD_1 is OK."
152 json_payload=json.load(json_file)
153 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
154 assert response.status_code == 200
155 assert data_response == response.data
157 # API: Get policy type ids, shall contain type STD_1
158 data_response = [ "STD_1" ]
159 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
160 assert response.status_code == 200
161 result=json.loads(response.data)
162 res=compare(data_response, result)
165 # Delete a policy type: STD_1
167 response=client.delete(SERVER_URL+'policytype?id=STD_1')
168 assert response.status_code == 204
169 assert data_response == response.data
171 # API: Get policy type ids, shall be empty
173 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
174 assert response.status_code == 200
175 result=json.loads(response.data)
176 res=compare(data_response, result)
179 # Put a policy type: STD_1
180 with open(testdata+'std_1.json') as json_file:
181 data_response = b"Policy type STD_1 is OK."
182 json_payload=json.load(json_file)
183 response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
184 assert response.status_code == 201
185 assert data_response == response.data
187 # API: Get policy type ids, shall contain type STD_1
188 data_response = [ "STD_1" ]
189 response=client.get(SERVER_URL+'A1-P/v2/policytypes')
190 assert response.status_code == 200
191 result=json.loads(response.data)
192 res=compare(data_response, result)
195 # Get counter: types (shall be 1)
196 response=client.get(SERVER_URL+'counter/num_types')
197 assert response.status_code == 200
198 assert response.data == b"1"
200 # API: Get policy type: STD_1
201 with open(testdata+'std_1.json') as json_file:
202 data_response = json.load(json_file)
203 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1')
204 assert response.status_code == 200
205 result=json.loads(response.data)
206 res=compare(data_response, result)
209 # API: API: Get policy instances, shall be empty
211 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
212 assert response.status_code == 200
213 result=json.loads(response.data)
214 res=compare(data_response, result)
217 # API: Create policy instance pi1 of type: STD_1
218 with open(testdata+'pi1.json') as json_file:
219 json_payload=json.load(json_file)
220 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', 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 # API: API: Get policy instance pi1 of type: STD_1
227 with open(testdata+'pi1.json') as json_file:
228 data_response = json.load(json_file)
229 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
230 assert response.status_code == 200
231 result=json.loads(response.data)
232 res=compare(data_response, result)
235 # API: Update policy instance pi1 of type: STD_1
236 with open(testdata+'pi1.json') as json_file:
237 json_payload=json.load(json_file)
238 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
239 assert response.status_code == 200
240 result=json.loads(response.data)
241 res=compare(json_payload, result)
244 # API: Update policy instance pi1 of type: STD_1
245 with open(testdata+'pi1_updated.json') as json_file:
246 json_payload=json.load(json_file)
247 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
248 assert response.status_code == 200
249 result=json.loads(response.data)
250 res=compare(json_payload, result)
253 # # API: Duplicate policy instance pi2 of type: STD_1 - and delete it
255 with open(testdata+'pi1_updated.json') as json_file:
256 json_payload=json.load(json_file)
257 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
258 assert response.status_code == 201
259 result=json.loads(response.data)
260 res=compare(json_payload, result)
263 response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
264 assert response.status_code == 204
265 assert data_response == response.data
268 # API: Get policy instances, shall contain pi1
269 data_response = ["pi1"]
270 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
271 assert response.status_code == 200
272 result=json.loads(response.data)
273 res=compare(data_response, result)
276 # Get counter: intstance
277 response=client.get(SERVER_URL+'counter/num_instances')
278 assert response.status_code == 200
279 assert response.data == b"1"
282 response=client.get(SERVER_URL+'counter/num_types')
283 assert response.status_code == 200
284 assert response.data == b"1"
286 # Set force response code 409. ==="
287 response=client.post(SERVER_URL+'forceresponse?code=409')
288 assert response.status_code == 200
290 # API: Get policy instances, shall fail
291 data_response = {"title" : "Conflict", "status" : 409, "detail" : "Request could not be processed in the current state of the resource"}
292 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
293 assert response.status_code == 409
294 result=json.loads(response.data)
295 res=compare(data_response, result)
298 # API: API: Get policy status
299 data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"}
300 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1/status')
301 assert response.status_code == 200
302 result=json.loads(response.data)
303 res=compare(data_response, result)
306 # API: Create policy instance pi2 of type: STD_1
307 with open(testdata+'pi2.json') as json_file:
308 json_payload=json.load(json_file)
309 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
310 assert response.status_code == 201
311 result=json.loads(response.data)
312 res=compare(json_payload, result)
315 # API: Update policy instance pi2 of type: STD_1
316 with open(testdata+'pi2.json') as json_file:
317 json_payload=json.load(json_file)
318 response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
319 assert response.status_code == 200
320 result=json.loads(response.data)
321 res=compare(json_payload, result)
324 # API: Get policy instances, shall contain pi1 and pi2
325 data_response = ["pi1","pi2"]
326 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
327 assert response.status_code == 200
328 result=json.loads(response.data)
329 res=compare(data_response, result)
332 # Get counter: intstance
333 response=client.get(SERVER_URL+'counter/num_instances')
334 assert response.status_code == 200
335 assert response.data == b"2"
338 response=client.get(SERVER_URL+'counter/num_types')
339 assert response.status_code == 200
340 assert response.data == b"1"
343 response=client.post(SERVER_URL+'forcedelay?delay=10')
344 assert response.status_code == 200
345 assert response.data == b"Force delay: 10 sec set for all A1 responses"
350 # API: Get policy instances, shall contain pi1 and pi2 and delayed 10 sec
351 data_response = ["pi1","pi2"]
352 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
353 assert response.status_code == 200
354 result=json.loads(response.data)
355 res=compare(data_response, result)
360 assert (end-start) > 9
363 response=client.post(SERVER_URL+'forcedelay')
364 assert response.status_code == 200
365 assert response.data == b"Force delay: None sec set for all A1 responses"
367 # API: API: Get policy instance pi1 of type: STD_1
368 with open(testdata+'pi1_updated.json') as json_file:
369 data_response = json.load(json_file)
370 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
371 assert response.status_code == 200
372 result=json.loads(response.data)
373 res=compare(data_response, result)
376 # API: API: Get policy instance pi2 of type: STD_1
377 with open(testdata+'pi2.json') as json_file:
378 data_response = json.load(json_file)
379 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
380 assert response.status_code == 200
381 result=json.loads(response.data)
382 res=compare(data_response, result)
385 # API: DELETE policy instance pi1
387 response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
388 assert response.status_code == 204
389 assert data_response == response.data
391 # API: Get policy instances, shall contain pi2
392 data_response = ["pi2"]
393 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
394 assert response.status_code == 200
395 result=json.loads(response.data)
396 res=compare(data_response, result)
399 # API: API: Get policy status
400 data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"}
401 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
402 assert response.status_code == 200
403 result=json.loads(response.data)
404 res=compare(data_response, result)
407 # Set status for policy instance pi2
408 response=client.put(SERVER_URL+'status?policyid=pi2&status=NOT_ENFORCED&reason=STATEMENT_NOT_APPLICABLE')
409 assert response.status_code == 200
411 # API: API: Get policy status
412 data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "STATEMENT_NOT_APPLICABLE"}
413 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
414 assert response.status_code == 200
415 result=json.loads(response.data)
416 res=compare(data_response, result)
420 # Set status for policy instance pi2
421 response=client.put(SERVER_URL+'status?policyid=pi2&status=NOT_ENFORCED&reason=OTHER_REASON')
422 assert response.status_code == 200
424 # API: API: Get policy status
425 data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"}
426 response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
427 assert response.status_code == 200
428 result=json.loads(response.data)
429 res=compare(data_response, result)
432 # Get counter: data_delivery
433 response=client.get(SERVER_URL+'counter/datadelivery')
434 assert response.status_code == 200
435 assert response.data == b"0"
439 response=client.post(SERVER_URL+'datadelivery', headers=header, data=json.dumps(json_payload))
440 assert response.status_code == 200
442 # Get counter: data_delivery
443 response=client.get(SERVER_URL+'counter/datadelivery')
444 assert response.status_code == 200
445 assert response.data == b"1"
447 # Get counter: interface
448 response=client.get(SERVER_URL+'counter/interface')
449 assert response.status_code == 200
450 assert response.data == b"STD_2.0.0"
452 # Get counter: remote hosts
453 response=client.get(SERVER_URL+'counter/remote_hosts')
454 assert response.status_code == 200
456 # Get counter: intstance
457 response=client.get(SERVER_URL+'counter/num_instances')
458 assert response.status_code == 200
459 assert response.data == b"1"
462 response=client.get(SERVER_URL+'counter/num_types')
463 assert response.status_code == 200
464 assert response.data == b"1"
466 def test_notificationDestination(client):
467 test_data = get_testdata_dir() + 'pi2.json'
468 # Header for json payload
469 header = { "Content-Type" : "application/json" }
471 # === API: Update policy instance pi2 of type: 2 ==="
472 with open(test_data) as json_file:
473 payload = json.load(json_file)
474 response = client.put(SERVER_URL+"A1-P/v2/policytypes/STD_1/policies/pi2?notificationDestination=http://localhost:8086/statustest", headers=header, data=json.dumps(payload))
475 assert response.status_code == 200
476 result = json.loads(response.data)
477 assert compare(payload, result) == True
479 def test_sendstatus(client):
480 # Create a new thread to run the Flask app in parallel on a different port so that we can call the callback.
481 proc = multiprocessing.Process(target=run_flask_app, args=())
484 test_data = get_testdata_dir() + 'pi2.json'
485 header = { "Content-Type" : "application/json" }
487 # === Send status for pi2===
488 with open(test_data) as json_file:
489 payload = json.load(json_file)
490 response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(payload))
492 assert response.status_code == 204
493 result = response.data
496 # Send status, negative test with missing parameter
497 response = client.post(SERVER_URL+'sendstatus', headers=header, data="")
498 assert response.status_code == 400
500 # Send status pi9, negative test for policy id not found
501 response = client.post(SERVER_URL+'sendstatus?policyid=pi9', headers=header, data="")
502 assert response.status_code == 404