X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=near-rt-ric-simulator%2Ftests%2Ftest_osc_2_1_0.py;h=b54e417ec5a75608db82b82606936e9304219e42;hb=8873ea581307f1544b6ece3cb31b78c781ae032e;hp=31e70fdb5f976bd76ae9673090bc1728d94da884;hpb=868107297d275a6f50790a5e6d3de6c65cbd4a3e;p=sim%2Fa1-interface.git diff --git a/near-rt-ric-simulator/tests/test_osc_2_1_0.py b/near-rt-ric-simulator/tests/test_osc_2_1_0.py index 31e70fd..b54e417 100644 --- a/near-rt-ric-simulator/tests/test_osc_2_1_0.py +++ b/near-rt-ric-simulator/tests/test_osc_2_1_0.py @@ -1,5 +1,6 @@ # ============LICENSE_START=============================================== -# Copyright (C) 2020 Nordix Foundation. All rights reserved. +# Copyright (C) 2021-2023 Nordix Foundation. All rights reserved. +# Copyright (C) 2023-2024 OpenInfra Foundation Europe. All Rights Reserved # ======================================================================== # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -15,19 +16,58 @@ # ============LICENSE_END================================================= # -# This test case test the OSC_2.1.0 version of the simulator +# This test case tests the OSC_2.1.0 version of the simulator -import json -#Version of simulator +# Version of simulator INTERFACE_VERSION="OSC_2.1.0" -from unittest_setup import SERVER_URL, setup_env, get_testdata_dir, client +import json +import pytest +import multiprocessing +from unittest_setup import SERVER_URL, PORT_NUMBER, setup_env, get_testdata_dir, client +from unittest_setup import run_flask_app -#Setup env and import paths +# 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): @@ -40,7 +80,7 @@ def test_apis(client): # Check used and implemented interfaces response=client.get(SERVER_URL+'container_interfaces') assert response.status_code == 200 - 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']" + 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']" # Reset simulator instances response=client.post(SERVER_URL+'deleteinstances') @@ -156,11 +196,15 @@ def test_apis(client): res=compare(data_policies_get, result) assert res == True - # API: Create policy instance pi2 (copy of pi1) of type: 1. Shall fail + # API: Create policy instance pi2 (copy of pi1) of type: 1. with open(testdata+'pi1.json') as json_file: policy_2 = json.load(json_file) response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi2', headers=header, data=json.dumps(policy_2)) - assert response.status_code == 400 + assert response.status_code == 202 + + # API: DELETE policy instance pi1 + response=client.delete(SERVER_URL+'a1-p/policytypes/1/policies/pi2') + assert response.status_code == 202 # Set force response code 401 response=client.post(SERVER_URL+'forceresponse?code=401') @@ -172,9 +216,8 @@ def test_apis(client): # API: Get policy status policy_status = { - "instance_status" : "NOT IN EFFECT", - "has_been_deleted" : "false", - "created_at" : "????" + "enforceStatus" : "NOT_ENFORCED", + "enforceReason" : "OTHER_REASON", } response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status') assert response.status_code == 200 @@ -266,7 +309,7 @@ def test_apis(client): # API: Create policy instance pi2 of type: 2. Missing param, shall fail with open(testdata+'pi2_missing_param.json') as json_file: policy_2 = json.load(json_file) - response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi1', headers=header, data=json.dumps(policy_2)) + response=client.put(SERVER_URL+'a1-p/policytypes/2/policies/pi2', headers=header, data=json.dumps(policy_2)) assert response.status_code == 400 # API: Create policy instance pi2 of type: 2 @@ -296,12 +339,6 @@ def test_apis(client): res=compare(data_policies_get, result) assert res == True - # API: Create policy instance pi11 (copy of pi1) of type: 1. Shall fail - with open(testdata+'pi1.json') as json_file: - policy_1 = json.load(json_file) - response=client.put(SERVER_URL+'a1-p/policytypes/1/policies/pi11', headers=header, data=json.dumps(policy_1)) - assert response.status_code == 400 - # Set force response code 409. ===" response=client.post(SERVER_URL+'forceresponse?code=401') assert response.status_code == 200 @@ -317,9 +354,8 @@ def test_apis(client): # API: Get policy status for pi1. Shall delay 10 sec policy_status = { - "instance_status" : "NOT IN EFFECT", - "has_been_deleted" : "false", - "created_at" : "????" + "enforceStatus" : "NOT_ENFORCED", + "enforceReason" : "OTHER_REASON", } response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status') assert response.status_code == 200 @@ -333,14 +369,13 @@ def test_apis(client): assert response.data == b"Force delay: None sec set for all A1 responses" # Set status for pi1 - response=client.put(SERVER_URL+'status?policyid=pi1&status=IN%20EFFECT') + response=client.put(SERVER_URL+'status?policyid=pi1&status=ENFORCED') assert response.status_code == 200 # API: Get policy status for pi1 policy_status = { - "instance_status" : "IN EFFECT", - "has_been_deleted" : "false", - "created_at" : "????" + "enforceStatus" : "ENFORCED", + "enforceReason" : None, } response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status') assert response.status_code == 200 @@ -349,14 +384,13 @@ def test_apis(client): assert res == True # Set status for pi1 - response=client.put(SERVER_URL+'status?policyid=pi1&status=IN%20EFFECT&deleted=true&created_at=2020-03-30%2012:00:00') + response=client.put(SERVER_URL+'status?policyid=pi1&status=NOT_ENFORCED&reason=SCOPE_NOT_APPLICABLE') assert response.status_code == 200 # API: Get policy status for pi1 policy_status = { - "instance_status" : "IN EFFECT", - "has_been_deleted" : "true", - "created_at" : "????" + "enforceStatus" : "NOT_ENFORCED", + "enforceReason" : "SCOPE_NOT_APPLICABLE", } response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi1/status') assert response.status_code == 200 @@ -364,7 +398,7 @@ def test_apis(client): res=compare(policy_status, result) assert res == True - # Get counter: intstance + # Get counter: num_instances response=client.get(SERVER_URL+'counter/num_instances') assert response.status_code == 200 assert response.data == b"2" @@ -497,14 +531,98 @@ def test_apis(client): response=client.get(SERVER_URL+'a1-p/policytypes/1/policies/pi111/status') assert response.status_code == 404 - # Load policy type, no type in url - shall faill + # Load policy type, no type in url - shall fail with open(testdata+'pt2.json') as json_file: policytype_2 = json.load(json_file) response=client.put(SERVER_URL+'policytype', headers=header, data=json.dumps(policytype_2)) assert response.status_code == 400 - # Load policy type - duplicatee - shall faill + # Load policy type - duplicatee - shall fail with open(testdata+'pt1.json') as json_file: policytype_1 = json.load(json_file) response=client.put(SERVER_URL+'policytype?id=2', headers=header, data=json.dumps(policytype_1)) assert response.status_code == 400 + + # Get counter: data_delivery + response=client.get(SERVER_URL+'counter/datadelivery') + assert response.status_code == 200 + assert response.data == b"0" + + # Send data to data-delivery with empty payload + json_payload={} + response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload)) + assert response.status_code == 400 + + # Send invalid data to data-delivery + json_payload={ + "job":"200", + "payload":"payload" + } + response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload)) + assert response.status_code == 404 + + # Send data to data-delivery with valid job + json_payload={ + "job":"100", + "payload":"payload" + } + response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload)) + assert response.status_code == 200 + + # Send data to data-delivery with valid job + json_payload={ + "job":"101", + "payload":"another payload" + } + response=client.post(SERVER_URL+'data-delivery', headers=header, data=json.dumps(json_payload)) + assert response.status_code == 200 + + # Get counter: data_delivery + response=client.get(SERVER_URL+'counter/datadelivery') + assert response.status_code == 200 + assert response.data == b"2" + +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/policytypes/2/policies/pi2?notificationDestination=http://localhost:8086/statustest", headers=header, data=json.dumps(payload)) + + assert response.status_code == 202 + result = response.data + assert result == b"" + + +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" } + + # Timeout can be removed with polling the endpoints if required + proc.join(timeout=10) + + # === 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 == 201 + result = response.data + assert result == b"OK" + + # 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()