near-rt-ric-simulator/tests/test_osc_2_1_0.py

   1 #  ============LICENSE_START===============================================
   2 #  Copyright (C) 2021-2023 Nordix Foundation. All rights reserved.
   3 #  Copyright (C) 2023-2024 OpenInfra Foundation Europe. All Rights Reserved
   4 #  ========================================================================
   5 #  Licensed under the Apache License, Version 2.0 (the "License");
   6 #  you may not use this file except in compliance with the License.
   7 #  You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #  Unless required by applicable law or agreed to in writing, software
  12 #  distributed under the License is distributed on an "AS IS" BASIS,
  13 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #  See the License for the specific language governing permissions and
  15 #  limitations under the License.
  16 #  ============LICENSE_END=================================================
  17 #
  18
  19 # This test case tests the OSC_2.1.0 version of the simulator
  20
  21
  22 # Version of simulator
  23 INTERFACE_VERSION="OSC_2.1.0"
  24
  25 import json
  26 import pytest
  27 import multiprocessing
  28 from unittest_setup import SERVER_URL, PORT_NUMBER, setup_env, get_testdata_dir, client
  29 from unittest_setup import run_flask_app
  30
  31 # Setup env and import paths
  32 setup_env(INTERFACE_VERSION)
  33
  34 from compare_json import compare
  35 from models.enforceStatus import EnforceStatus
  36
  37 def test_enforce_reason(client):
  38     """
  39     Test that we can set a valid enforce status and reason, and that we reject invalid cases.
  40     """
  41     enforceStatus = EnforceStatus()
  42
  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'
  48
  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'
  54
  55     enforceStatus.enforce_reason = 'OTHER_REASON'
  56     enforce_dict = enforceStatus.to_dict()
  57     assert enforce_dict['enforceReason'] == 'OTHER_REASON'
  58
  59     enforce_status = enforceStatus.enforce_status
  60     assert str(enforce_status) == 'ENFORCED'
  61
  62     enforce_reason = enforceStatus.enforce_reason
  63     assert str(enforce_reason) == 'OTHER_REASON'
  64
  65     with pytest.raises(ValueError):
  66         enforceStatus.enforce_status = 'ERROR'
  67
  68     with pytest.raises(ValueError):
  69         enforceStatus.enforce_reason = 'ERROR'
  70
  71
  72 def test_apis(client):
  73
  74     testdata=get_testdata_dir()
  75
  76     # Simulator hello world
  77     response=client.get(SERVER_URL)
  78     assert response.status_code == 200
  79
  80     # Check used and implemented interfaces
  81     response=client.get(SERVER_URL+'container_interfaces')
  82     assert response.status_code == 200
  83     assert response.data ==  b"Current interface: OSC_2.1.0  All supported A1 interface yamls in this container: ['OSC_2.1.0', 'STD_1.1.3', 'STD_2.0.0']"
  84
  85     # Reset simulator instances
  86     response=client.post(SERVER_URL+'deleteinstances')
  87     assert response.status_code == 200
  88
  89     # Reset simulator, all
  90     response=client.post(SERVER_URL+'deleteall')
  91     assert response.status_code == 200
  92
  93     # API: Healthcheck
  94     response=client.get(SERVER_URL+'a1-p/healthcheck')
  95     assert response.status_code == 200
  96
  97     # API: Get policy types, shall be empty
  98     data_policytypes_get = [ ]
  99     response=client.get(SERVER_URL+'a1-p/policytypes')
 100     assert response.status_code == 200
 101     result=json.loads(response.data)
 102     res=compare(data_policytypes_get, result)
 103     assert res == True
 104
 105     # API: Delete a policy type, shall fail
 106     response=client.delete(SERVER_URL+'a1-p/policytypes/1')
 107     assert response.status_code == 404
 108
 109     # API: Get policy instances for type 1, shall fail
 110     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
 111     assert response.status_code == 404
 112
 113     # Header for json payload
 114     header = {
 115         "Content-Type" : "application/json"
 116     }
 117
 118     # API: Put a policy type: 1
 119     with open(testdata+'pt1.json') as json_file:
 120         policytype_1 = json.load(json_file)
 121         response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
 122         assert response.status_code == 201
 123
 124     # API: Put a policy type: 1 again
 125     with open(testdata+'pt1.json') as json_file:
 126         policytype_1 = json.load(json_file)
 127         response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
 128         assert response.status_code == 201
 129
 130     # API: Delete a policy type
 131     response=client.delete(SERVER_URL+'a1-p/policytypes/1')
 132     assert response.status_code == 204
 133
 134     # API: Get policy type ids, shall be empty
 135     data_policytypes_get = [ ]
 136     response=client.get(SERVER_URL+'a1-p/policytypes')
 137     assert response.status_code == 200
 138     result=json.loads(response.data)
 139     res=compare(data_policytypes_get, result)
 140     assert res == True
 141
 142     # API: Put a policy type: 1
 143     with open(testdata+'pt1.json') as json_file:
 144         policytype_1 = json.load(json_file)
 145         response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
 146         assert response.status_code == 201
 147
 148     # API: Get policy type ids, shall contain '1'
 149     data_policytypes_get = [ 1 ]
 150     response=client.get(SERVER_URL+'a1-p/policytypes')
 151     assert response.status_code == 200
 152     result=json.loads(response.data)
 153     res=compare(data_policytypes_get, result)
 154     assert res == True
 155
 156     # API: Get instances for type 1, shall be empty
 157     data_policies_get = [ ]
 158     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
 159     assert response.status_code == 200
 160     result=json.loads(response.data)
 161     res=compare(data_policies_get, result)
 162     assert res == True
 163
 164     # API: Create policy instance pi1 of type: 1
 165     with open(testdata+'pi1.json') as json_file:
 166         policy_1 = json.load(json_file)
 167         response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi1', headers=header, data=json.dumps(policy_1))
 168         assert response.status_code == 202
 169
 170     # API: Get policy instance pi1 of type: 1
 171     with open(testdata+'pi1.json') as json_file:
 172         policy_1 = json.load(json_file)
 173         response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
 174         assert response.status_code == 200
 175         result=json.loads(response.data)
 176         res=compare(policy_1, result)
 177         assert res == True
 178
 179     # API: Update policy instance pi1 of type: 1
 180     with open(testdata+'pi1.json') as json_file:
 181         policy_1 = json.load(json_file)
 182         response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi1', headers=header, data=json.dumps(policy_1))
 183         assert response.status_code == 202
 184
 185     # API: Update policy type: 1, shall fail
 186     with open(testdata+'pt1.json') as json_file:
 187         policytype_1 = json.load(json_file)
 188         response=client.put(SERVER_URL+'a1-p/policytypes/1', headers=header, data=json.dumps(policytype_1))
 189         assert response.status_code == 400
 190
 191     # API: Get instances for type 1, shall contain 'pi1'
 192     data_policies_get = [ "pi1" ]
 193     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
 194     assert response.status_code == 200
 195     result=json.loads(response.data)
 196     res=compare(data_policies_get, result)
 197     assert res == True
 198
 199     # API: Create policy instance pi2 (copy of pi1) of type: 1.
 200     with open(testdata+'pi1.json') as json_file:
 201         policy_2 = json.load(json_file)
 202         response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi2', headers=header, data=json.dumps(policy_2))
 203         assert response.status_code == 202
 204
 205     # API: DELETE policy instance pi1
 206     response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi2')
 207     assert response.status_code == 202
 208
 209     # Set force response code 401
 210     response=client.post(SERVER_URL+'forceresponse?code=401')
 211     assert response.status_code == 200
 212
 213     # API: Get policy type 1. Shall fail with forced code
 214     response=client.get(SERVER_URL+'a1-p/policytypes/1')
 215     assert response.status_code == 401
 216
 217     # API: Get policy status
 218     policy_status = {
 219         "enforceStatus" : "NOT_ENFORCED",
 220         "enforceReason" : "OTHER_REASON",
 221     }
 222     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
 223     assert response.status_code == 200
 224     result=json.loads(response.data)
 225     res=compare(policy_status, result)
 226     assert res == True
 227
 228     # Load a policy type: 2
 229     with open(testdata+'pt2.json') as json_file:
 230         policytype_2 = json.load(json_file)
 231         response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
 232         assert response.status_code == 201
 233         assert response.data == b"Policy type 2 is OK."
 234
 235     # Load a policy type: 2, again
 236     with open(testdata+'pt2.json') as json_file:
 237         policytype_2 = json.load(json_file)
 238         response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
 239         assert response.status_code == 200
 240         assert response.data == b"Policy type 2 is OK."
 241
 242     # API: Get policy type ids, shall contain '1' and '2'
 243     data_policytypes_get = [ 1,2 ]
 244     response=client.get(SERVER_URL+'a1-p/policytypes')
 245     assert response.status_code == 200
 246     result=json.loads(response.data)
 247     res=compare(data_policytypes_get, result)
 248     assert res == True
 249
 250     # Get policy type ids, shall contain type 1 and 2 =="
 251     data_policytypes_get = [ "1","2" ]
 252     response=client.get(SERVER_URL+'policytypes')
 253     assert response.status_code == 200
 254     result=json.loads(response.data)
 255     res=compare(data_policytypes_get, result)
 256     assert res == True
 257
 258     # API: Get policy type 2
 259     with open(testdata+'pt2.json') as json_file:
 260         policytype_2 = json.load(json_file)
 261         response=client.get(SERVER_URL+'a1-p/policytypes/2')
 262         assert response.status_code == 200
 263         result=json.loads(response.data)
 264         res=compare(policytype_2, result)
 265         assert res == True
 266
 267     # Delete a policy type
 268     response=client.delete(SERVER_URL+'policytype?id=2')
 269     assert response.status_code == 204
 270
 271     # API: Get policy type ids, shall contain '1'
 272     data_policytypes_get = [ 1]
 273     response=client.get(SERVER_URL+'a1-p/policytypes')
 274     assert response.status_code == 200
 275     result=json.loads(response.data)
 276     res=compare(data_policytypes_get, result)
 277     assert res == True
 278
 279     # Load a policy type: 2
 280     with open(testdata+'pt2.json') as json_file:
 281         policytype_2 = json.load(json_file)
 282         response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_2))
 283         assert response.status_code == 201
 284         assert response.data == b"Policy type 2 is OK."
 285
 286     # API: Get policy type 2
 287     with open(testdata+'pt2.json') as json_file:
 288         policytype_2 = json.load(json_file)
 289         response=client.get(SERVER_URL+'a1-p/policytypes/2')
 290         assert response.status_code == 200
 291         result=json.loads(response.data)
 292         res=compare(policytype_2, result)
 293         assert res == True
 294
 295     # API: Get instances for type 2, shall be empty
 296     data_policies_get = [ ]
 297     response=client.get(SERVER_URL+'a1-p/policytypes/2/policies')
 298     assert response.status_code == 200
 299     result=json.loads(response.data)
 300     res=compare(data_policies_get, result)
 301     assert res == True
 302
 303     # API: Create policy instance pi1 of type: 2, shall fail
 304     with open(testdata+'pi1.json') as json_file:
 305         policy_1 = json.load(json_file)
 306         response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi1', headers=header, data=json.dumps(policy_1))
 307         assert response.status_code == 400
 308
 309     # API: Create policy instance pi2 of type: 2. Missing param, shall fail
 310     with open(testdata+'pi2_missing_param.json') as json_file:
 311         policy_2 = json.load(json_file)
 312         response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
 313         assert response.status_code == 400
 314
 315     # API: Create policy instance pi2 of type: 2
 316     with open(testdata+'pi2.json') as json_file:
 317         policy_2 = json.load(json_file)
 318         response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
 319         assert response.status_code == 202
 320
 321     with open(testdata+'pi2.json') as json_file:
 322         policy_2 = json.load(json_file)
 323         response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2))
 324         assert response.status_code == 202
 325
 326     # API: Get instances for type 1, shall contain pi1
 327     data_policies_get = [ "pi1" ]
 328     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
 329     assert response.status_code == 200
 330     result=json.loads(response.data)
 331     res=compare(data_policies_get, result)
 332     assert res == True
 333
 334     # API: Get instances for type 2, shall contain pi2
 335     data_policies_get = ["pi2" ]
 336     response=client.get(SERVER_URL+'a1-p/policytypes/2/policies')
 337     assert response.status_code == 200
 338     result=json.loads(response.data)
 339     res=compare(data_policies_get, result)
 340     assert res == True
 341
 342     # Set force response code 409. ==="
 343     response=client.post(SERVER_URL+'forceresponse?code=401')
 344     assert response.status_code == 200
 345
 346     # API: Get policy status for pi1, shall fail
 347     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
 348     assert response.status_code == 401
 349
 350     # Set force delay 10
 351     response=client.post(SERVER_URL+'forcedelay?delay=10')
 352     assert response.status_code == 200
 353     assert response.data ==  b"Force delay: 10 sec set for all A1 responses"
 354
 355     # API: Get policy status for pi1. Shall delay 10 sec
 356     policy_status = {
 357         "enforceStatus" : "NOT_ENFORCED",
 358         "enforceReason" : "OTHER_REASON",
 359     }
 360     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
 361     assert response.status_code == 200
 362     result=json.loads(response.data)
 363     res=compare(policy_status, result)
 364     assert res == True
 365
 366     # Reset force delay
 367     response=client.post(SERVER_URL+'forcedelay')
 368     assert response.status_code == 200
 369     assert response.data ==  b"Force delay: None sec set for all A1 responses"
 370
 371     #  Set status for pi1
 372     response=client.put(SERVER_URL+'status?policyid=pi1&status=ENFORCED')
 373     assert response.status_code == 200
 374
 375     # API: Get policy status for pi1
 376     policy_status = {
 377         "enforceStatus" : "ENFORCED",
 378         "enforceReason" : None,
 379     }
 380     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
 381     assert response.status_code == 200
 382     result=json.loads(response.data)
 383     res=compare(policy_status, result)
 384     assert res == True
 385
 386     #  Set status for pi1
 387     response=client.put(SERVER_URL+'status?policyid=pi1&status=NOT_ENFORCED&reason=SCOPE_NOT_APPLICABLE')
 388     assert response.status_code == 200
 389
 390     # API: Get policy status for pi1
 391     policy_status = {
 392         "enforceStatus" : "NOT_ENFORCED",
 393         "enforceReason" : "SCOPE_NOT_APPLICABLE",
 394     }
 395     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status')
 396     assert response.status_code == 200
 397     result=json.loads(response.data)
 398     res=compare(policy_status, result)
 399     assert res == True
 400
 401     # Get counter: num_instances
 402     response=client.get(SERVER_URL+'counter/num_instances')
 403     assert response.status_code == 200
 404     assert response.data ==  b"2"
 405
 406     # Get counter: types (shall be 2)
 407     response=client.get(SERVER_URL+'counter/num_types')
 408     assert response.status_code == 200
 409     assert response.data ==  b"2"
 410
 411     # Get counter: interface
 412     response=client.get(SERVER_URL+'counter/interface')
 413     assert response.status_code == 200
 414     assert response.data == b"OSC_2.1.0"
 415
 416     # Get counter: remote hosts
 417     response=client.get(SERVER_URL+'counter/remote_hosts')
 418     assert response.status_code == 200
 419
 420     # Get counter: test, shall fail
 421     response=client.get(SERVER_URL+'counter/test')
 422     assert response.status_code == 404
 423
 424     # API: DELETE policy instance pi1
 425     response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
 426     assert response.status_code == 202
 427
 428     # API: Get instances for type 1, shall be empty
 429     data_policies_get = [ ]
 430     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
 431     assert response.status_code == 200
 432     result=json.loads(response.data)
 433     res=compare(data_policies_get, result)
 434     assert res == True
 435
 436     # API: Get instances for type 2, shall contain pi2
 437     data_policies_get = ["pi2" ]
 438     response=client.get(SERVER_URL+'a1-p/policytypes/2/policies')
 439     assert response.status_code == 200
 440     result=json.loads(response.data)
 441     res=compare(data_policies_get, result)
 442     assert res == True
 443
 444     # Get counter: instances
 445     response=client.get(SERVER_URL+'counter/num_instances')
 446     assert response.status_code == 200
 447     assert response.data ==  b"1"
 448
 449
 450     ### Tests to increase code coverage
 451
 452     # Set force response code 500
 453     response=client.post(SERVER_URL+'forceresponse?code=500')
 454     assert response.status_code == 200
 455
 456     # API: Healthcheck
 457     response=client.get(SERVER_URL+'a1-p/healthcheck')
 458     assert response.status_code == 500
 459
 460     # Set force response code 501
 461     response=client.post(SERVER_URL+'forceresponse?code=501')
 462     assert response.status_code == 200
 463
 464     # API: Get policy types
 465     data_policytypes_get = [ ]
 466     response=client.get(SERVER_URL+'a1-p/policytypes')
 467     assert response.status_code == 501
 468
 469     # Set force response code 502
 470     response=client.post(SERVER_URL+'forceresponse?code=502')
 471     assert response.status_code == 200
 472
 473     # API: Delete a policy type, shall fail
 474     response=client.delete(SERVER_URL+'a1-p/policytypes/55')
 475     assert response.status_code == 502
 476
 477     # Set force response code 503. ==="
 478     response=client.post(SERVER_URL+'forceresponse?code=503')
 479     assert response.status_code == 200
 480
 481     with open(testdata+'pi1.json') as json_file:
 482         policy_1 = json.load(json_file)
 483         response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi11', headers=header, data=json.dumps(policy_1))
 484         assert response.status_code == 503
 485
 486     # Set force response code 504
 487     response=client.post(SERVER_URL+'forceresponse?code=504')
 488     assert response.status_code == 200
 489
 490     # API: Get instances for type 1, shall fail
 491     data_policies_get = [ ]
 492     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies')
 493     assert response.status_code == 504
 494
 495     # Set force response code 505. ==="
 496     response=client.post(SERVER_URL+'forceresponse?code=505')
 497     assert response.status_code == 200
 498
 499     # API: delete instance pi1, shall fail
 500     response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi1')
 501     assert response.status_code == 505
 502
 503     # API: Delete a policy type having instances, shall fail
 504     response=client.delete(SERVER_URL+'a1-p/policytypes/2')
 505     assert response.status_code == 400
 506
 507     # API: delete instance pi1 in type 5, shall fail
 508     response=client.delete(SERVER_URL+'a1-p/policytypes/5/policies/pi1')
 509     assert response.status_code == 404
 510
 511     # API: delete instance pi99 in type 1, shall fail
 512     response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi99')
 513     assert response.status_code == 404
 514
 515     # API: Create policy instance pi80 of type: 5
 516     with open(testdata+'pi1.json') as json_file:
 517         policy_80 = json.load(json_file)
 518         response=client.put(SERVER_URL+'a1-p/policytypes/5/policies/pi80', headers=header, data=json.dumps(policy_80))
 519         assert response.status_code == 404
 520
 521     # API: Get policy type
 522     data_policytypes_get = [ ]
 523     response=client.get(SERVER_URL+'a1-p/policytypes/55')
 524     assert response.status_code == 404
 525
 526     # API: Get status, bad type - shall fail
 527     response=client.get(SERVER_URL+'a1-p/policytypes/99/policies/pi1/status')
 528     assert response.status_code == 404
 529
 530     # API: Get status, bad instance - shall fail
 531     response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi111/status')
 532     assert response.status_code == 404
 533
 534     # Load policy type, no type in url - shall fail
 535     with open(testdata+'pt2.json') as json_file:
 536         policytype_2 = json.load(json_file)
 537         response=client.put(SERVER_URL+'policytype', headers=header, data=json.dumps(policytype_2))
 538         assert response.status_code == 400
 539
 540     # Load policy type - duplicatee - shall fail
 541     with open(testdata+'pt1.json') as json_file:
 542         policytype_1 = json.load(json_file)
 543         response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_1))
 544         assert response.status_code == 400
 545
 546     # Get counter: data_delivery
 547     response=client.get(SERVER_URL+'counter/datadelivery')
 548     assert response.status_code == 200
 549     assert response.data ==  b"0"
 550
 551     # Send data to data-delivery with empty payload
 552     json_payload={}
 553     response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
 554     assert response.status_code == 400
 555
 556     # Send invalid data to data-delivery
 557     json_payload={
 558         "job":"200",
 559         "payload":"payload"
 560     }
 561     response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
 562     assert response.status_code == 404
 563
 564     # Send data to data-delivery with valid job
 565     json_payload={
 566         "job":"100",
 567         "payload":"payload"
 568     }
 569     response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
 570     assert response.status_code == 200
 571
 572     # Send data to data-delivery with valid job
 573     json_payload={
 574         "job":"101",
 575         "payload":"another payload"
 576     }
 577     response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload))
 578     assert response.status_code == 200
 579
 580     # Get counter: data_delivery
 581     response=client.get(SERVER_URL+'counter/datadelivery')
 582     assert response.status_code == 200
 583     assert response.data ==  b"2"
 584
 585 def test_notificationDestination(client):
 586     test_data = get_testdata_dir() + 'pi2.json'
 587     # Header for json payload
 588     header = { "Content-Type" : "application/json" }
 589
 590     # === API: Update policy instance pi2 of type: 2 ==="
 591     with open(test_data) as json_file:
 592         payload = json.load(json_file)
 593         response = client.put(SERVER_URL+"a1-p/policytypes/2/policies/pi2?notificationDestination=http://localhost:8086/statustest", headers=header, data=json.dumps(payload))
 594
 595     assert response.status_code == 202
 596     result = response.data
 597     assert result == b""
 598
 599
 600 def test_sendstatus(client):
 601     # Create a new thread to run the Flask app in parallel on a different port so that we can call the callback.
 602     proc = multiprocessing.Process(target=run_flask_app, args=())
 603     proc.start()
 604
 605     test_data = get_testdata_dir() + 'pi2.json'
 606     header = { "Content-Type" : "application/json" }
 607
 608     # Timeout can be removed with polling the endpoints if required
 609     proc.join(timeout=10)
 610
 611     # === Send status for pi2===
 612     with open(test_data) as json_file:
 613         payload = json.load(json_file)
 614         response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(payload))
 615
 616     assert response.status_code == 201
 617     result = response.data
 618     assert result == b"OK"
 619
 620     # Send status, negative test with missing parameter
 621     response = client.post(SERVER_URL+'sendstatus', headers=header, data="")
 622     assert response.status_code == 400
 623
 624     # Send status pi9, negative test for policy id not found
 625     response = client.post(SERVER_URL+'sendstatus?policyid=pi9', headers=header, data="")
 626     assert response.status_code == 404
 627
 628     proc.terminate()