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 test the STD_2.0.0 version of the simulator
  19
  20 import json
  21 import time
  22
  23 #Version of simulator
  24 INTERFACE_VERSION="STD_2.0.0"
  25
  26 from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client
  27
  28 #Setup env and import paths
  29 setup_env(INTERFACE_VERSION)
  30
  31 from compare_json import compare
  32
  33 def test_apis(client):
  34
  35     testdata=get_testdata_dir()
  36
  37     # Header for json payload
  38     header = {
  39         "Content-Type" : "application/json"
  40     }
  41
  42     # Simulator hello world
  43     response=client.get(SERVER_URL)
  44     assert response.status_code == 200
  45
  46     # Check used and implemented interfaces
  47     response=client.get(SERVER_URL+'container_interfaces')
  48     assert response.status_code == 200
  49     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']"
  50
  51     # Reset simulator instances
  52     response=client.post(SERVER_URL+'deleteinstances')
  53     assert response.status_code == 200
  54
  55     # Reset simulator, all
  56     response=client.post(SERVER_URL+'deleteall')
  57     assert response.status_code == 200
  58
  59     # Get counter: interface
  60     response=client.get(SERVER_URL+'counter/interface')
  61     assert response.status_code == 200
  62     assert response.data ==  b"STD_2.0.0"
  63
  64     # Get counter: remote hosts
  65     response=client.get(SERVER_URL+'counter/remote_hosts')
  66     assert response.status_code == 200
  67
  68     # Get counter: intstance
  69     response=client.get(SERVER_URL+'counter/num_instances')
  70     assert response.status_code == 200
  71     assert response.data ==  b"0"
  72
  73     # Get counter: types
  74     response=client.get(SERVER_URL+'counter/num_types')
  75     assert response.status_code == 200
  76     assert response.data ==  b"0"
  77
  78     # API: Get policy type, shall be empty
  79     data_response = [ ]
  80     response=client.get(SERVER_URL+'A1-P/v2/policytypes')
  81     assert response.status_code == 200
  82     result=json.loads(response.data)
  83     res=compare(data_response, result)
  84     assert res == True
  85
  86     # API: Get policy instances for type 1, type not found
  87     data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "1"}
  88     response=client.get(SERVER_URL+'A1-P/v2/policytypes/1/policies')
  89     assert response.status_code == 404
  90     result=json.loads(response.data)
  91     res=compare(data_response, result)
  92     assert res == True
  93
  94     # API: Get policy instances, type not found
  95     data_response = {"title": "The policy type does not exist.", "status": 404, "instance": "test"}
  96     response=client.get(SERVER_URL+'A1-P/v2/policytypes/test/policies')
  97     assert response.status_code == 404
  98     result=json.loads(response.data)
  99     res=compare(data_response, result)
 100     assert res == True
 101
 102     # Put a policy type: STD_1
 103     with open(testdata+'std_1.json') as json_file:
 104         data_response = b"Policy type STD_1 is OK."
 105         json_payload=json.load(json_file)
 106         response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
 107         assert response.status_code == 201
 108         assert data_response == response.data
 109
 110     # Put a policy type: STD_1, again
 111     with open(testdata+'std_1.json') as json_file:
 112         data_response = b"Policy type STD_1 is OK."
 113         json_payload=json.load(json_file)
 114         response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
 115         assert response.status_code == 200
 116         assert data_response == response.data
 117
 118     # API: Get policy type ids, shall contain type STD_1
 119     data_response = [ "STD_1" ]
 120     response=client.get(SERVER_URL+'A1-P/v2/policytypes')
 121     assert response.status_code == 200
 122     result=json.loads(response.data)
 123     res=compare(data_response, result)
 124     assert res == True
 125
 126     # Delete a policy type: STD_1
 127     data_response = b""
 128     response=client.delete(SERVER_URL+'policytype?id=STD_1')
 129     assert response.status_code == 204
 130     assert data_response == response.data
 131
 132     # API: Get policy type ids, shall be empty
 133     data_response = [  ]
 134     response=client.get(SERVER_URL+'A1-P/v2/policytypes')
 135     assert response.status_code == 200
 136     result=json.loads(response.data)
 137     res=compare(data_response, result)
 138     assert res == True
 139
 140     # Put a policy type: STD_1
 141     with open(testdata+'std_1.json') as json_file:
 142         data_response = b"Policy type STD_1 is OK."
 143         json_payload=json.load(json_file)
 144         response=client.put(SERVER_URL+'policytype?id=STD_1', headers=header, data=json.dumps(json_payload))
 145         assert response.status_code == 201
 146         assert data_response == response.data
 147
 148     # API: Get policy type ids, shall contain type STD_1
 149     data_response = [ "STD_1" ]
 150     response=client.get(SERVER_URL+'A1-P/v2/policytypes')
 151     assert response.status_code == 200
 152     result=json.loads(response.data)
 153     res=compare(data_response, result)
 154     assert res == True
 155
 156     # Get counter: types (shall be 1)
 157     response=client.get(SERVER_URL+'counter/num_types')
 158     assert response.status_code == 200
 159     assert response.data ==  b"1"
 160
 161     # API: Get policy type: STD_1
 162     with open(testdata+'std_1.json') as json_file:
 163         data_response = json.load(json_file)
 164         response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1')
 165         assert response.status_code == 200
 166         result=json.loads(response.data)
 167         res=compare(data_response, result)
 168         assert res == True
 169
 170     # API: API: Get policy instances, shall be empty
 171     data_response = []
 172     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
 173     assert response.status_code == 200
 174     result=json.loads(response.data)
 175     res=compare(data_response, result)
 176     assert res == True
 177
 178     # API: Create policy instance pi1 of type: STD_1
 179     with open(testdata+'pi1.json') as json_file:
 180         json_payload=json.load(json_file)
 181         response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
 182         assert response.status_code == 201
 183         result=json.loads(response.data)
 184         res=compare(json_payload, result)
 185         assert res == True
 186
 187     # API: API: Get policy instance pi1 of type: STD_1
 188     with open(testdata+'pi1.json') as json_file:
 189         data_response = json.load(json_file)
 190         response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
 191         assert response.status_code == 200
 192         result=json.loads(response.data)
 193         res=compare(data_response, result)
 194         assert res == True
 195
 196     # API: Update policy instance pi1 of type: STD_1
 197     with open(testdata+'pi1.json') as json_file:
 198         json_payload=json.load(json_file)
 199         response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
 200         assert response.status_code == 200
 201         result=json.loads(response.data)
 202         res=compare(json_payload, result)
 203         assert res == True
 204
 205     # API: Update policy instance pi1 of type: STD_1
 206     with open(testdata+'pi1_updated.json') as json_file:
 207         json_payload=json.load(json_file)
 208         response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1', headers=header, data=json.dumps(json_payload))
 209         assert response.status_code == 200
 210         result=json.loads(response.data)
 211         res=compare(json_payload, result)
 212         assert res == True
 213
 214     # # API: Duplicate policy instance pi2 of type: STD_1 - and delete it
 215
 216     with open(testdata+'pi1_updated.json') as json_file:
 217         json_payload=json.load(json_file)
 218         response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
 219         assert response.status_code == 201
 220         result=json.loads(response.data)
 221         res=compare(json_payload, result)
 222         assert res == True
 223     data_response = b""
 224     response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
 225     assert response.status_code == 204
 226     assert data_response == response.data
 227
 228
 229     # API: Get policy instances, shall contain pi1
 230     data_response = ["pi1"]
 231     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
 232     assert response.status_code == 200
 233     result=json.loads(response.data)
 234     res=compare(data_response, result)
 235     assert res == True
 236
 237     # Get counter: intstance
 238     response=client.get(SERVER_URL+'counter/num_instances')
 239     assert response.status_code == 200
 240     assert response.data ==  b"1"
 241
 242     # Get counter: types
 243     response=client.get(SERVER_URL+'counter/num_types')
 244     assert response.status_code == 200
 245     assert response.data ==  b"1"
 246
 247     # Set force response code 409. ==="
 248     response=client.post(SERVER_URL+'forceresponse?code=409')
 249     assert response.status_code == 200
 250
 251     # API: Get policy instances, shall fail
 252     data_response = {"title" : "Conflict", "status" : 409, "detail" : "Request could not be processed in the current state of the resource"}
 253     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
 254     assert response.status_code == 409
 255     result=json.loads(response.data)
 256     res=compare(data_response, result)
 257     assert res == True
 258
 259     # API: API: Get policy status
 260     data_response = {"enforceStatus" : "", "enforceReason" : ""}
 261     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1/status')
 262     assert response.status_code == 200
 263     result=json.loads(response.data)
 264     res=compare(data_response, result)
 265     assert res == True
 266
 267     # API: Create policy instance pi2 of type: STD_1
 268     with open(testdata+'pi2.json') as json_file:
 269         json_payload=json.load(json_file)
 270         response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
 271         assert response.status_code == 201
 272         result=json.loads(response.data)
 273         res=compare(json_payload, result)
 274         assert res == True
 275
 276     # API: Update policy instance pi2 of type: STD_1
 277     with open(testdata+'pi2.json') as json_file:
 278         json_payload=json.load(json_file)
 279         response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
 280         assert response.status_code == 200
 281         result=json.loads(response.data)
 282         res=compare(json_payload, result)
 283         assert res == True
 284
 285     # API: Get policy instances, shall contain pi1 and pi2
 286     data_response = ["pi1","pi2"]
 287     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
 288     assert response.status_code == 200
 289     result=json.loads(response.data)
 290     res=compare(data_response, result)
 291     assert res == True
 292
 293     # Get counter: intstance
 294     response=client.get(SERVER_URL+'counter/num_instances')
 295     assert response.status_code == 200
 296     assert response.data ==  b"2"
 297
 298     # Get counter: types
 299     response=client.get(SERVER_URL+'counter/num_types')
 300     assert response.status_code == 200
 301     assert response.data ==  b"1"
 302
 303     # Set force delay 10
 304     response=client.post(SERVER_URL+'forcedelay?delay=10')
 305     assert response.status_code == 200
 306     assert response.data ==  b"Force delay: 10 sec set for all A1 responses"
 307
 308     #start time stamp
 309     start=time.time()
 310
 311     # API: Get policy instances, shall contain pi1 and pi2 and delayed 10 sec
 312     data_response = ["pi1","pi2"]
 313     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
 314     assert response.status_code == 200
 315     result=json.loads(response.data)
 316     res=compare(data_response, result)
 317     assert res == True
 318
 319     end=time.time()
 320
 321     assert (end-start) > 9
 322
 323     # Reset force delay
 324     response=client.post(SERVER_URL+'forcedelay')
 325     assert response.status_code == 200
 326     assert response.data ==  b"Force delay: None sec set for all A1 responses"
 327
 328     # API: API: Get policy instance pi1 of type: STD_1
 329     with open(testdata+'pi1_updated.json') as json_file:
 330         data_response = json.load(json_file)
 331         response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
 332         assert response.status_code == 200
 333         result=json.loads(response.data)
 334         res=compare(data_response, result)
 335         assert res == True
 336
 337     # API: API: Get policy instance pi2 of type: STD_1
 338     with open(testdata+'pi2.json') as json_file:
 339         data_response = json.load(json_file)
 340         response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
 341         assert response.status_code == 200
 342         result=json.loads(response.data)
 343         res=compare(data_response, result)
 344         assert res == True
 345
 346     # API: DELETE policy instance pi1
 347     data_response = b""
 348     response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1')
 349     assert response.status_code == 204
 350     assert data_response == response.data
 351
 352     # API: Get policy instances, shall contain pi2
 353     data_response = ["pi2"]
 354     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies')
 355     assert response.status_code == 200
 356     result=json.loads(response.data)
 357     res=compare(data_response, result)
 358     assert res == True
 359
 360     # API: API: Get policy status
 361     data_response = {"enforceStatus" : "", "enforceReason" : ""}
 362     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
 363     assert response.status_code == 200
 364     result=json.loads(response.data)
 365     res=compare(data_response, result)
 366     assert res == True
 367
 368     # Set status for policy instance pi2
 369     response=client.put(SERVER_URL+'status?policyid=pi2&status=OK')
 370     assert response.status_code == 200
 371
 372     # API: API: Get policy status
 373     data_response = {"enforceStatus" : "OK"}
 374     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
 375     assert response.status_code == 200
 376     result=json.loads(response.data)
 377     res=compare(data_response, result)
 378     print(response.data)
 379     assert res == True
 380
 381     # Set status for policy instance pi2
 382     response=client.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason')
 383     assert response.status_code == 200
 384
 385     # API: API: Get policy status
 386     data_response = {"enforceStatus" : "NOTOK", "enforceReason" : "notok_reason"}
 387     response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
 388     assert response.status_code == 200
 389     result=json.loads(response.data)
 390     res=compare(data_response, result)
 391     assert res == True
 392
 393
 394     # #Found no way to test these functions
 395     # #'sendstatus' will send a http request that will fail
 396     # #since no server will receive the call
 397     # #These function is instead tested when running the bash script in the 'test' dir
 398     # # # Send status for pi2
 399     # # response=client.post(SERVER_URL+'sendstatus?policyid=pi2')
 400     # # assert response.status_code == 200
 401     # # result=json.loads(response.data)
 402     # # res=compare(data_get_status, result)
 403     # # assert res == True
 404
 405     # # # Send status, shall fail
 406     # # response=client.post(SERVER_URL+'sendstatus')
 407     # # assert response.status_code == 400
 408
 409     # Get counter: data_delivery
 410     response=client.get(SERVER_URL+'counter/datadelivery')
 411     assert response.status_code == 200
 412     assert response.data ==  b"0"
 413
 414     # Send data
 415     json_payload={}
 416     response=client.post(SERVER_URL+'datadelivery', headers=header, data=json.dumps(json_payload))
 417     assert response.status_code == 200
 418
 419     # Get counter: data_delivery
 420     response=client.get(SERVER_URL+'counter/datadelivery')
 421     assert response.status_code == 200
 422     assert response.data ==  b"1"
 423
 424     # Get counter: interface
 425     response=client.get(SERVER_URL+'counter/interface')
 426     assert response.status_code == 200
 427     assert response.data ==  b"STD_2.0.0"
 428
 429     # Get counter: remote hosts
 430     response=client.get(SERVER_URL+'counter/remote_hosts')
 431     assert response.status_code == 200
 432
 433     # Get counter: intstance
 434     response=client.get(SERVER_URL+'counter/num_instances')
 435     assert response.status_code == 200
 436     assert response.data ==  b"1"
 437
 438     # Get counter: types
 439     response=client.get(SERVER_URL+'counter/num_types')
 440     assert response.status_code == 200
 441     assert response.data ==  b"1"