tests/test_qpd.py

   1 # ==================================================================================
   2 #       Copyright (c) 2020 AT&T Intellectual Property.
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #          http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 # ==================================================================================
  16 import json
  17 import time
  18 from contextlib import suppress
  19 from qpdriver import main, data
  20 from ricxappframe.xapp_frame import Xapp, RMRXapp
  21
  22 mock_ts_xapp = None
  23 mock_qp_xapp = None
  24
  25 """
  26  these tests are not currently parallelizable (do not use this tox flag)
  27  I would use setup_module, however that can't take monkeypatch fixtures
  28  Currently looking for the best way to make this better:
  29  https://stackoverflow.com/questions/60886013/python-monkeypatch-in-pytest-setup-module
  30 """
  31
  32
  33 def test_init_xapp(monkeypatch, ue_metrics, cell_metrics_1, cell_metrics_2, cell_metrics_3, ue_metrics_with_bad_cell):
  34     # monkeypatch post_init to set the data we want in SDL
  35     # the metrics arguments are JSON (dict) objects
  36
  37     _original_post_init = main.post_init
  38
  39     def fake_post_init(self):
  40         _original_post_init(self)
  41         self.sdl_set(data.UE_NS, "12345", json.dumps(ue_metrics).encode(), usemsgpack=False)
  42         self.sdl_set(data.UE_NS, "8675309", json.dumps(ue_metrics_with_bad_cell).encode(), usemsgpack=False)
  43         self.sdl_set(data.CELL_NS, "310-680-200-555001", json.dumps(cell_metrics_1).encode(), usemsgpack=False)
  44         self.sdl_set(data.CELL_NS, "310-680-200-555002", json.dumps(cell_metrics_2).encode(), usemsgpack=False)
  45         self.sdl_set(data.CELL_NS, "310-680-200-555003", json.dumps(cell_metrics_3).encode(), usemsgpack=False)
  46
  47     # patch
  48     monkeypatch.setattr("qpdriver.main.post_init", fake_post_init)
  49
  50     # start qpd
  51     main.start(thread=True)
  52
  53
  54 def test_rmr_flow(monkeypatch, qpd_to_qp, qpd_to_qp_bad_cell):
  55     """
  56     this flow mocks out the xapps on both sides of QP driver.
  57     It first stands up a mock qp, then it starts up a mock ts
  58     which will immediately send requests to the running qp driver.
  59     """
  60
  61     expected_result = {}
  62
  63     # define a mock qp predictor
  64     def mock_qp_default_handler(self, summary, sbuf):
  65         pass
  66
  67     def mock_qp_predict_handler(self, summary, sbuf):
  68         nonlocal expected_result  # closures ftw
  69         pay = json.loads(summary["payload"])
  70         expected_result[pay["PredictionUE"]] = pay
  71
  72     global mock_qp_xapp
  73     mock_qp_xapp = RMRXapp(mock_qp_default_handler, rmr_port=4666, use_fake_sdl=True)
  74     mock_qp_xapp.register_callback(mock_qp_predict_handler, 30001)
  75     mock_qp_xapp.run(thread=True)
  76
  77     time.sleep(1)
  78
  79     # define a mock traffic steering xapp
  80     def mock_ts_entry(self):
  81
  82         # make sure a bad steering request doesn't blow up in qpd
  83         val = "notevenjson".encode()
  84         self.rmr_send(val, 30000)
  85         val = json.dumps({"bad": "tothebone"}).encode()  # json but missing UEPredictionSet
  86         self.rmr_send(val, 30000)
  87
  88         # valid request body but missing cell id
  89         val = json.dumps({"UEPredictionSet": ["VOIDOFLIGHT"]}).encode()
  90         self.rmr_send(val, 30000)
  91
  92         # good traffic steering request
  93         val = json.dumps({"UEPredictionSet": ["12345", "8675309"]}).encode()
  94         self.rmr_send(val, 30000)
  95
  96         # should trigger the default handler and do nothing
  97         val = json.dumps({"test send 60001": 2}).encode()
  98         self.rmr_send(val, 60001)
  99
 100     global mock_ts_xapp
 101     mock_ts_xapp = Xapp(entrypoint=mock_ts_entry, rmr_port=4564, use_fake_sdl=True)
 102     mock_ts_xapp.run()  # this will return since entry isn't a loop
 103
 104     time.sleep(1)
 105
 106     assert expected_result == {"12345": qpd_to_qp, "8675309": qpd_to_qp_bad_cell}
 107     assert main.get_stats() == {"DefCalled": 1, "SteeringRequests": 4}
 108
 109     # break SDL and send traffic again
 110     def sdl_healthcheck_fails(self):
 111         return False
 112     monkeypatch.setattr("ricxappframe.xapp_sdl.SDLWrapper.healthcheck", sdl_healthcheck_fails)
 113     mock_ts_xapp.run()
 114
 115     # restore SDL and send traffic once more
 116     def sdl_healthcheck_passes(self):
 117         return True
 118     monkeypatch.setattr("ricxappframe.xapp_sdl.SDLWrapper.healthcheck", sdl_healthcheck_passes)
 119     mock_ts_xapp.run()
 120
 121
 122 def teardown_module():
 123     """
 124     this is like a "finally"; the name of this function is pytest magic
 125     safer to put down here since certain failures above can lead to pytest never returning
 126     for example if an exception gets raised before stop is called in any test function above,
 127     pytest will hang forever
 128     """
 129     with suppress(Exception):
 130         mock_ts_xapp.stop()
 131     with suppress(Exception):
 132         mock_qp_xapp.stop()
 133     with suppress(Exception):
 134         main.stop()