1 # ==================================================================================
2 # Copyright (c) 2020 AT&T Intellectual Property.
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
18 from contextlib import suppress
19 from qpdriver import main, data
20 from ricxappframe.xapp_frame import Xapp, RMRXapp
24 # tox.ini sets env var to this value
25 config_file_path = "/tmp/config.json"
28 these tests are not currently parallelizable (do not use this tox flag)
29 I would use setup_module, however that can't take monkeypatch fixtures
30 Currently looking for the best way to make this better:
31 https://stackoverflow.com/questions/60886013/python-monkeypatch-in-pytest-setup-module
35 def init_config_file():
36 with open(config_file_path, "w") as file:
37 file.write('{ "example_int" : 0 }')
40 def write_config_file():
41 # generate an inotify/config event
42 with open(config_file_path, "w") as file:
43 file.write('{ "example_int" : 1 }')
46 def test_init_xapp(monkeypatch, ue_metrics, cell_metrics_1, cell_metrics_2, cell_metrics_3, ue_metrics_with_bad_cell):
47 # monkeypatch post_init to set the data we want in SDL
48 # the metrics arguments are JSON (dict) objects
50 _original_post_init = main.post_init
52 def fake_post_init(self):
53 _original_post_init(self)
54 self.sdl_set(data.UE_NS, "12345", json.dumps(ue_metrics).encode(), usemsgpack=False)
55 self.sdl_set(data.UE_NS, "8675309", json.dumps(ue_metrics_with_bad_cell).encode(), usemsgpack=False)
56 self.sdl_set(data.CELL_NS, "310-680-200-555001", json.dumps(cell_metrics_1).encode(), usemsgpack=False)
57 self.sdl_set(data.CELL_NS, "310-680-200-555002", json.dumps(cell_metrics_2).encode(), usemsgpack=False)
58 self.sdl_set(data.CELL_NS, "310-680-200-555003", json.dumps(cell_metrics_3).encode(), usemsgpack=False)
61 monkeypatch.setattr("qpdriver.main.post_init", fake_post_init)
67 main.start(thread=True)
69 # wait a bit then update config
74 def test_rmr_flow(monkeypatch, qpd_to_qp, qpd_to_qp_bad_cell):
76 this flow mocks out the xapps on both sides of QP driver.
77 It first stands up a mock qp, then it starts up a mock ts
78 which will immediately send requests to the running qp driver.
83 # define a mock qp predictor
84 def mock_qp_default_handler(self, summary, sbuf):
87 def mock_qp_predict_handler(self, summary, sbuf):
88 nonlocal expected_result # closures ftw
89 pay = json.loads(summary["payload"])
90 expected_result[pay["PredictionUE"]] = pay
93 mock_qp_xapp = RMRXapp(mock_qp_default_handler, rmr_port=4666, use_fake_sdl=True)
94 mock_qp_xapp.register_callback(mock_qp_predict_handler, 30001)
95 mock_qp_xapp.run(thread=True)
99 # define a mock traffic steering xapp
100 def mock_ts_entry(self):
102 # make sure a bad steering request doesn't blow up in qpd
103 val = "notevenjson".encode()
104 self.rmr_send(val, 30000)
105 val = json.dumps({"bad": "tothebone"}).encode() # json but missing UEPredictionSet
106 self.rmr_send(val, 30000)
108 # valid request body but missing cell id
109 val = json.dumps({"UEPredictionSet": ["VOIDOFLIGHT"]}).encode()
110 self.rmr_send(val, 30000)
112 # good traffic steering request
113 val = json.dumps({"UEPredictionSet": ["12345", "8675309"]}).encode()
114 self.rmr_send(val, 30000)
116 # should trigger the default handler and do nothing
117 val = json.dumps({"test send 60001": 2}).encode()
118 self.rmr_send(val, 60001)
121 mock_ts_xapp = Xapp(entrypoint=mock_ts_entry, rmr_port=4564, use_fake_sdl=True)
122 mock_ts_xapp.run() # this will return since entry isn't a loop
126 assert expected_result == {"12345": qpd_to_qp, "8675309": qpd_to_qp_bad_cell}
127 assert main.get_stats() == {"DefCalled": 1, "SteeringRequests": 4}
129 # break SDL and send traffic again
130 def sdl_healthcheck_fails(self):
132 monkeypatch.setattr("ricxappframe.xapp_sdl.SDLWrapper.healthcheck", sdl_healthcheck_fails)
135 # restore SDL and send traffic once more
136 def sdl_healthcheck_passes(self):
138 monkeypatch.setattr("ricxappframe.xapp_sdl.SDLWrapper.healthcheck", sdl_healthcheck_passes)
142 def teardown_module():
144 this is like a "finally"; the name of this function is pytest magic
145 safer to put down here since certain failures above can lead to pytest never returning
146 for example if an exception gets raised before stop is called in any test function above,
147 pytest will hang forever
149 with suppress(Exception):
151 with suppress(Exception):
153 with suppress(Exception):