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