1 # =================================================================================2
2 # Copyright (c) 2020 AT&T Intellectual Property.
3 # Copyright (c) 2020 Nokia
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
20 from mdclogpy import Logger
21 from ricxappframe.metric import metric
22 from ricxappframe.metric.exceptions import EmptyReport
23 from ricxappframe.metric.metric import MetricData, MetricsReport, MetricsManager
24 from ricxappframe.rmr import rmr
26 mdc_logger = Logger(name=__name__)
34 test metric module setup
37 MRC_SEND = rmr.rmr_init(b"4568", rmr.RMR_MAX_RCV_BYTES, 0x00)
38 while rmr.rmr_ready(MRC_SEND) == 0:
42 MRC_RCV = rmr.rmr_init(b"4569", rmr.RMR_MAX_RCV_BYTES, 0x00)
43 while rmr.rmr_ready(MRC_RCV) == 0:
46 # let all the RMR output appear
48 mdc_logger.debug("RMR instances initialized")
51 def teardown_module():
53 test metric module teardown
55 mdc_logger.debug("Closing RMR instances")
56 rmr.rmr_close(MRC_SEND)
57 rmr.rmr_close(MRC_RCV)
60 def test_metric_set_get(monkeypatch):
61 mdc_logger.debug("Testing set functions")
62 md = MetricData("id", "value", "type")
63 assert md[metric.KEY_DATA_ID] == "id"
64 assert md[metric.KEY_DATA_VALUE] == "value"
65 assert md[metric.KEY_DATA_TYPE] == "type"
67 mr = MetricsReport("reporter", "generator", [md])
68 assert mr[metric.KEY_REPORTER] == "reporter"
69 assert mr[metric.KEY_GENERATOR] == "generator"
70 assert len(mr[metric.KEY_DATA]) == 1
71 assert mr[metric.KEY_DATA][0] == md
73 mgr = metric.MetricsManager(MRC_SEND, "reporter", "generator")
74 assert mgr is not None
75 assert mgr.reporter == "reporter"
76 assert mgr.generator == "generator"
78 mr = MetricsReport("empty", "empty", [])
79 with pytest.raises(EmptyReport):
83 def _receive_metric_rpt(rptr: str):
85 delays briefly, receives a message, checks the message type and reporter
88 sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
89 sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
90 rcv_summary = rmr.message_summary(sbuf_rcv)
91 mdc_logger.debug("Receive result is {}".format(rcv_summary[rmr.RMR_MS_MSG_STATE]))
92 assert rcv_summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
93 assert rcv_summary[rmr.RMR_MS_MSG_TYPE] == metric.RIC_METRICS
95 data = json.loads(rcv_summary[rmr.RMR_MS_PAYLOAD].decode())
96 mdc_logger.debug("Received reporter {}".format(data[metric.KEY_REPORTER]))
97 assert data[metric.KEY_REPORTER] == rptr
100 def test_metrics_manager():
102 test send functions and ensure a message arrives
104 mdc_logger.debug("Testing metrics-send functions")
107 mgr = MetricsManager(MRC_SEND, rptr, "generator")
108 assert mgr is not None
110 md = MetricData("id", "value", "type")
111 assert md is not None
113 success = mgr.send_metrics([md])
115 _receive_metric_rpt(rptr)