+++ /dev/null
-#!/usr/bin/python
-#
-import logging.config
-from onapsdk.configuration import settings
-from onaptests.steps.loop.clamp import ClampStep
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Instantiate Basic Clamp")
-
- basic_clamp = ClampStep(
- cleanup=settings.CLEANUP_FLAG)
- basic_clamp.execute()
- basic_clamp.reports_collection.generate_report()
+++ /dev/null
-import logging.config
-import time
-import onaptests.utils.exceptions as onap_test_exceptions
-from onapsdk.configuration import settings
-from onaptests.scenario.basic_cnf_macro import BasicCnfMacro
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Instantiate Apache CNF with macro procedure")
- step = BasicCnfMacro(cleanup=settings.CLEANUP_FLAG)
- try:
- step.execute()
- if settings.CLEANUP_FLAG:
- logger.info("Starting to clean up in {} seconds".format(settings.CLEANUP_ACTIVITY_TIMER))
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- step.cleanup()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Ubuntu NSO VM configuration error")
- except onap_test_exceptions.ServiceInstantiateException:
- logger.error("Ubuntu NSO VM instantiation error")
- except onap_test_exceptions.ServiceCleanupException:
- logger.error("Ubuntu NSO VM instance cleanup error")
- except onap_test_exceptions.VnfInstantiateException:
- logger.error("Ubuntu NSO VM Vnf instantiation error")
- except onap_test_exceptions.VnfCleanupException:
- logger.error("Ubuntu NSO VM Vnf instance cleanup error")
- except onap_test_exceptions.VfModuleInstantiateException:
- logger.error("Ubuntu NSO VM Module instantiation error")
- except onap_test_exceptions.VfModuleCleanupException:
- logger.error("Ubuntu NSO VM Module cleanup error")
+++ /dev/null
-import logging.config
-import onaptests.utils.exceptions as onap_test_exceptions
-from onapsdk.configuration import settings
-from onaptests.steps.onboard.cps import CreateCpsAnchorStep, CreateCpsSchemaSetStep, CreateCpsDataspaceStep, CreateCpsAnchorNodeStep
-
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Basic CPS")
-
- basic_cps = CreateCpsAnchorNodeStep(
- cleanup=settings.CLEANUP_FLAG)
- try:
- basic_cps.execute()
- basic_cps.cleanup()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Basic CPS configuration error")
- basic_cps.reports_collection.generate_report()
+++ /dev/null
-import logging.config
-import time
-from onapsdk.configuration import settings
-from onaptests.steps.instantiate.vl_ala_carte import YamlTemplateVlAlaCarteInstantiateStep
-
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Instantiate Basic Network without multicloud")
-
- basic_network_instantiate = YamlTemplateVlAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- basic_network_instantiate.execute()
- if settings.CLEANUP_FLAG:
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- try:
- basic_network_instantiate.cleanup()
- except ValueError as error:
- logger.info("service instance deleted as expected {0}".format(error))
- basic_network_instantiate.reports_collection.generate_report()
+++ /dev/null
-import logging.config
-import onaptests.utils.exceptions as onap_test_exceptions
-from onapsdk.configuration import settings
-from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
-
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Basic Onboard")
-
- basic_vm_onboard = YamlTemplateServiceOnboardStep(
- cleanup=settings.CLEANUP_FLAG)
- try:
- basic_vm_onboard.execute()
- basic_vm_onboard.cleanup()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Basic Onboard configuration error")
- basic_vm_onboard.reports_collection.generate_report()
+++ /dev/null
-import logging.config
-from onapsdk.configuration import settings
-
-from onaptests.scenario.basic_sdnc import BasicSdnc
-import onaptests.utils.exceptions as onap_test_exceptions
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Basic SDNC")
-
- basic_sdnc = BasicSdnc(cleanup=settings.CLEANUP_FLAG)
- try:
- basic_sdnc.run()
- basic_sdnc.clean()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Basic SDNC configuration error")
+++ /dev/null
-import logging.config
-from onapsdk.configuration import settings
-from onaptests.steps.instantiate.vf_module_ala_carte import YamlTemplateVfModuleAlaCarteInstantiateStep
-
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
-
- basic_vm_instantiate = YamlTemplateVfModuleAlaCarteInstantiateStep()
- basic_vm_instantiate.execute()
+++ /dev/null
-import logging.config
-import time
-import onaptests.utils.exceptions as onap_test_exceptions
-from onapsdk.configuration import settings
-from onaptests.steps.instantiate.vf_module_ala_carte import YamlTemplateVfModuleAlaCarteInstantiateStep
-
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Instantiate Basic VM without multicloud")
-
- basic_vm_instantiate = YamlTemplateVfModuleAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- try:
- basic_vm_instantiate.execute()
- if settings.CLEANUP_FLAG:
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- basic_vm_instantiate.cleanup()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Basic VM configuration error")
- except onap_test_exceptions.ServiceInstantiateException:
- logger.error("Basic VM instantiation error")
- except onap_test_exceptions.ServiceCleanupException:
- logger.error("Basic VM instance cleanup error")
- except onap_test_exceptions.VnfInstantiateException:
- logger.error("Basic VM Vnf instantiation error")
- except onap_test_exceptions.VnfCleanupException:
- logger.error("Basic VM Vnf instance cleanup error")
- except onap_test_exceptions.VfModuleInstantiateException:
- logger.error("Basic VM Module instantiation error")
- except onap_test_exceptions.VfModuleCleanupException:
- logger.error("Basic VM Module cleanup error")
-
- basic_vm_instantiate.reports_collection.generate_report()
+++ /dev/null
-import logging.config
-import time
-import onaptests.utils.exceptions as onap_test_exceptions
-from onapsdk.configuration import settings
-from onaptests.scenario.multi_vnf_macro import MultiVnfUbuntuMacroStep
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Instantiate Ubuntu multi VNF without multicloud")
- step = MultiVnfUbuntuMacroStep(cleanup=settings.CLEANUP_FLAG)
- try:
- step.execute()
- if settings.CLEANUP_FLAG:
- logger.info("Starting to clean up in {} seconds".format(settings.CLEANUP_ACTIVITY_TIMER))
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- step.cleanup()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Ubuntu NSO VM configuration error")
- except onap_test_exceptions.ServiceInstantiateException:
- logger.error("Ubuntu NSO VM instantiation error")
- except onap_test_exceptions.ServiceCleanupException:
- logger.error("Ubuntu NSO VM instance cleanup error")
- except onap_test_exceptions.VnfInstantiateException:
- logger.error("Ubuntu NSO VM Vnf instantiation error")
- except onap_test_exceptions.VnfCleanupException:
- logger.error("Ubuntu NSO VM Vnf instance cleanup error")
- except onap_test_exceptions.VfModuleInstantiateException:
- logger.error("Ubuntu NSO VM Module instantiation error")
- except onap_test_exceptions.VfModuleCleanupException:
- logger.error("Ubuntu NSO VM Module cleanup error")
+++ /dev/null
-import logging.config
-import onaptests.utils.exceptions as onap_test_exceptions
-from onapsdk.configuration import settings
-from onaptests.scenario.status import Status
-
-
-if __name__ == "__main__":
- # logging configuration for onapsdk, it is not requested for onaptests
- # Correction requested in onapsdk to avoid having this duplicate code
- logging.config.dictConfig(settings.LOG_CONFIG)
- logger = logging.getLogger("Status Check")
-
- status = Status()
- try:
- status.run()
- status.clean()
- except onap_test_exceptions.TestConfigurationException:
- logger.error("Status Check configuration error")
--- /dev/null
+import configparser
+import importlib
+import logging.config
+import os
+import sys
+
+import onaptests.utils.exceptions as onap_test_exceptions
+
+
+def get_entrypoints():
+ config = configparser.ConfigParser()
+ config.read('setup.cfg')
+ entry_points = config['entry_points']['xtesting.testcase']
+ config = configparser.ConfigParser()
+ config.read_string(f"[entry_points]\n{entry_points}")
+ entry_points = config['entry_points']
+ entry_points_result = {}
+ for test_name, entrypoint in entry_points.items():
+ test_scenario_module, test_class = entrypoint.split(":")
+ entry_points_result[test_name] = {
+ "module": test_scenario_module,
+ "class": test_class
+ }
+ return entry_points_result
+
+def run_test(test_name, validation, entry_point, settings_module):
+ settings_env = "ONAP_PYTHON_SDK_SETTINGS"
+ if validation:
+ validation_env = "PYTHON_SDK_TESTS_VALIDATION"
+ os.environ[validation_env] = "True"
+ os.environ[settings_env] = f"onaptests.configuration.{test_name}_settings"
+ if not settings_module:
+ settings_module = importlib.import_module("onapsdk.configuration")
+ else:
+ settings_module = importlib.reload(settings_module)
+ settings = settings_module.settings
+ # logging configuration for onapsdk, it is not requested for onaptests
+ # Correction requested in onapsdk to avoid having this duplicate code
+ logging.config.dictConfig(settings.LOG_CONFIG)
+ logger = logging.getLogger(test_name)
+ logger.info(f"Running {test_name} test")
+
+ test_module = importlib.import_module(entry_point["module"])
+
+ test_instance = getattr(test_module, entry_point["class"])()
+ try:
+ test_instance.run()
+ test_instance.clean()
+ if validation:
+ test_instance.validate_execution()
+ except onap_test_exceptions.TestConfigurationException:
+ logger.error("Status Check configuration error")
+ return settings_module
+
+def main(argv):
+ """Script is used to run one or all the tests.
+
+ You need to specify a name of the test like 'basic_cps' or
+ keyword 'all' that tells to run all the tests. You can also
+ pass a second argument of any value that tells the script to run
+ test(s) in the validation mode that checks only a basic setup of
+ steps (like cleanup) and their execution in a certain order.
+
+ Examplary use:
+ - python run_test.py basic_vm_macro
+ - python run_test.py basic_cps validation
+ - python run_test.py all true
+ """
+ if len(argv) == 0:
+ print("Required test name argument missing", file=sys.stderr)
+ print("\nExample: python run_test.py basic_cps\n")
+ exit(1)
+ validation = len(argv) > 1
+ test_name = argv[0]
+ entry_points = get_entrypoints()
+ if test_name == "all":
+ settings_module = None
+ for test_name, entry_point in entry_points.items():
+ settings_module = run_test(test_name, validation, entry_point, settings_module)
+ else:
+ entry_point = entry_points[test_name]
+ run_test(test_name, validation, entry_point, None)
+
+if __name__ == "__main__":
+ main(sys.argv[1:])
license = Apache 2.0
classifiers =
Programming Language :: Python :: 3
- Programming Language :: Python :: 3.7
- Programming Language :: Python :: 3.8
+ Programming Language :: Python :: 3.11
[options]
zip_safe = False
include_package_data = True
-python_requires = >=3.7,<4
+python_requires = >=3.11,<4
package_dir=
=src
packages=find_namespace:
basic_vm_macro_stability = onaptests.scenario.basic_vm_macro_stability:BasicVmMacroStability
basic_network = onaptests.scenario.basic_network:BasicNetwork
basic_cnf = onaptests.scenario.basic_cnf:BasicCnf
- basic_cds = onaptests.scenario.cds_blueprint_enrichment:CDSBlueprintEnrichment
+ basic_cds = onaptests.scenario.basic_cds:CDSBlueprintEnrichment
clearwater_ims = onaptests.scenario.clearwater_ims:ClearwaterIms
basic_onboard = onaptests.scenario.basic_onboard:BasicOnboard
pnf_macro = onaptests.scenario.pnf_macro:PnfMacro
basic_clamp = onaptests.scenario.basic_clamp:BasicClamp
cds_resource_resolution = onaptests.scenario.cds_resource_resolution:CDSResourceResolution
- multi_vnf_ubuntu_macro = onaptests.scenario.multi_vnf_macro:MultiVnfUbuntuMacro
+ multi_vnf_macro = onaptests.scenario.multi_vnf_macro:MultiVnfUbuntuMacro
basic_cnf_macro = onaptests.scenario.basic_cnf_macro:BasicCnfMacro
basic_cps = onaptests.scenario.basic_cps:BasicCps
status = onaptests.scenario.status:Status
}
}
CLEANUP_FLAG = False
+CLEANUP_ACTIVITY_TIMER = 5
SDC_CLEANUP = False
REPORTING_FILE_DIRECTORY = "/tmp/"
CHECK_POD_VERSIONS = True
IGNORE_EMPTY_REPLICAS = False
STATUS_DETAILS_JSON = "status-details.json"
+INCLUDE_ALL_RES_IN_DETAILS = True
FULL_LOGS_CONTAINERS = [
'dcae-bootstrap', 'dcae-cloudify-manager', 'aai-resources',
--- /dev/null
+#!/usr/bin/env python
+"""Simple CDS blueprint erichment test scenario."""
+from onaptests.scenario.scenario_base import ScenarioBase
+from onaptests.steps.onboard.cds import CbaEnrichStep
+
+
+class CDSBlueprintEnrichment(ScenarioBase):
+ """Enrich simple blueprint using CDS blueprintprocessor."""
+
+ def __init__(self, **kwargs):
+ """Init CDS blueprint enrichment use case."""
+ super().__init__('basic_cds', **kwargs)
+ self.test = CbaEnrichStep()
"""Basic Clamp test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import APIError, SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.loop.clamp import ClampStep
-from onaptests.utils.exceptions import OnapTestException
class BasicClamp(ScenarioBase):
"""Onboard, update a model with a loop, design the loop and deploy it."""
- __logger = logging.getLogger(__name__)
def __init__(self, **kwargs):
"""Init Basic Clamp, onboard a VM, design and deploy a loop with CLAMP."""
super().__init__('basic_clamp', **kwargs)
- self.test = ClampStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run Basic CLAMP onap test."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.__logger.info("VNF basic_clamp successfully created")
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("VNF basic_clamp cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except (SDKException, APIError):
- self.result = 0
- self.__logger.exception("SDK exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = ClampStep()
#!/usr/bin/env python
"""Basic CNF test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.instantiate.vf_module_ala_carte import \
YamlTemplateVfModuleAlaCarteInstantiateStep
-from onaptests.utils.exceptions import OnapTestException
class BasicCnf(ScenarioBase):
"""Onboard then instantiate a simple CNF with ONAP."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init BasicCnf."""
super().__init__('basic_cnf', **kwargs)
- self.test = YamlTemplateVfModuleAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run onap_tests with basic_cnf VM."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.__logger.info("basic_cnf successfully created")
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("basic_cnf cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = YamlTemplateVfModuleAlaCarteInstantiateStep()
"""Instantiate basic cnf using SO macro flow."""
-import logging
-import time
-
from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaPublishStep
-from onaptests.utils.exceptions import OnapTestException
from yaml import SafeLoader, load
-class BasicCnfMacroStep(YamlTemplateBaseStep):
+class BasicCnfMacroStep(YamlTemplateBaseScenarioStep):
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- CbaPublishStep
- YamlTemplateServiceMacroInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self._yaml_template: dict = None
- self.add_step(CbaPublishStep(
- cleanup=cleanup
- ))
- self.add_step(YamlTemplateServiceMacroInstantiateStep(
- cleanup=cleanup
- ))
+ self.add_step(CbaPublishStep())
+ self.add_step(YamlTemplateServiceMacroInstantiateStep())
@property
def description(self) -> str:
str: Component name
"""
- return "PythonSDK-tests"
+ return "TEST"
@property
def yaml_template(self) -> dict:
class BasicCnfMacro(ScenarioBase):
"""Instantiate a basic cnf macro."""
- __logger = logging.getLogger()
-
def __init__(self, **kwargs):
"""Init Basic Cnf Macro use case."""
super().__init__('basic_cnf_macro', **kwargs)
- self.test = BasicCnfMacroStep(cleanup=settings.CLEANUP_FLAG)
-
- def run(self):
- """Run basic cnf macro test."""
- self.start_time = time.time()
- try:
- self.test.execute()
- self.test.cleanup()
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Generate report."""
- self.test.reports_collection.generate_report()
+ self.test = BasicCnfMacroStep()
#!/usr/bin/env python
"""Basic CPS test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.onboard.cps import CreateCpsAnchorNodeStep
-from onaptests.utils.exceptions import OnapTestException
class BasicCps(ScenarioBase):
"""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init Basic CPS."""
super().__init__('basic_cps', **kwargs)
- self.test = CreateCpsAnchorNodeStep(cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run basic_cps and create some CPS resources"""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.test.cleanup()
- self.__logger.info("Basic CPS test passed")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = CreateCpsAnchorNodeStep()
#!/usr/bin/env python
"""Basic VM test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.instantiate.vl_ala_carte import \
YamlTemplateVlAlaCarteInstantiateStep
-from onaptests.utils.exceptions import OnapTestException
class BasicNetwork(ScenarioBase):
"""Onboard then instantiate a simple Network with ONAP."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init Basic Network use case."""
# import basic_network_nomulticloud_settings needed
super().__init__('basic_network', **kwargs)
- self.test = YamlTemplateVlAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run onap_tests with basic network."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.__logger.info("Service basic_network successfully created")
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("Service basic_network cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- try:
- self.test.cleanup()
- except SDKException as error:
- self.__logger.info("service deletion error {0}".format(error))
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = YamlTemplateVlAlaCarteInstantiateStep()
#!/usr/bin/env python
"""Basic Onboard test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
-from onaptests.utils.exceptions import OnapTestException
class BasicOnboard(ScenarioBase):
"""Onboard a simple VM with ONAP."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init BasicOnboard."""
# import basic_onboard_settings needed
super().__init__('basic_onboard', **kwargs)
- self.test = YamlTemplateServiceOnboardStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run basic_onboard and onboard a simple service"""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.test.cleanup()
- self.__logger.info("VNF basic_vm successfully onboarded")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = YamlTemplateServiceOnboardStep()
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
+#!/usr/bin/env python
+"""Basic Onboard test case."""
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.instantiate.sdnc_service import TestSdncStep
-from onaptests.utils.exceptions import OnapTestException
class BasicSdnc(ScenarioBase):
Check and delete the service.
"""
- __logger = logging.getLogger()
-
def __init__(self, **kwargs):
"""Init Basic SDNC use case."""
super().__init__('basic_sdnc', **kwargs)
- self.test = TestSdncStep(cleanup=settings.CLEANUP_FLAG)
-
- def run(self):
- """Run basic SDNC test."""
- self.start_time = time.time()
- try:
- self.test.execute()
- self.test.cleanup()
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Generate report."""
- self.test.reports_collection.generate_report()
+ self.test = TestSdncStep()
#!/usr/bin/env python
"""Basic VM test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.instantiate.vf_module_ala_carte import \
YamlTemplateVfModuleAlaCarteInstantiateStep
-from onaptests.utils.exceptions import OnapTestException
class BasicVm(ScenarioBase):
"""Onboard then instantiate a simple VM with ONAP."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init BasicVM."""
super().__init__('basic_vm', **kwargs)
- self.test = YamlTemplateVfModuleAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run Basic VM onap test."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.__logger.info("VNF basic_vm successfully created")
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("VNF basic_vm cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = YamlTemplateVfModuleAlaCarteInstantiateStep()
"""Instantiate basic vm using SO macro flow."""
-import logging
-import time
-
from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaPublishStep
-from onaptests.utils.exceptions import OnapTestException
from yaml import SafeLoader, load
-class BasicVmMacroStep(YamlTemplateBaseStep):
+class BasicVmMacroStep(YamlTemplateBaseScenarioStep):
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- CbaPublishStep
- YamlTemplateServiceMacroInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self._yaml_template: dict = None
- self.add_step(CbaPublishStep(
- cleanup=cleanup
- ))
- self.add_step(YamlTemplateServiceMacroInstantiateStep(
- cleanup=cleanup
- ))
+ self.add_step(CbaPublishStep())
+ self.add_step(YamlTemplateServiceMacroInstantiateStep())
@property
def description(self) -> str:
class BasicVmMacro(ScenarioBase):
"""Instantiate a basic vm macro."""
- __logger = logging.getLogger()
-
def __init__(self, **kwargs):
"""Init Basic Macro use case."""
super().__init__('basic_vm_macro', **kwargs)
- self.test = BasicVmMacroStep(cleanup=settings.CLEANUP_FLAG)
-
- def run(self):
- """Run basic vm macro test."""
- self.start_time = time.time()
- try:
- self.test.execute()
- self.test.cleanup()
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Generate report."""
- self.test.reports_collection.generate_report()
+ self.test = BasicVmMacroStep()
"""Instantiate basic vm using SO macro flow."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
-from onaptests.utils.exceptions import OnapTestException
class BasicVmMacroStability(ScenarioBase):
"""Instantiate a basic vm macro."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init Basic Macro use case."""
super().__init__('basic_vm_macro_stability', **kwargs)
- self.test = YamlTemplateServiceMacroInstantiateStep(cleanup=settings.CLEANUP_FLAG)
-
- def run(self):
- """Run basic vm macro test."""
- self.start_time = time.time()
- try:
- self.test.execute()
- self.test.cleanup()
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Generate report."""
- self.test.reports_collection.generate_report()
+ self.test = YamlTemplateServiceMacroInstantiateStep()
+++ /dev/null
-#!/usr/bin/env python
-"""Simple CDS blueprint erichment test scenario."""
-
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.onboard.cds import CbaEnrichStep
-from onaptests.utils.exceptions import OnapTestException
-
-
-class CDSBlueprintEnrichment(ScenarioBase):
- """Enrich simple blueprint using CDS blueprintprocessor."""
-
- __logger = logging.getLogger()
-
- def __init__(self, **kwargs):
- """Init CDS blueprint enrichment use case."""
- super().__init__('basic_cds', **kwargs)
- self.test = CbaEnrichStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- self.__logger.debug("CDS blueprint enrichment run")
- self.start_time = time.time()
- try:
- self.test.execute()
- if settings.CLEANUP_FLAG:
- self.test.cleanup()
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
#!/usr/bin/env python
"""CDS resource resolution test scenario."""
-
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.base import BaseStep
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, BaseScenarioStep
from onaptests.steps.onboard.cds import CbaProcessStep
from onaptests.steps.simulator.cds_mockserver import \
CdsMockserverCnfConfigureStep
-from onaptests.utils.exceptions import OnapTestException
-class CDSResourceResolutionStep(BaseStep):
+class CDSResourceResolutionStep(BaseScenarioStep):
"""Step created to run scenario and generate report."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- CdsMockserverCnfConfigureStep,
- CbaProcessStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CdsMockserverCnfConfigureStep(
- cleanup=cleanup
- ))
- self.add_step(CbaProcessStep(
- cleanup=cleanup
- ))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CdsMockserverCnfConfigureStep())
+ self.add_step(CbaProcessStep())
@property
def description(self) -> str:
str: Component name
"""
- return "PythonSDK-tests"
+ return "TEST"
class CDSResourceResolution(ScenarioBase):
"""Enrich simple blueprint using CDS blueprintprocessor."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init CDS resource resolution use case."""
super().__init__('basic_cds', **kwargs)
- self.test = CDSResourceResolutionStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- self.__logger.debug("CDS resource resolution run")
- self.start_time = time.time()
- try:
- for test_phase in (self.test.execute, self.test.cleanup):
- try:
- test_phase()
- self.result += 50
- except OnapTestException as exc:
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = CDSResourceResolutionStep()
#!/usr/bin/env python
"""vIMS VM test case."""
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.instantiate.vf_module_ala_carte import \
YamlTemplateVfModuleAlaCarteInstantiateStep
-from onaptests.utils.exceptions import OnapTestException
class ClearwaterIms(ScenarioBase):
"""Onboard then instantiate a clearwater vIMS with ONAP."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init Clearwater IMS."""
# import clearwater_ims_nomulticloud_settings needed
super().__init__('clearwater_ims', **kwargs)
- self.test = YamlTemplateVfModuleAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run vIMS test."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.__logger.info("VNF clearwater IMS successfully created")
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("VNF clearwater IMS cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
- self.stop_time = time.time()
- except OnapTestException as exc:
- self.result = 0
- self.__logger.error(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.error("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- try:
- self.test.reports_collection.generate_report()
- except: # noqa
- self.__logger.error("Impossible to generate reporting")
+ self.test = YamlTemplateVfModuleAlaCarteInstantiateStep()
"""Instantiate basic vm using SO macro flow."""
-import logging
-import time
-
from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaPublishStep
-from onaptests.utils.exceptions import OnapTestException
from yaml import SafeLoader, load
-class MultiVnfUbuntuMacroStep(YamlTemplateBaseStep):
+class MultiVnfUbuntuMacroStep(YamlTemplateBaseScenarioStep):
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- CbaPublishStep
- YamlTemplateServiceAlaCarteInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self._yaml_template: dict = None
self._model_yaml_template: dict = None
- self.add_step(CbaPublishStep(
- cleanup=settings.CLEANUP_FLAG
- ))
- self.add_step(YamlTemplateServiceMacroInstantiateStep(
- cleanup=settings.CLEANUP_FLAG
- ))
+ self.add_step(CbaPublishStep())
+ self.add_step(YamlTemplateServiceMacroInstantiateStep())
@property
def description(self) -> str:
class MultiVnfUbuntuMacro(ScenarioBase):
"""Instantiate a basic vm macro."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init Basic Macro use case."""
super().__init__('nso_ubuntu_macro', **kwargs)
- self.test = MultiVnfUbuntuMacroStep(cleanup=settings.CLEANUP_FLAG)
-
- def run(self):
- """Run NSO Ubuntu macro test."""
- self.start_time = time.time()
- try:
- self.test.execute()
- self.__logger.info("Starting to clean up in {} seconds".format(
- settings.CLEANUP_ACTIVITY_TIMER))
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.error(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.error("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Generate report."""
- self.test.reports_collection.generate_report()
+ self.test = MultiVnfUbuntuMacroStep()
"""Instantiate service with PNF using SO macro flow."""
-import logging
-import time
-
from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
-from onaptests.scenario.scenario_base import ScenarioBase
-from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaEnrichStep
from onaptests.steps.simulator.pnf_simulator_cnf.pnf_register import \
PnfSimulatorCnfRegisterStep
-from onaptests.utils.exceptions import OnapTestException
from yaml import SafeLoader, load
-class PnfMacroScenarioStep(YamlTemplateBaseStep):
+class PnfMacroScenarioStep(YamlTemplateBaseScenarioStep):
"""Step created to run scenarion and generate report."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- YamlTemplateServiceAlaCarteInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self._yaml_template: dict = None
- self.add_step(PnfSimulatorCnfRegisterStep(
- cleanup=cleanup
- ))
- self.add_step(CbaEnrichStep(
- cleanup=cleanup
- ))
- self.add_step(YamlTemplateServiceMacroInstantiateStep(
- cleanup=cleanup
- ))
+ self.add_step(PnfSimulatorCnfRegisterStep())
+ self.add_step(CbaEnrichStep())
+ self.add_step(YamlTemplateServiceMacroInstantiateStep())
@property
def description(self) -> str:
str: Component name
"""
- return "PythonSDK-tests"
+ return "TEST"
@property
def yaml_template(self) -> dict:
class PnfMacro(ScenarioBase):
"""Run PNF simulator and onboard then instantiate a service with PNF."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init Basic Network use case."""
super().__init__('pnf_macro', **kwargs)
- self.test = PnfMacroScenarioStep(cleanup=settings.CLEANUP_FLAG)
-
- def run(self):
- """Run PNF macro test."""
- self.start_time = time.time()
- try:
- for test_phase in (self.test.execute, self.test.cleanup):
- try:
- test_phase()
- self.result += 50
- except OnapTestException as exc:
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Generate report."""
- self.test.reports_collection.generate_report()
+ self.test = PnfMacroScenarioStep()
import logging
+import time
from onapsdk.configuration import settings
+from onapsdk.exceptions import SDKException
+from onaptests.steps.base import BaseStep, YamlTemplateBaseStep
+from onaptests.utils.exceptions import OnapTestException
from xtesting.core import testcase
"""Init base scenario."""
if "case_name" not in kwargs:
kwargs["case_name"] = case_name_override
- self.__logger.info("Global Configuration:")
+ self.scenario_name = kwargs["case_name"].replace("_", " ")
+ self.scenario_name = str.title(self.scenario_name)
+
+ self.__logger.info("%s Global Configuration:", self.scenario_name)
for val in settings._settings:
self.__logger.info("%s: %s", val, settings._settings[val])
+
+ self.__logger.debug("%s init started", self.scenario_name)
super().__init__(**kwargs)
- self.__logger.debug("%s init started", kwargs["case_name"])
+ self.general_exception = None
+
+ def run(self):
+ """Run scenario and cleanup resources afterwards"""
+ self.start_time = time.time()
+ try:
+ for test_phase in (self.test.execute, self.test.cleanup):
+ phase_name = test_phase.__name__
+ try:
+ if (phase_name == "cleanup" and settings.CLEANUP_FLAG and
+ settings.CLEANUP_ACTIVITY_TIMER > 0):
+ time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
+ self.__logger.info("%s %s Phase Started",
+ self.scenario_name, phase_name.title())
+ test_phase()
+ self.result += 50
+ except OnapTestException as exc:
+ self.__logger.exception("%s on %s", exc.error_message, phase_name)
+ except SDKException:
+ self.__logger.exception("SDK Exception on %s", phase_name)
+ except Exception as e:
+ self.__logger.exception("General Exception on %s", phase_name)
+ if self.general_exception:
+ e = ExceptionGroup("General Exceptions", [self.general_exception, e]) # noqa
+ self.general_exception = e
+ finally:
+ self.stop_time = time.time()
+ self.__logger.info(f"{self.scenario_name} Execution {self.result}% Completed")
+ if self.general_exception:
+ raise self.general_exception
+
+ def clean(self):
+ """Clean Additional resources if needed."""
+ self.__logger.info("Generate %s Test report", self.scenario_name)
+ self.test.reports_collection.generate_report()
+
+ def validate_execution(self):
+ self.test.validate_execution()
+
+
+class BaseScenarioStep(BaseStep):
+
+ def __init__(self, cleanup=False):
+ """Initialize BaseScenarioStep step."""
+ super().__init__(cleanup=cleanup)
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ super().execute()
+
+
+class YamlTemplateBaseScenarioStep(YamlTemplateBaseStep, BaseScenarioStep):
+
+ def __init__(self, cleanup=False):
+ """Initialize YamlTemplateBaseScenarioStep step."""
+ super().__init__(cleanup=cleanup)
-import logging
-import time
-
-from onapsdk.configuration import settings
-from onapsdk.exceptions import SDKException
from onaptests.scenario.scenario_base import ScenarioBase
from onaptests.steps.cloud.check_status import CheckNamespaceStatusStep
-from onaptests.utils.exceptions import OnapTestException
class Status(ScenarioBase):
"""Retrieve status of Kubernetes resources in the nemaspace."""
- __logger = logging.getLogger(__name__)
-
def __init__(self, **kwargs):
"""Init the testcase."""
super().__init__('status', **kwargs)
- self.test = CheckNamespaceStatusStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Status check test."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- try:
- self.test.execute()
- self.__logger.info("Status check successfully completed")
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("Status check cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
- except OnapTestException as exc:
- self.result = 0
- self.__logger.exception(exc.error_message)
- except SDKException:
- self.result = 0
- self.__logger.exception("SDK Exception")
- finally:
- self.stop_time = time.time()
-
- def clean(self):
- """Clean Additional resources if needed."""
- self.__logger.info("Generate Test report")
- self.test.reports_collection.generate_report()
+ self.test = CheckNamespaceStatusStep()
import itertools
import logging
import logging.config
+import os
import time
-
from abc import ABC, abstractmethod
from typing import Iterator, List, Optional
+
from onapsdk.aai.business import Customer
from onapsdk.configuration import settings
from onapsdk.exceptions import SDKException, SettingsError
+from onaptests.steps.reports_collection import (Report, ReportsCollection,
+ ReportStepStatus)
+from onaptests.utils.exceptions import (OnapTestException,
+ OnapTestExceptionGroup,
+ SkipExecutionException,
+ SubstepExecutionException,
+ SubstepExecutionExceptionGroup)
+
+
+IF_VALIDATION = "PYTHON_SDK_TESTS_VALIDATION"
-from onaptests.steps.reports_collection import Report, ReportsCollection, ReportStepStatus
-from onaptests.utils.exceptions import OnapTestException, SubstepExecutionException
+class StoreStateHandler(ABC):
+
+ @classmethod
+ def store_state(cls, fun=None, *, cleanup=False):
+ if fun is None:
+ return functools.partial(cls.store_state, cleanup=cleanup)
-class BaseStep(ABC):
+ @functools.wraps(fun)
+ def wrapper(self, *args, **kwargs):
+ if (cleanup and self._state_clean) or (not cleanup and self._state_execute):
+ raise RuntimeError(f"Sate of {self._step_title(cleanup)} already stored")
+ if cleanup:
+ self._state_clean = True
+ else:
+ self._state_execute = True
+ initial_exception = None
+ try:
+ execution_status: Optional[ReportStepStatus] = ReportStepStatus.FAIL
+ if cleanup:
+ self._start_cleanup_time = time.time()
+ try:
+ if (self._executed and self._cleanup and
+ (self._is_validation_only or
+ self.check_preconditions(cleanup=True))):
+ self._log_execution_state("START", cleanup)
+ if not self._is_validation_only:
+ fun(self, *args, **kwargs)
+ self._cleaned_up = True
+ execution_status = ReportStepStatus.PASS
+ else:
+ execution_status = ReportStepStatus.NOT_EXECUTED
+ except (OnapTestException, SDKException) as test_exc:
+ initial_exception = test_exc
+ finally:
+ self._log_execution_state(execution_status.name, cleanup)
+ self._cleanup_substeps()
+ if initial_exception:
+ new_exception = initial_exception
+ initial_exception = None
+ raise new_exception
+ else:
+ if self._is_validation_only or self.check_preconditions():
+ self._log_execution_state("START", cleanup)
+ fun(self, *args, **kwargs)
+ execution_status = ReportStepStatus.PASS
+ self._executed = True
+ else:
+ execution_status = ReportStepStatus.NOT_EXECUTED
+ except SkipExecutionException:
+ execution_status = ReportStepStatus.PASS
+ self._executed = True
+ except SubstepExecutionException as substep_exc:
+ if not cleanup:
+ execution_status = ReportStepStatus.NOT_EXECUTED
+ if initial_exception:
+ substep_exc = OnapTestExceptionGroup("Cleanup Exceptions",
+ [initial_exception, substep_exc])
+ raise substep_exc
+ except (OnapTestException, SDKException) as test_exc:
+ if initial_exception:
+ test_exc = OnapTestExceptionGroup("Cleanup Exceptions",
+ [initial_exception, test_exc])
+ raise test_exc
+ finally:
+ if not cleanup:
+ self._log_execution_state(execution_status.name, cleanup)
+ if cleanup:
+ self._cleanup_report = Report(
+ step_description=self._step_title(cleanup),
+ step_execution_status=execution_status,
+ step_execution_duration=time.time() - self._start_cleanup_time,
+ step_component=self.component
+ )
+ else:
+ if not self._start_execution_time:
+ if execution_status != ReportStepStatus.NOT_EXECUTED:
+ self._logger.error("No execution start time saved for %s step. "
+ "Fix it by call `super.execute()` "
+ "in step class `execute()` method definition",
+ self.name)
+ self._start_execution_time = time.time()
+ self._execution_report = Report(
+ step_description=self._step_title(cleanup),
+ step_execution_status=(execution_status if execution_status else
+ ReportStepStatus.FAIL),
+ step_execution_duration=time.time() - self._start_execution_time,
+ step_component=self.component
+ )
+ return wrapper
+
+
+class BaseStep(StoreStateHandler, ABC):
"""Base step class."""
+ """Indicates that Step has no dedicated cleanup method"""
+ HAS_NO_CLEANUP = False
+
_logger: logging.Logger = logging.getLogger("")
def __init_subclass__(cls):
except SettingsError:
pass
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self, cleanup: bool = False, break_on_error=True) -> None:
"""Step initialization.
Args:
cleanup(bool, optional): Determines if cleanup action should be called.
+ break_on_error(bool, optional): Determines if fail on execution should
+ result with continuation of further steps
"""
self._steps: List["BaseStep"] = []
self._start_cleanup_time: float = None
self._execution_report: ReportStepStatus = None
self._cleanup_report: ReportStepStatus = None
+ self._executed: bool = False
+ self._cleaned_up: bool = False
+ self._state_execute: bool = False
+ self._state_clean: bool = False
+ self._nesting_level: int = 0
+ self._break_on_error: bool = break_on_error
+ self._is_validation_only = os.environ.get(IF_VALIDATION) is not None
def add_step(self, step: "BaseStep") -> None:
"""Add substep.
"""
self._steps.append(step)
step._parent: "BaseStep" = self
+ step._update_nesting_level()
+
+ def _update_nesting_level(self) -> None:
+ """Update step nesting level.
+
+ Step nesting level allows to display relatino of steps during validation
+ """
+ self._nesting_level = 1 + self._parent._nesting_level
+ for step in self._steps:
+ step._update_nesting_level()
@property
def parent(self) -> "BaseStep":
"""
return self._parent
+ @property
+ def is_executed(self) -> bool:
+ """Is step executed.
+
+ Step is executed if execute() method was completed without errors
+
+ Returns:
+ bool: True if step is executed, False otherwise
+
+ """
+ return self._executed
+
@property
def is_root(self) -> bool:
"""Is a root step.
if self._cleanup:
if self._cleanup_report:
yield self._cleanup_report
- for step in self._steps:
- yield from step.cleanup_reports
+ for step in reversed(self._steps):
+ yield from step.cleanup_reports
@property
def name(self) -> str:
"""
- @classmethod
- def store_state(cls, fun=None, *, cleanup=False):
- if fun is None:
- return functools.partial(cls.store_state, cleanup=cleanup)
+ def _step_title(self, cleanup=False):
+ cleanup_label = " Cleanup:" if cleanup else ":"
+ return f"[{self.component}] {self.name}{cleanup_label} {self.description}"
- @functools.wraps(fun)
- def wrapper(self, *args, **kwargs):
- try:
- if cleanup:
- self._start_cleanup_time = time.time()
- self._logger.info("*****************************************************")
- self._logger.info(f"START [{self.component}] {self.name} cleanup: "
- f"{self.description}")
- self._logger.info("*****************************************************")
- else:
- self._logger.info("*****************************************************")
- self._logger.info(f"START [{self.component}] {self.name}: {self.description}")
- self._logger.info("*****************************************************")
- execution_status: Optional[ReportStepStatus] = None
- ret = fun(self, *args, **kwargs)
- execution_status = ReportStepStatus.PASS
- return ret
- except SubstepExecutionException:
- execution_status = (ReportStepStatus.PASS if cleanup else
- ReportStepStatus.NOT_EXECUTED)
- raise
- except (OnapTestException, SDKException):
- execution_status = ReportStepStatus.FAIL
- raise
- finally:
- if cleanup:
- self._logger.info("*****************************************************")
- self._logger.info(f"STOP [{self.component}] {self.name} cleanup: "
- f"{self.description}")
- self._logger.info("*****************************************************")
- self._cleanup_report = Report(
- step_description=(f"[{self.component}] {self.name} cleanup: "
- f"{self.description}"),
- step_execution_status=execution_status,
- step_execution_duration=time.time() - self._start_cleanup_time,
- step_component=self.component
- )
- else:
- self._logger.info("*****************************************************")
- self._logger.info(f"STOP [{self.component}] {self.name}: {self.description}")
- self._logger.info("*****************************************************")
- if not self._start_execution_time:
- if execution_status != ReportStepStatus.NOT_EXECUTED:
- self._logger.error("No execution start time saved for %s step. "
- "Fix it by call `super.execute()` "
- "in step class `execute()` method definition",
- self.name)
- self._start_execution_time = time.time()
- self._execution_report = Report(
- step_description=f"[{self.component}] {self.name}: {self.description}",
- step_execution_status=(execution_status if execution_status else
- ReportStepStatus.FAIL),
- step_execution_duration=time.time() - self._start_execution_time,
- step_component=self.component
- )
- return wrapper
+ def _log_execution_state(self, state: str, cleanup=False):
+ nesting_label = "" + " " * self._nesting_level
+ description = f"| {state} {self._step_title(cleanup)} |"
+ self._logger.info(nesting_label + "*" * len(description))
+ self._logger.info(nesting_label + description)
+ self._logger.info(nesting_label + "*" * len(description))
+
+ def check_preconditions(self, cleanup=False) -> bool:
+ """Check preconditions.
+
+ Check if step preconditions are satisfied. If not, step is skipped
+ without further consequences. If yes, execution is initiated
+
+ Returns:
+ bool: True if preconditions are satisfied, False otherwise
+
+ """
+ return True
def execute(self) -> None:
"""Step's action execution.
Override this method and remember to call `super().execute()` before.
"""
+ substep_error = False
for step in self._steps:
try:
step.execute()
except (OnapTestException, SDKException) as substep_err:
- raise SubstepExecutionException from substep_err
+ substep_error = True
+ if step._break_on_error:
+ raise SubstepExecutionException from substep_err
+ else:
+ self._logger.exception(substep_err)
if self._steps:
- self._logger.info("*****************************************************")
- self._logger.info(f"CONTINUE [{self.component}] {self.name}: {self.description}")
- self._logger.info("*****************************************************")
+ if substep_error and self._break_on_error:
+ raise SubstepExecutionException("Cannot continue due to failed substeps")
+ self._log_execution_state("CONTINUE")
self._start_execution_time = time.time()
+ if self._is_validation_only:
+ raise SkipExecutionException()
+
+ def _cleanup_substeps(self) -> None:
+ """Substeps' cleanup.
+
+ Substeps are cleaned-up in reversed order.
+ We also try to cleanup steps if others failed
+
+ """
+ exceptions_to_raise = []
+ for step in reversed(self._steps):
+ try:
+ if step._cleanup:
+ step.cleanup()
+ else:
+ step._default_cleanup_handler()
+ except (OnapTestException, SDKException) as substep_err:
+ try:
+ raise SubstepExecutionException from substep_err
+ except Exception as e:
+ exceptions_to_raise.append(e)
+ if len(exceptions_to_raise) > 0:
+ if len(exceptions_to_raise) == 1:
+ raise exceptions_to_raise[0]
+ else:
+ raise SubstepExecutionExceptionGroup("Substep Exceptions", exceptions_to_raise)
def cleanup(self) -> None:
"""Step's cleanup.
Not all steps has to have cleanup method
"""
- if self._cleanup:
- for step in self._steps:
- try:
- step.cleanup()
- except (OnapTestException, SDKException) as substep_err:
- raise SubstepExecutionException from substep_err
+ # Step itself was cleaned-up, now time for children
+ if not self._cleanup:
+ # in this case we just make sure that store_state is run
+ self._default_cleanup_handler()
+
+ @StoreStateHandler.store_state(cleanup=True)
+ def _default_cleanup_handler(self):
+ pass
@classmethod
def set_proxy(cls, sock_http):
onap_proxy['https'] = sock_http
Customer.set_proxy(onap_proxy)
+ def validate_execution(self):
+ if self._is_validation_only:
+ self._log_execution_state(f"VALIDATE [{self._executed}, {self._cleanup}]")
+ if self._executed and self._cleanup and not self._cleaned_up:
+ self._logger.error(f"{self._step_title()} Cleanup Not Executed")
+ assert self._cleaned_up
+ for step in reversed(self._steps):
+ step.validate_execution()
+
class YamlTemplateBaseStep(BaseStep, ABC):
"""Base YAML template step."""
__logger = logging.getLogger(__name__)
- def __init__(self, resource_type: str, **kwargs):
+ def __init__(self, resource_type: str, break_on_error=False):
"""Init CheckK8sResourcesStep."""
- super().__init__(cleanup=False)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP, break_on_error=break_on_error)
self.core = client.CoreV1Api()
self.batch = client.BatchV1Api()
self.app = client.AppsV1Api()
return f"Check status of all k8s {self.resource_type}s in the {NAMESPACE} namespace."
def _init_resources(self):
- self.__logger.debug(f"Loading all k8s {self.resource_type}s in the {NAMESPACE} namespace")
+ if self.resource_type != "":
+ self.__logger.debug(f"Loading all k8s {self.resource_type}s"
+ " in the {NAMESPACE} namespace")
def _parse_resources(self):
"""Parse the resources."""
len(self.all_resources),
self.resource_type,
len(self.failing_resources))
- except (ConnectionRefusedError, MaxRetryError, NewConnectionError):
+ if self.failing:
+ raise StatusCheckException(f"{self.resource_type} test failed")
+ except (ConnectionRefusedError, MaxRetryError, NewConnectionError) as e:
self.__logger.error("Test of k8s %ss failed.", self.resource_type)
self.__logger.error("Cannot connect to Kubernetes.")
+ raise StatusCheckException from e
class CheckBasicK8sResourcesStep(CheckK8sResourcesStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, resource_type: str, k8s_res_class, cleanup: bool = False, **kwargs):
+ def __init__(self, resource_type: str, k8s_res_class):
"""Init CheckBasicK8sResourcesStep."""
super().__init__(resource_type)
self.k8s_res_class = k8s_res_class
class CheckK8sConfigMapsStep(CheckBasicK8sResourcesStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, cleanup: bool = False, **kwargs):
+ def __init__(self):
"""Init CheckK8sConfigMapsStep."""
super().__init__("configmap", ConfigMap)
class CheckK8sSecretsStep(CheckBasicK8sResourcesStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, cleanup: bool = False, **kwargs):
+ def __init__(self):
"""Init CheckK8sSecretsStep."""
super().__init__("secret", Secret)
class CheckK8sIngressesStep(CheckBasicK8sResourcesStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, cleanup: bool = False, **kwargs):
+ def __init__(self):
"""Init CheckK8sIngressesStep."""
super().__init__("ingress", Ingress)
class CheckK8sPvcsStep(CheckK8sResourcesStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, cleanup: bool = False, **kwargs):
+ def __init__(self):
"""Init CheckK8sPvcsStep."""
super().__init__("pvc")
class CheckK8sResourcesUsingPodsStep(CheckK8sResourcesStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, resource_type: str, pods_source, cleanup: bool = False, **kwargs):
+ def __init__(self, resource_type: str, pods_source):
"""Init CheckK8sResourcesUsingPodsStep."""
super().__init__(resource_type)
self.pods_source = pods_source
class CheckK8sJobsStep(CheckK8sResourcesUsingPodsStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, cleanup: bool = False, **kwargs):
+ def __init__(self):
"""Init CheckK8sJobsStep."""
super().__init__("job", None)
__logger = logging.getLogger(__name__)
- def __init__(self, pods, cleanup: bool = False, **kwargs):
+ def __init__(self, pods):
"""Init CheckK8sPodsStep."""
super().__init__("pod", pods)
class CheckK8sServicesStep(CheckK8sResourcesUsingPodsStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, pods, cleanup: bool = False, **kwargs):
+ def __init__(self, pods):
"""Init CheckK8sServicesStep."""
super().__init__("service", pods)
class CheckK8sDeploymentsStep(CheckK8sResourcesUsingPodsStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, pods, cleanup: bool = False, **kwargs):
+ def __init__(self, pods):
"""Init CheckK8sDeploymentsStep."""
super().__init__("deployment", pods)
self.all_resources.append(deployment)
-class CheckK8sResplicaSetsStep(CheckK8sResourcesUsingPodsStep):
-
- __logger = logging.getLogger(__name__)
+class CheckK8sReplicaSetsStep(CheckK8sResourcesUsingPodsStep):
- def __init__(self, pods, cleanup: bool = False, **kwargs):
- """Init CheckK8sResplicaSetsStep."""
+ def __init__(self, pods):
+ """Init CheckK8sReplicaSetsStep."""
super().__init__("replicaset", pods)
def _init_resources(self):
class CheckK8sStatefulSetsStep(CheckK8sResourcesUsingPodsStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, pods, cleanup: bool = False, **kwargs):
+ def __init__(self, pods):
"""Init CheckK8sStatefulSetsStep."""
super().__init__("statefulset", pods)
class CheckK8sDaemonSetsStep(CheckK8sResourcesUsingPodsStep):
- __logger = logging.getLogger(__name__)
-
- def __init__(self, pods, cleanup: bool = False, **kwargs):
+ def __init__(self, pods):
"""Init CheckK8sDaemonSetsStep."""
super().__init__("daemonset", pods)
__logger = logging.getLogger(__name__)
- def __init__(self, cleanup: bool = False, **kwargs):
+ def __init__(self):
"""Init CheckNamespaceStatusStep."""
- super().__init__("")
+ super().__init__(resource_type="")
self.__logger.debug("%s namespace status test init started", NAMESPACE)
if settings.IN_CLUSTER:
config.load_incluster_config()
self.pod_list_step = CheckK8sPodsStep(self.job_list_step)
self.service_list_step = CheckK8sServicesStep(self.pod_list_step)
self.deployment_list_step = CheckK8sDeploymentsStep(self.pod_list_step)
- self.replicaset_list_step = CheckK8sResplicaSetsStep(self.pod_list_step)
+ self.replicaset_list_step = CheckK8sReplicaSetsStep(self.pod_list_step)
self.statefulset_list_step = CheckK8sStatefulSetsStep(self.pod_list_step)
self.daemonset_list_step = CheckK8sDaemonSetsStep(self.pod_list_step)
self.configmap_list_step = CheckK8sConfigMapsStep()
step.resource_type,
len(step.failing_resources))
details[step.resource_type] = {
- 'number_all': len(step.all_resources),
'number_failing': len(step.failing_resources),
- 'all': self.map_by_name(step.all_resources),
'failing': self.map_by_name(step.failing_resources)
}
+ if settings.INCLUDE_ALL_RES_IN_DETAILS:
+ details[step.resource_type]['all'] = self.map_by_name(step.all_resources)
+ details[step.resource_type]['number_all'] = len(step.all_resources)
+
with (Path(self.res_dir).joinpath(settings.STATUS_DETAILS_JSON)).open('w') as file:
json.dump(details, file, indent=4)
if self.failing:
class CloudRegionCreateStep(BaseStep):
"""Cloud region creation step."""
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+
@property
def description(self) -> str:
"""Step description."""
class ComplexCreateStep(BaseStep):
"""Complex creation step."""
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+
@property
def description(self) -> str:
"""Step description."""
class ConnectServiceSubToCloudRegionStep(BaseStep):
"""Connect service subscription to cloud region step."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- CustomerServiceSubscriptionCreateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
if settings.CLOUD_REGION_TYPE == settings.K8S_REGION_TYPE:
- self.add_step(K8SConnectivityInfoStep(cleanup=cleanup))
- self.add_step(RegisterCloudRegionStep(cleanup=cleanup))
- self.add_step(LinkCloudRegionToComplexStep(cleanup=cleanup))
- self.add_step(CustomerServiceSubscriptionCreateStep(cleanup=cleanup))
+ self.add_step(K8SConnectivityInfoStep())
+ self.add_step(RegisterCloudRegionStep())
+ self.add_step(LinkCloudRegionToComplexStep())
+ self.add_step(CustomerServiceSubscriptionCreateStep())
@property
def description(self) -> str:
class CustomerCreateStep(BaseStep):
"""Customer creation step."""
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+
@property
def description(self) -> str:
"""Step description."""
class CustomerServiceSubscriptionCreateStep(BaseStep):
"""Cutomer service subsription creation step"""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- CustomerCreateStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CustomerCreateStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CustomerCreateStep())
@property
def description(self) -> str:
class K8SConnectivityInfoStep(BaseStep):
"""CreateConnnectivityInfoStep."""
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+
@property
def description(self) -> str:
"""Step description."""
class LinkCloudRegionToComplexStep(BaseStep):
"""Link cloud region to complex step"""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- ComplexCreateStep,
- CloudRegionCreateStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(ComplexCreateStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(ComplexCreateStep())
@property
def description(self) -> str:
class RegisterCloudRegionStep(BaseStep):
"""Cloud region registration step."""
- def __init__(self, cleanup: bool) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CloudRegionCreateStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CloudRegionCreateStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CloudRegionCreateStep())
@property
def description(self) -> str:
class K8SProfileStep(BaseStep):
"""CreateK8sProfileStep."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
self._service_instance: ServiceInstance = None
- self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(cleanup))
+ self.add_step(YamlTemplateVnfAlaCarteInstantiateStep())
@property
def description(self) -> str:
class CreateInstanceStep(BaseStep):
"""Create MSB k8s instance step."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateProfileStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CreateProfileStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(CreateProfileStep())
self.instance: Instance = None
@property
from onapsdk.sdnc import VfModulePreload
from onapsdk.sdnc.preload import PreloadInformation
from onapsdk.sdnc.services import Service
+from onaptests.scenario.scenario_base import BaseScenarioStep
+from onaptests.steps.base import BaseStep
from onaptests.utils.exceptions import OnapTestException
-from ..base import BaseStep
-
class BaseSdncStep(BaseStep):
"""Basic SDNC step."""
class ServiceCreateStep(BaseSdncStep):
"""Service creation step."""
- def __init__(self, service: Service = None, cleanup: bool = False):
+ def __init__(self, service: Service = None):
"""Initialize step."""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self.service = service
@property
return "Create SDNC service."
@BaseStep.store_state
- def execute(self):
+ def execute(self) -> None:
"""Create service at SDNC."""
super().execute()
self._logger.info("Create new service instance in SDNC by GR-API")
else:
raise OnapTestException("SDNC service creation failed.")
- @BaseStep.store_state()
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup Service."""
if self.service is not None:
The step needs in an existing SDNC service as a prerequisite.
"""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Sub steps:
- ServiceCreateStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(ServiceCreateStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(ServiceCreateStep())
@property
def description(self) -> str:
"""
return "Update SDNC service"
- @BaseStep.store_state
- def execute(self):
+ @BaseSdncStep.store_state
+ def execute(self) -> None:
super().execute()
self._logger.info("Get existing SDNC service instance and update it over GR-API")
try:
Upload preload information for VfModule over GR-API.
"""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step."""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
@property
def description(self) -> str:
"""
return "Upload Preload information for VfModule"
- @BaseStep.store_state
- def execute(self):
+ @BaseSdncStep.store_state
+ def execute(self) -> None:
super().execute()
self._logger.info("Upload VfModule preload information over GR-API")
VfModulePreload.upload_vf_module_preload(
Get preload information from SDNC over GR-API.
"""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Sub steps:
- UploadVfModulePreloadStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(UploadVfModulePreloadStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(UploadVfModulePreloadStep())
@property
def description(self) -> str:
"""
return "Get Preload information"
- @BaseStep.store_state
- def execute(self):
+ @BaseSdncStep.store_state
+ def execute(self) -> None:
super().execute()
self._logger.info("Get existing SDNC service instance and update it over GR-API")
preloads = PreloadInformation.get_all()
print(preload_information)
-class TestSdncStep(BaseStep):
+class TestSdncStep(BaseScenarioStep):
"""Top level step for SDNC tests."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Sub steps:
- UpdateSdncService.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(
- UpdateSdncService(cleanup=cleanup)
- )
- self.add_step(
- GetSdncPreloadStep(cleanup=cleanup)
- )
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(UpdateSdncService())
+ self.add_step(GetSdncPreloadStep())
@property
def description(self) -> str:
str: Component name
"""
- return "PythonSDK-tests"
+ return "TEST"
class YamlTemplateServiceAlaCarteInstantiateStep(YamlTemplateBaseStep):
"""Instantiate service a'la carte using YAML template."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- YamlTemplateServiceOnboardStep,
- ConnectServiceSubToCloudRegionStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
self._service_instance: str = None
if not settings.ONLY_INSTANTIATE:
- self.add_step(YamlTemplateServiceOnboardStep(cleanup))
- self.add_step(ConnectServiceSubToCloudRegionStep(cleanup))
+ self.add_step(YamlTemplateServiceOnboardStep())
+ self.add_step(ConnectServiceSubToCloudRegionStep())
@property
def description(self) -> str:
class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep):
"""Instantiate service a'la carte using YAML template."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- ConnectServiceSubToCloudRegionStep,
- CustomerServiceSubscriptionCreateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._model_yaml_template: dict = None
self._service_instance_name: str = None
self._service_instance: str = None
if not settings.ONLY_INSTANTIATE:
- self.add_step(YamlTemplateServiceOnboardStep(cleanup))
+ self.add_step(YamlTemplateServiceOnboardStep())
if any(
filter(lambda x: x in self.yaml_template[self.service_name].keys(),
["vnfs", "networks"])):
# can additionally contain "pnfs", no difference
- self.add_step(ConnectServiceSubToCloudRegionStep(cleanup))
+ self.add_step(ConnectServiceSubToCloudRegionStep())
else: # only pnfs
- self.add_step(CustomerServiceSubscriptionCreateStep(cleanup))
+ self.add_step(CustomerServiceSubscriptionCreateStep())
@property
def description(self) -> str:
class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep):
"""Instantiate vf module a'la carte using YAML template."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- YamlTemplateVnfAlaCarteInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
if settings.CLOUD_REGION_TYPE == settings.K8S_REGION_TYPE:
# K8SProfileStep creates the requested profile and then calls
# YamlTemplateVnfAlaCarteInstantiateStep step
- self.add_step(K8SProfileStep(cleanup))
+ self.add_step(K8SProfileStep())
else:
- self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(cleanup))
+ self.add_step(YamlTemplateVnfAlaCarteInstantiateStep())
@property
def description(self) -> str:
class YamlTemplateVlAlaCarteInstantiateStep(YamlTemplateBaseStep):
"""Instantiate vl a'la carte using YAML template."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- YamlTemplateServiceAlaCarteInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
self._service_instance: ServiceInstance = None
- self.add_step(YamlTemplateServiceAlaCarteInstantiateStep(cleanup))
+ self.add_step(YamlTemplateServiceAlaCarteInstantiateStep())
@property
def description(self) -> str:
class YamlTemplateVnfAlaCarteInstantiateStep(YamlTemplateBaseStep):
"""Instantiate vnf a'la carte using YAML template."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- YamlTemplateServiceAlaCarteInstantiateStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._service_instance_name: str = None
self._service_instance: ServiceInstance = None
- self.add_step(YamlTemplateServiceAlaCarteInstantiateStep(cleanup))
+ self.add_step(YamlTemplateServiceAlaCarteInstantiateStep())
@property
def description(self) -> str:
count: int = 0
- def __init__(self, cleanup=False):
- super().__init__(cleanup=cleanup)
+ def __init__(self):
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
- self.add_step(OnboardClampStep(cleanup=cleanup))
+ self.add_step(OnboardClampStep())
Clamp()
self.loop_instance = None
import urllib3
from onaptests.steps.base import BaseStep
-
from onaptests.utils.exceptions import OnapTestException
class ExposeCDSBlueprintprocessorNodePortStep(CDSBaseStep):
"""Expose CDS blueprintsprocessor port."""
- def __init__(self, cleanup: bool) -> None:
+ def __init__(self) -> None:
"""Initialize step."""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self.service_name: str = "cds-blueprints-processor-http"
if settings.IN_CLUSTER:
config.load_incluster_config()
else:
self._logger.debug("Service already patched, skip")
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Step cleanup.
class BootstrapBlueprintprocessor(CDSBaseStep):
"""Bootstrap blueprintsprocessor."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- ExposeCDSBlueprintprocessorNodePortStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
if settings.EXPOSE_SERVICES_NODE_PORTS:
- self.add_step(ExposeCDSBlueprintprocessorNodePortStep(cleanup=cleanup))
+ self.add_step(ExposeCDSBlueprintprocessorNodePortStep())
@property
def description(self) -> str:
class DataDictionaryUploadStep(CDSBaseStep):
"""Upload data dictionaries to CDS step."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize data dictionary upload step."""
- super().__init__(cleanup=cleanup)
- self.add_step(BootstrapBlueprintprocessor(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(BootstrapBlueprintprocessor())
@property
def description(self) -> str:
class CbaEnrichStep(CDSBaseStep):
"""Enrich CBA file step."""
- def __init__(self, cleanup=False) -> None:
+ def __init__(self) -> None:
"""Initialize CBA enrichment step."""
- super().__init__(cleanup=cleanup)
- self.add_step(DataDictionaryUploadStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(DataDictionaryUploadStep())
@property
def description(self) -> str:
class CbaPublishStep(CDSBaseStep):
"""Publish CBA file step."""
- def __init__(self, cleanup=False) -> None:
+ def __init__(self) -> None:
"""Initialize CBA publish step."""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
"""Let's skip enrichment if enriched CBA is already present"""
if Path.is_file(settings.CDS_CBA_UNENRICHED):
- self.add_step(CbaEnrichStep(cleanup=cleanup))
+ self.add_step(CbaEnrichStep())
elif settings.EXPOSE_SERVICES_NODE_PORTS:
- self.add_step(ExposeCDSBlueprintprocessorNodePortStep(cleanup=cleanup))
+ self.add_step(ExposeCDSBlueprintprocessorNodePortStep())
@property
def description(self) -> str:
class CbaProcessStep(CDSBaseStep):
"""Process CBA step."""
- def __init__(self, cleanup=False) -> None:
+ def __init__(self) -> None:
"""Initialize CBA process step."""
- super().__init__(cleanup=cleanup)
- self.add_step(CbaPublishStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CbaPublishStep())
@property
def description(self) -> str:
from onapsdk.configuration import settings
-from ..base import YamlTemplateBaseStep
+from ..base import BaseStep, YamlTemplateBaseStep
from .service import YamlTemplateVfOnboardStep
class OnboardClampStep(YamlTemplateBaseStep):
"""Onboard class to create CLAMP templates."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize Clamp Onboard object."""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self._yaml_template: dict = None
- self.add_step(YamlTemplateVfOnboardStep(cleanup=cleanup))
+ self.add_step(YamlTemplateVfOnboardStep())
# if "service_name" in kwargs:
# self.service_name = kwargs['service_name']
# else:
class CreateCpsDataspaceStep(CpsBaseStep):
"""Step to create a dataspace."""
+ def __init__(self) -> None:
+ """Initialize step."""
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+
@property
def description(self) -> str:
"""Step description."""
class CreateCpsSchemaSetStep(CpsBaseStep):
"""Step to check schema-set creation."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateCpsDataspaceStep.
"""
- super().__init__(cleanup)
- self.add_step(CreateCpsDataspaceStep(cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(CreateCpsDataspaceStep())
@property
def description(self) -> str:
class CreateCpsAnchorStep(CpsBaseStep):
"""Step to create an anchor."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateCpsSchemaSetStep.
"""
- super().__init__(cleanup)
- self.add_step(CreateCpsSchemaSetStep(cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(CreateCpsSchemaSetStep())
@property
def description(self) -> str:
class CreateCpsAnchorNodeStep(CpsBaseStep):
"""Step to check node on anchor creation."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateCpsAnchorStep.
"""
- super().__init__(cleanup)
- self.add_step(CreateCpsAnchorStep(cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(CreateCpsAnchorStep())
@property
def description(self) -> str:
class CreateDefinitionStep(BaseStep):
"""Create definition step initialization."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CloudRegionCreateStep,
- K8SConnectivityInfoStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CloudRegionCreateStep(cleanup=cleanup))
- self.add_step(K8SConnectivityInfoStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CloudRegionCreateStep())
+ self.add_step(K8SConnectivityInfoStep())
self.definition: Definition = None
@property
class CreateProfileStep(BaseStep):
"""Create profile step."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateDefinitionStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CreateDefinitionStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CreateDefinitionStep())
self.profile: Profile = None
@property
class PnfOnboardStep(BaseStep):
"""PNF onboard step."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Step initialization.
Substeps:
cleanup(bool, optional): Determines if cleanup action should be called.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(VspOnboardStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(VspOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@BaseStep.store_state
def execute(self) -> None:
"""Onboard PNF in SDC.
class YamlTemplatePnfOnboardStep(YamlTemplateBaseStep):
"""PNF onboard using YAML template step."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Step initialization.
Substeps:
cleanup(bool, optional): Determines if cleanup action should be called.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(YamlTemplateVspOnboardStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(YamlTemplateVspOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@property
def yaml_template(self) -> dict:
"""YAML template.
class ServiceOnboardStep(BaseStep):
"""Service onboard step."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- VfOnboardStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
if settings.VF_NAME != "":
- self.add_step(VfOnboardStep(cleanup=cleanup))
+ self.add_step(VfOnboardStep())
if settings.PNF_NAME != "":
- self.add_step(PnfOnboardStep(cleanup=cleanup))
+ self.add_step(PnfOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@BaseStep.store_state
def execute(self):
"""Onboard service.
class YamlTemplateServiceOnboardStep(YamlTemplateBaseStep):
"""Service onboard using YAML template step."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- YamlTemplateVfOnboardStep.
"""
- super().__init__(cleanup=cleanup)
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
self._yaml_template: dict = None
self._model_yaml_template: dict = None
if "vnfs" in self.yaml_template[self.service_name]:
- self.add_step(YamlTemplateVfOnboardStep(cleanup=cleanup))
+ self.add_step(YamlTemplateVfOnboardStep())
if "pnfs" in self.yaml_template[self.service_name]:
- self.add_step(YamlTemplatePnfOnboardStep(cleanup=cleanup))
+ self.add_step(YamlTemplatePnfOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@property
def yaml_template(self) -> dict:
"""Step YAML template.
if self.is_root:
if not self._model_yaml_template:
with open(settings.MODEL_YAML_TEMPLATE, "r") as model_yaml_template:
- self._model_yaml_template: dict = load(model_yaml_template)
+ self._model_yaml_template: dict = load(model_yaml_template, SafeLoader)
return self._model_yaml_template
return self.parent.model_yaml_template
@YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup service onboard step."""
- if settings.SDC_CLEANUP:
- service: Service = Service(name=self.service_name)
- if service.exists():
- service.archive()
- service.delete()
- super().cleanup()
+ service: Service = Service(name=self.service_name)
+ if service.exists():
+ service.archive()
+ service.delete()
+ super().cleanup()
class VendorOnboardStep(BaseStep):
"""Vendor onboard step."""
+ def __init__(self):
+ """Initialize step."""
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+
@property
def description(self) -> str:
"""Step description."""
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@BaseStep.store_state
def execute(self):
"""Onboard vendor.
class VfOnboardStep(BaseStep):
"""Vf onboard step."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- VspOnboardStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(VspOnboardStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(VspOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@BaseStep.store_state
def execute(self):
"""Onboard Vf.
class YamlTemplateVfOnboardStep(YamlTemplateBaseStep):
"""Vf onboard using YAML template step."""
- def __init__(self, cleanup=False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- YamlTemplateVspOnboardStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(YamlTemplateVspOnboardStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(YamlTemplateVspOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@property
def yaml_template(self) -> dict:
"""YAML template.
class VspOnboardStep(BaseStep):
"""Vsp onboard step."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- VendorOnboardStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(VendorOnboardStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(VendorOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@BaseStep.store_state
def execute(self):
"""Onboard Vsp.
class YamlTemplateVspOnboardStep(YamlTemplateBaseStep):
"""Vsp onboard using YAML template step."""
- def __init__(self, cleanup=False):
+ def __init__(self):
"""Initialize step.
Substeps:
- VendorOnboardStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(VendorOnboardStep(cleanup=cleanup))
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.add_step(VendorOnboardStep())
@property
def description(self) -> str:
"""Component name."""
return "SDC"
+ def check_preconditions(self, cleanup=False) -> bool:
+ if not super().check_preconditions(cleanup):
+ return False
+ if cleanup:
+ return settings.SDC_CLEANUP
+ return True
+
@property
def yaml_template(self) -> dict:
"""YAML template.
class CdsMockserverCnfConfigureStep(BaseStep):
"""Configure cds mockserver expectations."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateInstanceStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CreateInstanceStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CreateInstanceStep())
@property
def description(self) -> str:
class PnfSimulatorCnfRegisterStep(BaseStep):
"""PNF simulator registration step."""
- def __init__(self, cleanup: bool = False) -> None:
+ def __init__(self) -> None:
"""Initialize step.
Substeps:
- CreateInstanceStep.
"""
- super().__init__(cleanup=cleanup)
- self.add_step(CreateInstanceStep(cleanup=cleanup))
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ self.add_step(CreateInstanceStep())
@property
def description(self) -> str:
error_message = 'Generic OnapTest exception'
+class OnapTestExceptionGroup(ExceptionGroup, OnapTestException): # noqa
+ """Group of Onap Test Exceptions."""
+ error_message = 'Generic OnapTest exception group'
+
+
+class SkipExecutionException(OnapTestException):
+ """Used only for validation purposes"""
+
+
class TestConfigurationException(OnapTestException):
"""Raise when configuration of the use case is incomplete or buggy."""
error_message = 'Configuration error'
"""Exception raised if substep execution fails."""
+class SubstepExecutionExceptionGroup(ExceptionGroup, SubstepExecutionException): # noqa
+ """Group of Substep Exceptions."""
+
+
class EnvironmentCleanupException(OnapTestException):
"""Test environment cleanup exception."""
error_message = "Test couldn't finish a cleanup"
Code Review / testsuite / pythonsdk-tests.git / commitdiff