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=f5222c967ac38a4170eac711bcdbedcfdeb87960;hb=cabc0720dd873af3e08ab16da9ed5694a95de5e2;hp=5c9b7809dc896addd33cb66fe5c2f662e6552707;hpb=eae5626936f20d51af41c1b2ea4a7bcc10fa4086;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 5c9b780..f5222c9 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'
@@ -582,57 +581,44 @@ def test_apis(client):
     assert response.status_code == 200
     assert response.data ==  b"2"
 
-
 def test_notificationDestination(client):
-    testdata=get_testdata_dir()
+    test_data = get_testdata_dir() + 'pi2.json'
     # Header for json payload
-    header = {
-        "Content-Type" : "application/json"
-    }
+    header = { "Content-Type" : "application/json" }
 
     # === API: Update policy instance pi2 of type: 2 ==="
-    with open(testdata+'pi2.json') as json_file:
-        policytype_2 = 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(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()
+    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" }
+
+    # === 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()