run_test.py

   1 import configparser
   2 import importlib
   3 import importlib.util
   4 import logging.config
   5 import os
   6 import sys
   7 import glob
   8
   9 from onapsdk.exceptions import ModuleError
  10 import onaptests.utils.exceptions as onap_test_exceptions
  11
  12 SETTING_FILE_EXCEPTIONS = {
  13     "clearwater_ims": "clearwater_ims_nomulticloud_settings",
  14     "basic_cnf": "basic_cnf_yaml_settings",
  15     "basic_cds": "cba_enrichment_settings",
  16     "basic_network": "basic_network_nomulticloud_settings",
  17     "multi_vnf_macro": "multi_vnf_ubuntu_settings",
  18     "add_pnf_in_running_service": "instantiate_pnf_without_registration_event_settings"
  19 }
  20
  21 MODULES_TO_RELOAD = [
  22     "onapsdk",
  23     "onaptests"
  24 ]
  25
  26 def get_entrypoints():
  27     config = configparser.ConfigParser()
  28     config.read('setup.cfg')
  29     entry_points = config['entry_points']['xtesting.testcase']
  30     config = configparser.ConfigParser()
  31     config.read_string(f"[entry_points]\n{entry_points}")
  32     entry_points = config['entry_points']
  33     entry_points_result = {}
  34     for test_name, entrypoint in entry_points.items():
  35         test_scenario_module, test_class = entrypoint.split(":")
  36         entry_points_result[test_name] = {
  37             "module": test_scenario_module,
  38             "class": test_class
  39         }
  40     return entry_points_result
  41
  42 def check_scenarios(scenarios, entry_points):
  43     path = importlib.util.find_spec(scenarios.__name__).submodule_search_locations
  44     modules = glob.glob(f"{path._path[0]}/*.py")
  45     all_mods = [ os.path.basename(f)[:-3] for f in modules if os.path.isfile(f) and not f.endswith('__init__.py')]
  46     for mod_name in all_mods:
  47         full_mod_name = f"{scenarios.__name__}.{mod_name}"
  48         if mod_name != "scenario_base":
  49             exists = False
  50             for test_name, entry_point in entry_points.items():
  51                 if entry_point["module"] == full_mod_name:
  52                     exists = True
  53                     break
  54             if not exists:
  55                 raise onap_test_exceptions.TestConfigurationException(
  56                     f"Scenario defined in {full_mod_name}.py is not added to setup.cfg file")
  57
  58 def run_test(test_name, validation, force_cleanup, entry_point):
  59     print(f"Configuring {test_name} test")
  60     settings_env = "ONAP_PYTHON_SDK_SETTINGS"
  61     if force_cleanup:
  62         validation_env = "PYTHON_SDK_TESTS_FORCE_CLEANUP"
  63         os.environ[validation_env] = "True"
  64
  65     setting_file_name = f"{test_name}_settings"
  66     if test_name in SETTING_FILE_EXCEPTIONS:
  67         setting_file_name = SETTING_FILE_EXCEPTIONS[test_name]
  68     os.environ[settings_env] = f"onaptests.configuration.{setting_file_name}"
  69     try:
  70         basic_settings = importlib.import_module("onaptests.configuration.settings")
  71         if validation:
  72             basic_settings.IF_VALIDATION = True
  73         settings_module = importlib.import_module("onapsdk.configuration")
  74     except ModuleError:
  75         raise onap_test_exceptions.TestConfigurationException(
  76             f"Expected setting file {os.environ[settings_env]} not found")
  77
  78     if validation:
  79         settings_module.settings.CLEANUP_FLAG = True
  80         settings_module.settings.CLEANUP_ACTIVITY_TIMER = 1
  81         settings_module.settings.SDC_CLEANUP = True
  82         settings_module.settings.SERVICE_DISTRIBUTION_SLEEP_TIME = 1
  83
  84     # logging configuration for onapsdk, it is not requested for onaptests
  85     # Correction requested in onapsdk to avoid having this duplicate code
  86     logging.config.dictConfig(settings_module.settings.LOG_CONFIG)
  87     logger = logging.getLogger(test_name)
  88     logger.info(f"Running {test_name} test")
  89
  90     scenarios = importlib.import_module("onaptests.scenario")
  91     test_module = importlib.import_module(entry_point["module"])
  92
  93     test_instance = getattr(test_module, entry_point["class"])()
  94     if validation:
  95         validate_scenario_base_class(test_name, test_instance, scenarios)
  96     test_instance.run()
  97     test_instance.clean()
  98     if validation:
  99         logger.info(f"Validating {test_name} test")
 100         test_instance.validate()
 101     return scenarios
 102
 103 def validate_scenario_base_class(test_name, scenario, scenarios):
 104     has_scenario_base = False
 105     if test_name != scenario.case_name:
 106         raise onap_test_exceptions.TestConfigurationException(
 107             f"[{test_name}] Mismatch of scenario case name: {test_name} != {scenario.case_name}")
 108     if scenario.__class__.run != scenarios.scenario_base.ScenarioBase.run:
 109         raise onap_test_exceptions.TestConfigurationException(
 110             f"[{test_name}] scenario class overrides run() method what is forbidden")
 111     if scenario.__class__.clean != scenarios.scenario_base.ScenarioBase.clean:
 112         raise onap_test_exceptions.TestConfigurationException(
 113             f"[{test_name}] scenario class overrides clean() method what is forbidden")
 114     for base in scenario.__class__.__bases__:
 115         if base.__name__ in "ScenarioBase":
 116             has_scenario_base = True
 117             break
 118     if not has_scenario_base:
 119         raise onap_test_exceptions.TestConfigurationException(
 120             f"[{test_name}] {scenario.__class__.__name__} class does not inherit from ScenarioBase")
 121
 122 def main(argv):
 123     """Script is used to run one or all the tests.
 124
 125     You need to specify a name of the test like 'basic_cps' or
 126     keyword 'all' that tells to run all the tests. You can also
 127     pass a second argument of any value that tells the script to run
 128     test(s) in the validation mode that checks only a basic setup of
 129     steps (like cleanup) and their execution in a certain order.
 130
 131     Examplary use:
 132     - python run_test.py basic_vm_macro
 133     - python run_test.py basic_cps validation
 134     - python run_test.py all true
 135     """
 136     if len(argv) == 0:
 137         print("Required test name argument missing", file=sys.stderr)
 138         print("\nExample: python run_test.py basic_cps\n")
 139         exit(1)
 140     validation = len(argv) > 1
 141     force_cleanup = len(argv) > 2
 142     test_name = argv[0]
 143     entry_points = get_entrypoints()
 144     if test_name == "all":
 145         modules_reload = False
 146         scenarios = None
 147         for test_name, entry_point in entry_points.items():
 148             if modules_reload:
 149                 modules_to_keep = dict()
 150                 for module in sys.modules:
 151                     reload_module = False
 152                     for module_to_reload in MODULES_TO_RELOAD:
 153                         if module_to_reload in module:
 154                             reload_module = True
 155                             break
 156                     if not reload_module:
 157                         modules_to_keep[module] = sys.modules[module]
 158                 sys.modules.clear()
 159                 sys.modules.update(modules_to_keep)
 160             scenarios = run_test(
 161                 test_name, validation, force_cleanup, entry_point)
 162             modules_reload = True
 163         if validation:
 164             check_scenarios(scenarios, entry_points)
 165     else:
 166         entry_point = entry_points[test_name]
 167         run_test(test_name, validation, force_cleanup, entry_point)
 168
 169 if __name__ == "__main__":
 170     main(sys.argv[1:])