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
  68
  69 def test_default_config_handler(monkeypatch):
  70     """Just for coverage"""
  71     init_config_file()
  72     monkeypatch.setenv(Constants.CONFIG_FILE_ENV, config_file_path)
  73
  74     def default_rmr_handler(self, summary, sbuf):
  75         pass
  76
  77     # listen port is irrelevant, no messages arrive
  78     global rmr_xapp_defconfig
  79     rmr_xapp_defconfig = RMRXapp(default_rmr_handler, rmr_port=4567, use_fake_sdl=True)
  80     # in unit tests we need to thread here or else execution is not returned!
  81     rmr_xapp_defconfig.run(thread=True, rmr_timeout=1)
  82     write_config_file()
  83     # give the work loop a chance to timeout on RMR and process the config event
  84     time.sleep(3)
  85     rmr_xapp_defconfig.stop()
  86
  87
  88 def test_custom_config_handler(monkeypatch):
  89     # point watcher at the file
  90     init_config_file()
  91     monkeypatch.setenv(Constants.CONFIG_FILE_ENV, config_file_path)
  92
  93     def default_handler(self, summary, sbuf):
  94         pass
  95
  96     startup_config_event = False
  97     change_config_event = False
  98
  99     def config_handler(self, json):
 100         mdc_logger.info("config_handler: json {}".format(json))
 101         nonlocal startup_config_event
 102         nonlocal change_config_event
 103         if "start" in json:
 104             startup_config_event = True
 105         if "change" in json:
 106             change_config_event = True
 107
 108     # listen port is irrelevant, no messages arrive
 109     global rmr_xapp_config
 110     rmr_xapp_config = RMRXapp(default_handler, config_handler=config_handler, rmr_port=4567, use_fake_sdl=True)
 111     assert startup_config_event
 112     rmr_xapp_config.run(thread=True, rmr_timeout=1)  # in unit tests we need to thread here or else execution is not returned!
 113     write_config_file()
 114     # give the work loop a chance to timeout on RMR and process the config event
 115     time.sleep(3)
 116     assert change_config_event
 117     rmr_xapp_config.stop()
 118
 119
 120 def teardown_module():
 121     """
 122     this is like a "finally"; the name of this function is pytest magic
 123     safer to put down here since certain failures above can lead to pytest never returning
 124     for example if an exception gets raised before stop is called in any test function above,
 125     pytest will hang forever
 126     """
 127     os.remove(config_file_path)
 128     with suppress(Exception):
 129         rmr_xapp_config.stop()
 130     with suppress(Exception):
 131         rmr_xapp_defconfig.stop()
 132     with suppress(Exception):
 133         rmr_xapp_noconfig.stop()