Raise alarm if SDL connection is not healthy

[ric-app/qp-driver.git] / tests / test_qpd.py

diff --git a/tests/test_qpd.py b/tests/test_qpd.py
index ac72c2c..ef63727 100644 (file)
--- a/tests/test_qpd.py
+++ b/tests/test_qpd.py
@@ -33,9 +33,11 @@ mock_qp_xapp = None
 def test_init_xapp(monkeypatch, ue_metrics, cell_metrics_1, cell_metrics_2, cell_metrics_3, ue_metrics_with_bad_cell):
     # monkeypatch post_init to set the data we want in SDL
     # the metrics arguments are JSON (dict) objects
+
+    _original_post_init = main.post_init
+
     def fake_post_init(self):
-        self.def_hand_called = 0
-        self.traffic_steering_requests = 0
+        _original_post_init(self)
         self.sdl_set(data.UE_NS, "12345", json.dumps(ue_metrics).encode(), usemsgpack=False)
         self.sdl_set(data.UE_NS, "8675309", json.dumps(ue_metrics_with_bad_cell).encode(), usemsgpack=False)
         self.sdl_set(data.CELL_NS, "310-680-200-555001", json.dumps(cell_metrics_1).encode(), usemsgpack=False)
@@ -104,6 +106,18 @@ def test_rmr_flow(monkeypatch, qpd_to_qp, qpd_to_qp_bad_cell):
     assert expected_result == {"12345": qpd_to_qp, "8675309": qpd_to_qp_bad_cell}
     assert main.get_stats() == {"DefCalled": 1, "SteeringRequests": 4}
 
+    # break SDL and send traffic again
+    def sdl_healthcheck_fails(self):
+        return False
+    monkeypatch.setattr("ricxappframe.xapp_sdl.SDLWrapper.healthcheck", sdl_healthcheck_fails)
+    mock_ts_xapp.run()
+
+    # restore SDL and send traffic once more
+    def sdl_healthcheck_passes(self):
+        return True
+    monkeypatch.setattr("ricxappframe.xapp_sdl.SDLWrapper.healthcheck", sdl_healthcheck_passes)
+    mock_ts_xapp.run()
+
 
 def teardown_module():
     """