tests/test_metric.py

   1 # =================================================================================2
   2 #       Copyright (c) 2020 AT&T Intellectual Property.
   3 #       Copyright (c) 2020 Nokia
   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 pytest
  19 import time
  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
  25
  26 mdc_logger = Logger(name=__name__)
  27 MRC_SEND = None
  28 MRC_RCV = None
  29 SIZE = 256
  30
  31
  32 def setup_module():
  33     """
  34     test metric module setup
  35     """
  36     global MRC_SEND
  37     MRC_SEND = rmr.rmr_init(b"4568", rmr.RMR_MAX_RCV_BYTES, 0x00)
  38     while rmr.rmr_ready(MRC_SEND) == 0:
  39         time.sleep(1)
  40
  41     global MRC_RCV
  42     MRC_RCV = rmr.rmr_init(b"4569", rmr.RMR_MAX_RCV_BYTES, 0x00)
  43     while rmr.rmr_ready(MRC_RCV) == 0:
  44         time.sleep(1)
  45
  46     # let all the RMR output appear
  47     time.sleep(2)
  48     mdc_logger.debug("RMR instances initialized")
  49
  50
  51 def teardown_module():
  52     """
  53     test metric module teardown
  54     """
  55     mdc_logger.debug("Closing RMR instances")
  56     rmr.rmr_close(MRC_SEND)
  57     rmr.rmr_close(MRC_RCV)
  58
  59
  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"
  66
  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
  72
  73     mgr = metric.MetricsManager(MRC_SEND, "reporter", "generator")
  74     assert mgr is not None
  75     assert mgr.reporter == "reporter"
  76     assert mgr.generator == "generator"
  77
  78     mr = MetricsReport("empty", "empty", [])
  79     with pytest.raises(EmptyReport):
  80         mgr.send_report(mr)
  81
  82
  83 def _receive_metric_rpt(rptr: str):
  84     """
  85     delays briefly, receives a message, checks the message type and reporter
  86     """
  87     time.sleep(0.5)
  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
  94     # parse JSON
  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
  98
  99
 100 def test_metrics_manager():
 101     """
 102     test send functions and ensure a message arrives
 103     """
 104     mdc_logger.debug("Testing metrics-send functions")
 105     rptr = "metr"
 106
 107     mgr = MetricsManager(MRC_SEND, rptr, "generator")
 108     assert mgr is not None
 109
 110     md = MetricData("id", "value", "type")
 111     assert md is not None
 112
 113     success = mgr.send_metrics([md])
 114     assert success
 115     _receive_metric_rpt(rptr)