tests/test_config.py

   1 # ==================================================================================
   2 #       Copyright (c) 2020 Nokia
   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
  18 import time
  19 import os
  20 from contextlib import suppress
  21 from mdclogpy import Logger
  22
  23 from ricxappframe.util.constants import Constants
  24 from ricxappframe.xapp_frame import RMRXapp
  25
  26 mdc_logger = Logger(name=__name__)
  27 rmr_xapp_config = None
  28 rmr_xapp_defconfig = None
  29 rmr_xapp_noconfig = None
  30 config_file_path = "/tmp/file.json"
  31
  32
  33 def init_config_file():
  34     with open(config_file_path, "w") as file:
  35         file.write('{ "start" : "value" }')
  36
  37
  38 def write_config_file():
  39     # generate an inotify/config event
  40     with open(config_file_path, "w") as file:
  41         file.write('{ "change" : "value2" }')
  42
  43
  44 def test_config_no_env(monkeypatch):
  45     init_config_file()
  46     monkeypatch.delenv(Constants.CONFIG_FILE_ENV, raising=False)
  47
  48     def default_rmr_handler(self, summary, sbuf):
  49         pass
  50
  51     config_event_seen = False
  52
  53     def config_handler(self, json):
  54         nonlocal config_event_seen
  55         config_event_seen = True
  56
  57     global rmr_xapp_noconfig
  58     rmr_xapp_noconfig = RMRXapp(default_rmr_handler, config_handler=config_handler, rmr_port=4652, use_fake_sdl=True)
  59     # in unit tests we need to thread here or else execution is not returned!
  60     rmr_xapp_noconfig.run(thread=True, rmr_timeout=1)
  61
  62     write_config_file()
  63     # give the work loop a chance to timeout on RMR and process the config event
  64     time.sleep(3)
  65     assert not config_event_seen
  66     rmr_xapp_noconfig.stop()
  67     rmr_xapp_noconfig = None
  68
  69
  70 def test_default_config_handler(monkeypatch):
  71     """Just for coverage"""
  72     init_config_file()
  73     monkeypatch.setenv(Constants.CONFIG_FILE_ENV, config_file_path)
  74
  75     def default_rmr_handler(self, summary, sbuf):
  76         pass
  77
  78     # listen port is irrelevant, no messages arrive
  79     global rmr_xapp_defconfig
  80     rmr_xapp_defconfig = RMRXapp(default_rmr_handler, rmr_port=4567, use_fake_sdl=True)
  81     # in unit tests we need to thread here or else execution is not returned!
  82     rmr_xapp_defconfig.run(thread=True, rmr_timeout=1)
  83     write_config_file()
  84     # give the work loop a chance to timeout on RMR and process the config event
  85     time.sleep(3)
  86     rmr_xapp_defconfig.stop()
  87     rmr_xapp_defconfig = None
  88
  89
  90 def test_custom_config_handler(monkeypatch):
  91     # point watcher at the file
  92     init_config_file()
  93     monkeypatch.setenv(Constants.CONFIG_FILE_ENV, config_file_path)
  94
  95     def default_handler(self, summary, sbuf):
  96         pass
  97
  98     startup_config_event = False
  99     change_config_event = False
 100
 101     def config_handler(self, json):
 102         mdc_logger.info("config_handler: json {}".format(json))
 103         nonlocal startup_config_event
 104         nonlocal change_config_event
 105         if "start" in json:
 106             startup_config_event = True
 107         if "change" in json:
 108             change_config_event = True
 109
 110     # listen port is irrelevant, no messages arrive
 111     global rmr_xapp_config
 112     rmr_xapp_config = RMRXapp(default_handler, config_handler=config_handler, rmr_port=4567, use_fake_sdl=True)
 113     assert startup_config_event
 114     rmr_xapp_config.run(thread=True, rmr_timeout=1)  # in unit tests we need to thread here or else execution is not returned!
 115     write_config_file()
 116     # give the work loop a chance to timeout on RMR and process the config event
 117     time.sleep(3)
 118     assert change_config_event
 119     rmr_xapp_config.stop()
 120     rmr_xapp_config = None
 121
 122
 123 def teardown_module():
 124     """
 125     this is like a "finally"; the name of this function is pytest magic
 126     safer to put down here since certain failures above can lead to pytest never returning
 127     for example if an exception gets raised before stop is called in any test function above,
 128     pytest will hang forever
 129     """
 130     os.remove(config_file_path)
 131     with suppress(Exception):
 132         if rmr_xapp_config:
 133             rmr_xapp_config.stop()
 134         if rmr_xapp_defconfig:
 135             rmr_xapp_defconfig.stop()
 136         if rmr_xapp_noconfig:
 137             rmr_xapp_noconfig.stop()