tests/test_qp.py

   1 # ==================================================================================
   2 #       Copyright (c) 2020 HCL Technologies Limited.
   3 #       Copyright (c) 2020 AT&T Intellectual Property.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 # ==================================================================================
  17 import json
  18 import time
  19 from contextlib import suppress
  20 from src import main
  21 from ricxappframe.xapp_frame import Xapp, RMRXapp
  22
  23 mock_qp_xapp = None
  24 mock_ts_xapp = None
  25
  26 """
  27  these tests are not currently parallelizable (do not use this tox flag)
  28  I would use setup_module, however that can't take monkeypatch fixtures
  29  Currently looking for the best way to make this better:
  30  https://stackoverflow.com/questions/60886013/python-monkeypatch-in-pytest-setup-module
  31 """
  32
  33
  34 def test_init_xapp(monkeypatch):
  35
  36     # monkeypatch post_init to set the data we want
  37     def fake_post_init(self):
  38         self.predict_requests = 0
  39
  40     # patch
  41     monkeypatch.setattr("src.main.post_init", fake_post_init)
  42
  43     # start qp
  44     main.start(thread=True)
  45
  46
  47 def test_database_connection(monkeypatch):
  48     main.connectdb(thread=True)
  49
  50
  51 def test_training(monkeypatch, qp_train):
  52     main.train_model(qp_train)
  53
  54
  55 def test_predict(monkeypatch, ts_to_qp):
  56     main.predict(ts_to_qp)
  57
  58
  59 def test_rmr_flow(monkeypatch, ts_to_qp, qp_prediction):
  60     """
  61     this flow mocks out the xapps on both sides of QP.
  62     It first stands up a mock ts, then it starts up a mock qp-driver
  63     which will immediately send requests to the running qp.
  64     """
  65
  66     expected_result = {}
  67
  68     # define a mock traffic steering xapp that listens on 4563
  69     def mock_ts_default_handler(self, summary, sbuf):
  70         pass
  71
  72     def mock_ts_prediction_handler(self, summary, sbuf):
  73         nonlocal expected_result  # closures ftw
  74         pay = json.loads(summary["payload"])
  75         expected_result = pay
  76
  77     global mock_ts_xapp
  78     mock_ts_xapp = RMRXapp(mock_ts_default_handler, rmr_port=4563, use_fake_sdl=True)
  79     mock_ts_xapp.register_callback(mock_ts_prediction_handler, 30002)
  80     mock_ts_xapp.run(thread=True)
  81
  82     time.sleep(1)
  83
  84     # define a mock qp xapp that listens on 4560
  85     def mock_qp_entry(self):
  86
  87         # good traffic steering request
  88         val = json.dumps(ts_to_qp).encode()
  89         self.rmr_send(val, 30000)
  90
  91         # should trigger the default handler and do nothing
  92         val = json.dumps({"test send 60001": 2}).encode()
  93         self.rmr_send(val, 60001)
  94
  95     global mock_qp_xapp
  96     mock_qp_xapp = Xapp(entrypoint=mock_qp_entry, rmr_port=4560, use_fake_sdl=True)
  97     mock_qp_xapp.run()  # this will return since entry isn't a loop
  98
  99
 100 def teardown_module():
 101     """
 102     this is like a "finally"; the name of this function is pytest magic
 103     safer to put down here since certain failures above can lead to pytest never returning
 104     for example if an exception gets raised before stop is called in any test function above,
 105     pytest will hang forever
 106     """
 107     with suppress(Exception):
 108         mock_ts_xapp.stop()
 109     with suppress(Exception):
 110         mock_qp_xapp.stop()
 111     with suppress(Exception):
 112         main.stop()