tests/test_alarm.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 ricxappframe.alarm import alarm
  21 from ricxappframe.alarm.alarm import AlarmAction, AlarmDetail, AlarmManager, AlarmSeverity, ALARM_MGR_SERVICE_NAME_ENV, ALARM_MGR_SERVICE_PORT_ENV
  22 from ricxappframe.alarm.exceptions import InitFailed
  23 from ricxappframe.rmr import rmr
  24
  25 MRC_SEND = None
  26 MRC_RCV = None
  27 SIZE = 256
  28
  29
  30 def setup_module():
  31     """
  32     test alarm module setup
  33     """
  34     global MRC_SEND
  35     MRC_SEND = rmr.rmr_init(b"4566", rmr.RMR_MAX_RCV_BYTES, 0x00)
  36     while rmr.rmr_ready(MRC_SEND) == 0:
  37         time.sleep(1)
  38
  39     global MRC_RCV
  40     MRC_RCV = rmr.rmr_init(b"4567", rmr.RMR_MAX_RCV_BYTES, 0x00)
  41     while rmr.rmr_ready(MRC_RCV) == 0:
  42         time.sleep(1)
  43
  44
  45 def teardown_module():
  46     """
  47     test alarm module teardown
  48     """
  49     rmr.rmr_close(MRC_SEND)
  50     rmr.rmr_close(MRC_RCV)
  51
  52
  53 def test_alarm_set_get(monkeypatch):
  54     """
  55     test set functions
  56     """
  57     act = AlarmAction.RAISE
  58     assert act is not None
  59
  60     sev = AlarmSeverity.CRITICAL
  61     assert sev is not None
  62
  63     det = AlarmDetail("1", "2", 3, AlarmSeverity.MINOR, "4", "5")
  64     assert det[alarm.KEY_MANAGED_OBJECT_ID] == "1"
  65     assert det[alarm.KEY_APPLICATION_ID] == "2"
  66     assert det[alarm.KEY_SPECIFIC_PROBLEM] == 3
  67     assert det[alarm.KEY_PERCEIVED_SEVERITY] == AlarmSeverity.MINOR.name
  68     assert det[alarm.KEY_IDENTIFYING_INFO] == "4"
  69     assert det[alarm.KEY_ADDITIONAL_INFO] == "5"
  70
  71     # missing environment variables
  72     with pytest.raises(InitFailed):
  73         alarm.AlarmManager(MRC_SEND, "missing", "envvars")
  74
  75     # invalid environment variables
  76     monkeypatch.setenv(ALARM_MGR_SERVICE_NAME_ENV, "0")
  77     monkeypatch.setenv(ALARM_MGR_SERVICE_PORT_ENV, "a")
  78     with pytest.raises(InitFailed):
  79         alarm.AlarmManager(MRC_SEND, "bogus", "envvars")
  80
  81     # good environment variables
  82     monkeypatch.setenv(ALARM_MGR_SERVICE_NAME_ENV, "127.0.0.1")     # do NOT use localhost
  83     monkeypatch.setenv(ALARM_MGR_SERVICE_PORT_ENV, "4567")  # any int is ok here
  84     mgr = alarm.AlarmManager(MRC_SEND, "moid2", "appid2")
  85     assert mgr is not None
  86     assert mgr.managed_object_id == "moid2"
  87     assert mgr.application_id == "appid2"
  88
  89
  90 def _receive_alarm_msg(action: AlarmAction):
  91     """
  92     delays briefly, receives a message, checks the message type and action
  93     """
  94     time.sleep(0.5)
  95     sbuf_rcv = rmr.rmr_alloc_msg(MRC_RCV, SIZE)
  96     sbuf_rcv = rmr.rmr_torcv_msg(MRC_RCV, sbuf_rcv, 2000)
  97     rcv_summary = rmr.message_summary(sbuf_rcv)
  98     assert rcv_summary[rmr.RMR_MS_MSG_STATE] == rmr.RMR_OK
  99     assert rcv_summary[rmr.RMR_MS_MSG_TYPE] == alarm.RIC_ALARM_UPDATE
 100     # parse JSON
 101     data = json.loads(rcv_summary[rmr.RMR_MS_PAYLOAD].decode())
 102     assert data[alarm.KEY_ALARM_ACTION] == action.name
 103
 104
 105 def test_alarm_manager(monkeypatch):
 106     """
 107     test send functions and ensure a message arrives
 108     """
 109     monkeypatch.setenv(ALARM_MGR_SERVICE_NAME_ENV, "127.0.0.1")     # do NOT use localhost
 110     monkeypatch.setenv(ALARM_MGR_SERVICE_PORT_ENV, "4567")  # must match rcv port above
 111     mgr = AlarmManager(MRC_SEND, "moid", "appid")
 112     assert mgr is not None
 113
 114     det = mgr.create_alarm(3, AlarmSeverity.DEFAULT, "identifying", "additional")
 115     assert det is not None
 116
 117     success = mgr.raise_alarm(det)
 118     assert success
 119     _receive_alarm_msg(AlarmAction.RAISE)
 120
 121     success = mgr.clear_alarm(det)
 122     assert success
 123     _receive_alarm_msg(AlarmAction.CLEAR)
 124
 125     success = mgr.reraise_alarm(det)
 126     assert success
 127     _receive_alarm_msg(AlarmAction.CLEAR)
 128     _receive_alarm_msg(AlarmAction.RAISE)
 129
 130     success = mgr.clear_all_alarms()
 131     assert success
 132     _receive_alarm_msg(AlarmAction.CLEARALL)