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     def fake_post_init(self):
  37         self.def_hand_called = 0
  38         self.traffic_steering_requests = 0
  39         self.sdl_set(data.UE_NS, "12345", json.dumps(ue_metrics).encode(), usemsgpack=False)
  40         self.sdl_set(data.UE_NS, "8675309", json.dumps(ue_metrics_with_bad_cell).encode(), usemsgpack=False)
  41         self.sdl_set(data.CELL_NS, "310-680-200-555001", json.dumps(cell_metrics_1).encode(), usemsgpack=False)
  42         self.sdl_set(data.CELL_NS, "310-680-200-555002", json.dumps(cell_metrics_2).encode(), usemsgpack=False)
  43         self.sdl_set(data.CELL_NS, "310-680-200-555003", json.dumps(cell_metrics_3).encode(), usemsgpack=False)
  44
  45     # patch
  46     monkeypatch.setattr("qpdriver.main.post_init", fake_post_init)
  47
  48     # start qpd
  49     main.start(thread=True)
  50
  51
  52 def test_rmr_flow(monkeypatch, qpd_to_qp, qpd_to_qp_bad_cell):
  53     """
  54     this flow mocks out the xapps on both sides of QP driver.
  55     It first stands up a mock qp, then it starts up a mock ts
  56     which will immediately send requests to the running qp driver.
  57     """
  58
  59     expected_result = {}
  60
  61     # define a mock qp predictor
  62     def mock_qp_default_handler(self, summary, sbuf):
  63         pass
  64
  65     def mock_qp_predict_handler(self, summary, sbuf):
  66         nonlocal expected_result  # closures ftw
  67         pay = json.loads(summary["payload"])
  68         expected_result[pay["PredictionUE"]] = pay
  69
  70     global mock_qp_xapp
  71     mock_qp_xapp = RMRXapp(mock_qp_default_handler, rmr_port=4666, use_fake_sdl=True)
  72     mock_qp_xapp.register_callback(mock_qp_predict_handler, 30001)
  73     mock_qp_xapp.run(thread=True)
  74
  75     time.sleep(1)
  76
  77     # define a mock traffic steering xapp
  78     def mock_ts_entry(self):
  79
  80         # make sure a bad steering request doesn't blow up in qpd
  81         val = "notevenjson".encode()
  82         self.rmr_send(val, 30000)
  83         val = json.dumps({"bad": "tothebone"}).encode()  # json but missing UEPredictionSet
  84         self.rmr_send(val, 30000)
  85
  86         # valid request body but missing cell id
  87         val = json.dumps({"UEPredictionSet": ["VOIDOFLIGHT"]}).encode()
  88         self.rmr_send(val, 30000)
  89
  90         # good traffic steering request
  91         val = json.dumps({"UEPredictionSet": ["12345", "8675309"]}).encode()
  92         self.rmr_send(val, 30000)
  93
  94         # should trigger the default handler and do nothing
  95         val = json.dumps({"test send 60001": 2}).encode()
  96         self.rmr_send(val, 60001)
  97
  98     global mock_ts_xapp
  99     mock_ts_xapp = Xapp(entrypoint=mock_ts_entry, rmr_port=4564, use_fake_sdl=True)
 100     mock_ts_xapp.run()  # this will return since entry isn't a loop
 101
 102     time.sleep(1)
 103
 104     assert expected_result == {"12345": qpd_to_qp, "8675309": qpd_to_qp_bad_cell}
 105     assert main.get_stats() == {"DefCalled": 1, "SteeringRequests": 4}
 106
 107
 108 def teardown_module():
 109     """
 110     this is like a "finally"; the name of this function is pytest magic
 111     safer to put down here since certain failures above can lead to pytest never returning
 112     for example if an exception gets raised before stop is called in any test function above,
 113     pytest will hang forever
 114     """
 115     with suppress(Exception):
 116         mock_ts_xapp.stop()
 117     with suppress(Exception):
 118         mock_qp_xapp.stop()
 119     with suppress(Exception):
 120         main.stop()