From 5fa91815945b1b4469c0f345ebb56c9aba5c72c1 Mon Sep 17 00:00:00 2001 From: DenisGNoonan Date: Tue, 29 Aug 2023 17:33:09 +0100 Subject: [PATCH] NONRTRIC-892: A1-Simulator - add unit test for sendstatus Change-Id: Iafd13306c9a7156a42f8d5f553188cc1bc57c28b Signed-off-by: DenisGNoonan --- near-rt-ric-simulator/tests/test_osc_2_1_0.py | 90 ++++++++++++++------------- near-rt-ric-simulator/tests/unittest_setup.py | 8 +-- 2 files changed, 50 insertions(+), 48 deletions(-) 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 5c9b780..169f360 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 @@ -23,10 +23,9 @@ INTERFACE_VERSION="OSC_2.1.0" import json import pytest -import requests -import threading +import multiprocessing from unittest_setup import SERVER_URL, PORT_NUMBER, setup_env, get_testdata_dir, client -# from unittest_setup import run_flask_app +from unittest_setup import run_flask_app # Setup env and import paths setup_env(INTERFACE_VERSION) @@ -51,7 +50,7 @@ def test_enforce_reason(client): 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' @@ -596,43 +595,46 @@ def test_notificationDestination(client): response = client.put(SERVER_URL+"a1-p/policytypes/2/policies/pi2?notificationDestination=http://localhost:8085/statustest", headers=header, data=json.dumps(policytype_2)) assert response.status_code == 202 result = response.data - assert result == b"" - -# def test_sendstatus(client): -# testdata=get_testdata_dir() -# # Header for json payload -# header = { -# "Content-Type" : "application/json" -# } - -# # === Send status for pi2===" -# with open(testdata+'pi2.json') as json_file: -# policytype_2 = json.load(json_file) -# response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(policytype_2)) -# assert response.status_code == 201 -# result = response.data -# assert result == b"OK" - - -# def test_multithreaded(client): -# # Create a new thread to run the Flask app -# app_thread = threading.Thread(target=run_flask_app) -# app_thread.start() - -# # Perform your tests here -# testdata=get_testdata_dir() -# # Header for json payload -# header = { -# "Content-Type" : "application/json" -# } - -# # === Send status for pi2===" -# with open(testdata+'pi2.json') as json_file: -# policytype_2 = json.load(json_file) -# response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(policytype_2)) -# assert response.status_code == 201 -# result = response.data -# assert result == b"OK" - -# # Wait for the Flask app thread to finish -# app_thread.join() + assert result == b"" + +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:8085/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" } + + # === 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() diff --git a/near-rt-ric-simulator/tests/unittest_setup.py b/near-rt-ric-simulator/tests/unittest_setup.py index 06567cf..f2c2955 100644 --- a/near-rt-ric-simulator/tests/unittest_setup.py +++ b/near-rt-ric-simulator/tests/unittest_setup.py @@ -58,7 +58,7 @@ def client(): with app.app.test_client() as client: yield client -# # Run the Flask app in a separate thread for testing -# def run_flask_app(): -# from main import app -# app.app.run(port=8085, host="127.0.0.1", threaded=True) +# Run the Flask app in a separate thread for testing +def run_flask_app(): + from main import app + app.app.run(port=8085, host="127.0.0.1") -- 2.16.6