X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=near-rt-ric-simulator%2Ftests%2Ftest_std_2_0_0.py;h=0da2467025c9484003b46efe023bcf70a3567735;hb=79992d7e456696d6211edcb64d6b2ce2d8b730c6;hp=2a01d836d80857421514c92d3cee23e1088d065f;hpb=24ee2701e54884c8fc2ed7e6cf522d71286a3b33;p=sim%2Fa1-interface.git diff --git a/near-rt-ric-simulator/tests/test_std_2_0_0.py b/near-rt-ric-simulator/tests/test_std_2_0_0.py index 2a01d83..0da2467 100644 --- a/near-rt-ric-simulator/tests/test_std_2_0_0.py +++ b/near-rt-ric-simulator/tests/test_std_2_0_0.py @@ -15,21 +15,60 @@ # ============LICENSE_END================================================= # -# This test case test the STD_2.0.0 version of the simulator +# This test case tests the STD_2.0.0 version of the simulator. import json +import pytest import time +import multiprocessing +from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client +from unittest_setup import run_flask_app -#Version of simulator -INTERFACE_VERSION="STD_2.0.0" +# Setup env and import paths -from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client +# Version of simulator +INTERFACE_VERSION="STD_2.0.0" -#Setup env and import paths setup_env(INTERFACE_VERSION) - from compare_json import compare +from models.enforceStatus import EnforceStatus + +def test_enforce_reason(client): + """ + Test that we can set a valid enforce status and reason, and that we reject invalid cases. + """ + enforceStatus = EnforceStatus() + + enforceStatus.enforce_status = 'NOT_ENFORCED' + enforceStatus.enforce_reason = 'SCOPE_NOT_APPLICABLE' + enforce_dict = enforceStatus.to_dict() + assert enforce_dict['enforceStatus'] == 'NOT_ENFORCED' + assert enforce_dict['enforceReason'] == 'SCOPE_NOT_APPLICABLE' + + enforceStatus.enforce_status = 'ENFORCED' + enforceStatus.enforce_reason = 'STATEMENT_NOT_APPLICABLE' + enforce_dict = enforceStatus.to_dict() + assert enforce_dict['enforceStatus'] == 'ENFORCED' + assert enforce_dict['enforceReason'] == 'STATEMENT_NOT_APPLICABLE' + + enforceStatus.enforce_reason = 'OTHER_REASON' + enforce_dict = enforceStatus.to_dict() + assert enforce_dict['enforceReason'] == 'OTHER_REASON' + + enforce_status = enforceStatus.enforce_status + assert str(enforce_status) == 'ENFORCED' + + enforce_reason = enforceStatus.enforce_reason + assert str(enforce_reason) == 'OTHER_REASON' + + with pytest.raises(ValueError): + enforceStatus.enforce_status = 'ERROR' + + with pytest.raises(ValueError): + enforceStatus.enforce_reason = 'ERROR' + + def test_apis(client): testdata=get_testdata_dir() @@ -257,7 +296,7 @@ def test_apis(client): assert res == True # API: API: Get policy status - data_response = {"enforceStatus" : "", "enforceReason" : ""} + data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"} response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1/status') assert response.status_code == 200 result=json.loads(response.data) @@ -358,7 +397,7 @@ def test_apis(client): assert res == True # API: API: Get policy status - data_response = {"enforceStatus" : "", "enforceReason" : ""} + data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"} response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status') assert response.status_code == 200 result=json.loads(response.data) @@ -366,11 +405,11 @@ def test_apis(client): assert res == True # Set status for policy instance pi2 - response=client.put(SERVER_URL+'status?policyid=pi2&status=OK') + response=client.put(SERVER_URL+'status?policyid=pi2&status=NOT_ENFORCED&reason=STATEMENT_NOT_APPLICABLE') assert response.status_code == 200 # API: API: Get policy status - data_response = {"enforceStatus" : "OK"} + data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "STATEMENT_NOT_APPLICABLE"} response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status') assert response.status_code == 200 result=json.loads(response.data) @@ -379,33 +418,17 @@ def test_apis(client): assert res == True # Set status for policy instance pi2 - response=client.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason') + response=client.put(SERVER_URL+'status?policyid=pi2&status=NOT_ENFORCED&reason=OTHER_REASON') assert response.status_code == 200 # API: API: Get policy status - data_response = {"enforceStatus" : "NOTOK", "enforceReason" : "notok_reason"} + data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"} response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status') assert response.status_code == 200 result=json.loads(response.data) res=compare(data_response, result) assert res == True - - # #Found no way to test these functions - # #'sendstatus' will send a http request that will fail - # #since no server will receive the call - # #These function is instead tested when running the bash script in the 'test' dir - # # # Send status for pi2 - # # response=client.post(SERVER_URL+'sendstatus?policyid=pi2') - # # assert response.status_code == 200 - # # result=json.loads(response.data) - # # res=compare(data_get_status, result) - # # assert res == True - - # # # Send status, shall fail - # # response=client.post(SERVER_URL+'sendstatus') - # # assert response.status_code == 400 - # Get counter: data_delivery response=client.get(SERVER_URL+'counter/datadelivery') assert response.status_code == 200 @@ -438,4 +461,44 @@ def test_apis(client): # Get counter: types response=client.get(SERVER_URL+'counter/num_types') assert response.status_code == 200 - assert response.data == b"1" \ No newline at end of file + assert response.data == b"1" + +def test_notificationDestination(client): + test_data = get_testdata_dir() + 'pi2.json' + # Header for json payload + header = { "Content-Type" : "application/json" } + + # === API: Update policy instance pi2 of type: 2 ===" + with open(test_data) as json_file: + payload = json.load(json_file) + response = client.put(SERVER_URL+"A1-P/v2/policytypes/STD_1/policies/pi2?notificationDestination=http://localhost:8086/statustest", headers=header, data=json.dumps(payload)) + assert response.status_code == 200 + result = json.loads(response.data) + assert compare(payload, result) == True + +def test_sendstatus(client): + # Create a new thread to run the Flask app in parallel on a different port so that we can call the callback. + proc = multiprocessing.Process(target=run_flask_app, args=()) + proc.start() + + test_data = get_testdata_dir() + 'pi2.json' + header = { "Content-Type" : "application/json" } + + # === Send status for pi2=== + with open(test_data) as json_file: + payload = json.load(json_file) + response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(payload)) + + assert response.status_code == 204 + result = response.data + assert result == b"" + + # Send status, negative test with missing parameter + response = client.post(SERVER_URL+'sendstatus', headers=header, data="") + assert response.status_code == 400 + + # Send status pi9, negative test for policy id not found + response = client.post(SERVER_URL+'sendstatus?policyid=pi9', headers=header, data="") + assert response.status_code == 404 + + proc.terminate()