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

   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
   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 tests the STD_2.0.0 version of the simulator.
  19
  20 import json
  21 import pytest
  22 import time
  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
  26
  27 # Setup env and import paths
  28
  29 # Version of simulator
  30 INTERFACE_VERSION="STD_2.0.0"
  31
  32 setup_env(INTERFACE_VERSION)
  33 from compare_json import compare
  34
  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     # Header for json payload
  77     header = {
  78         "Content-Type" : "application/json"
  79     }
  80
  81     # Simulator hello world
  82     response=client.get(SERVER_URL)
  83     assert response.status_code == 200
  84
  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']"
  89
  90     # Reset simulator instances
  91     response=client.post(SERVER_URL+'deleteinstances')
  92     assert response.status_code == 200
  93
  94     # Reset simulator, all
  95     response=client.post(SERVER_URL+'deleteall')
  96     assert response.status_code == 200
  97
  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"
 102
 103     # Get counter: remote hosts
 104     response=client.get(SERVER_URL+'counter/remote_hosts')
 105     assert response.status_code == 200
 106
 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"
 111
 112     # Get counter: types
 113     response=client.get(SERVER_URL+'counter/num_types')
 114     assert response.status_code == 200
 115     assert response.data ==  b"0"
 116
 117     # API: Get policy type, shall be empty
 118     data_response = [ ]
 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)
 123     assert res == True
 124
 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)
 131     assert res == True
 132
 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)
 139     assert res == True
 140
 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
 148
 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
 156
 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)
 163     assert res == True
 164
 165     # Delete a policy type: STD_1
 166     data_response = b""
 167     response=client.delete(SERVER_URL+'policytype?id=STD_1')
 168     assert response.status_code == 204
 169     assert data_response == response.data
 170
 171     # API: Get policy type ids, shall be empty
 172     data_response = [  ]
 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)
 177     assert res == True
 178
 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
 186
 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)
 193     assert res == True
 194
 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"
 199
 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)
 207         assert res == True
 208
 209     # API: API: Get policy instances, shall be empty
 210     data_response = []
 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)
 215     assert res == True
 216
 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)
 224         assert res == True
 225
 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)
 233         assert res == True
 234
 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)
 242         assert res == True
 243
 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)
 251         assert res == True
 252
 253     # # API: Duplicate policy instance pi2 of type: STD_1 - and delete it
 254
 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)
 261         assert res == True
 262     data_response = b""
 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
 266
 267
 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)
 274     assert res == True
 275
 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"
 280
 281     # Get counter: types
 282     response=client.get(SERVER_URL+'counter/num_types')
 283     assert response.status_code == 200
 284     assert response.data ==  b"1"
 285
 286     # Set force response code 409. ==="
 287     response=client.post(SERVER_URL+'forceresponse?code=409')
 288     assert response.status_code == 200
 289
 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)
 296     assert res == True
 297
 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)
 304     assert res == True
 305
 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)
 313         assert res == True
 314
 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)
 322         assert res == True
 323
 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)
 330     assert res == True
 331
 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"
 336
 337     # Get counter: types
 338     response=client.get(SERVER_URL+'counter/num_types')
 339     assert response.status_code == 200
 340     assert response.data ==  b"1"
 341
 342     # Set force delay 10
 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"
 346
 347     #start time stamp
 348     start=time.time()
 349
 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)
 356     assert res == True
 357
 358     end=time.time()
 359
 360     assert (end-start) > 9
 361
 362     # Reset force delay
 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"
 366
 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)
 374         assert res == True
 375
 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)
 383         assert res == True
 384
 385     # API: DELETE policy instance pi1
 386     data_response = b""
 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
 390
 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)
 397     assert res == True
 398
 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)
 405     assert res == True
 406
 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
 410
 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)
 417     print(response.data)
 418     assert res == True
 419
 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
 423
 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)
 430     assert res == True
 431
 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"
 436
 437     # Send data
 438     json_payload={}
 439     response=client.post(SERVER_URL+'datadelivery', headers=header, data=json.dumps(json_payload))
 440     assert response.status_code == 200
 441
 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"
 446
 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"
 451
 452     # Get counter: remote hosts
 453     response=client.get(SERVER_URL+'counter/remote_hosts')
 454     assert response.status_code == 200
 455
 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"
 460
 461     # Get counter: types
 462     response=client.get(SERVER_URL+'counter/num_types')
 463     assert response.status_code == 200
 464     assert response.data ==  b"1"
 465
 466 def test_notificationDestination(client):
 467     test_data = get_testdata_dir() + 'pi2.json'
 468     # Header for json payload
 469     header = { "Content-Type" : "application/json" }
 470
 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
 478
 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=())
 482     proc.start()
 483
 484     test_data = get_testdata_dir() + 'pi2.json'
 485     header = { "Content-Type" : "application/json" }
 486
 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))
 491
 492     assert response.status_code == 204
 493     result = response.data
 494     assert result == b""
 495
 496     # Send status, negative test with missing parameter
 497     response = client.post(SERVER_URL+'sendstatus', headers=header, data="")
 498     assert response.status_code == 400
 499
 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
 503
 504     proc.terminate()