aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/scenario
diff options
context:
space:
mode:
authorMicha? Jagie??o <michal.jagiello@t-mobile.pl>2023-08-07 13:06:39 +0000
committerGerrit Code Review <gerrit@onap.org>2023-08-07 13:06:39 +0000
commitf79b8e825e0e3cbde157e29a3225f9357e4198bd (patch)
tree72494a6d436ce591b54cf621ccec4c68f8154ed7 /src/onaptests/scenario
parentbe6ea937c20fb6ba83d6c7b35f6f5677f9a355cc (diff)
parent4bccbde3060931c8fcf61fbf8b61db4a85b3200e (diff)
Merge "Change cleanup process of tests"
Diffstat (limited to 'src/onaptests/scenario')
-rw-r--r--src/onaptests/scenario/basic_cds.py13
-rw-r--r--src/onaptests/scenario/basic_clamp.py43
-rw-r--r--src/onaptests/scenario/basic_cnf.py44
-rw-r--r--src/onaptests/scenario/basic_cnf_macro.py48
-rw-r--r--src/onaptests/scenario/basic_cps.py36
-rw-r--r--src/onaptests/scenario/basic_network.py47
-rw-r--r--src/onaptests/scenario/basic_onboard.py37
-rw-r--r--src/onaptests/scenario/basic_sdnc.py32
-rw-r--r--src/onaptests/scenario/basic_vm.py44
-rw-r--r--src/onaptests/scenario/basic_vm_macro.py46
-rw-r--r--src/onaptests/scenario/basic_vm_macro_stability.py30
-rw-r--r--src/onaptests/scenario/cds_blueprint_enrichment.py48
-rw-r--r--src/onaptests/scenario/cds_resource_resolution.py54
-rw-r--r--src/onaptests/scenario/clearwater_ims.py47
-rw-r--r--src/onaptests/scenario/multi_vnf_macro.py49
-rw-r--r--src/onaptests/scenario/pnf_macro.py51
-rw-r--r--src/onaptests/scenario/scenario_base.py68
-rw-r--r--src/onaptests/scenario/status.py44
18 files changed, 130 insertions, 651 deletions
diff --git a/src/onaptests/scenario/basic_cds.py b/src/onaptests/scenario/basic_cds.py
new file mode 100644
index 0000000..9cb4a54
--- /dev/null
+++ b/src/onaptests/scenario/basic_cds.py
@@ -0,0 +1,13 @@
+#!/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()
diff --git a/src/onaptests/scenario/basic_clamp.py b/src/onaptests/scenario/basic_clamp.py
index 2e24cac..e04e372 100644
--- a/src/onaptests/scenario/basic_clamp.py
+++ b/src/onaptests/scenario/basic_clamp.py
@@ -1,53 +1,12 @@
"""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()
diff --git a/src/onaptests/scenario/basic_cnf.py b/src/onaptests/scenario/basic_cnf.py
index 51a5cc5..5641f1f 100644
--- a/src/onaptests/scenario/basic_cnf.py
+++ b/src/onaptests/scenario/basic_cnf.py
@@ -1,56 +1,14 @@
#!/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()
diff --git a/src/onaptests/scenario/basic_cnf_macro.py b/src/onaptests/scenario/basic_cnf_macro.py
index 18435ca..c96ce4f 100644
--- a/src/onaptests/scenario/basic_cnf_macro.py
+++ b/src/onaptests/scenario/basic_cnf_macro.py
@@ -1,35 +1,25 @@
"""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:
@@ -54,7 +44,7 @@ class BasicCnfMacroStep(YamlTemplateBaseStep):
str: Component name
"""
- return "PythonSDK-tests"
+ return "TEST"
@property
def yaml_template(self) -> dict:
@@ -101,29 +91,7 @@ class BasicCnfMacroStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/scenario/basic_cps.py b/src/onaptests/scenario/basic_cps.py
index a1c7316..13a3d37 100644
--- a/src/onaptests/scenario/basic_cps.py
+++ b/src/onaptests/scenario/basic_cps.py
@@ -1,13 +1,7 @@
#!/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):
@@ -21,35 +15,7 @@ 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()
diff --git a/src/onaptests/scenario/basic_network.py b/src/onaptests/scenario/basic_network.py
index 32d2095..55427bf 100644
--- a/src/onaptests/scenario/basic_network.py
+++ b/src/onaptests/scenario/basic_network.py
@@ -1,60 +1,15 @@
#!/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()
diff --git a/src/onaptests/scenario/basic_onboard.py b/src/onaptests/scenario/basic_onboard.py
index cb1d0fe..6f10765 100644
--- a/src/onaptests/scenario/basic_onboard.py
+++ b/src/onaptests/scenario/basic_onboard.py
@@ -1,49 +1,14 @@
#!/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()
diff --git a/src/onaptests/scenario/basic_sdnc.py b/src/onaptests/scenario/basic_sdnc.py
index 9255d29..d33f5f5 100644
--- a/src/onaptests/scenario/basic_sdnc.py
+++ b/src/onaptests/scenario/basic_sdnc.py
@@ -1,11 +1,7 @@
-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):
@@ -13,29 +9,7 @@ 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()
diff --git a/src/onaptests/scenario/basic_vm.py b/src/onaptests/scenario/basic_vm.py
index cbf57db..21ce800 100644
--- a/src/onaptests/scenario/basic_vm.py
+++ b/src/onaptests/scenario/basic_vm.py
@@ -1,56 +1,14 @@
#!/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()
diff --git a/src/onaptests/scenario/basic_vm_macro.py b/src/onaptests/scenario/basic_vm_macro.py
index 7159686..e526d14 100644
--- a/src/onaptests/scenario/basic_vm_macro.py
+++ b/src/onaptests/scenario/basic_vm_macro.py
@@ -1,35 +1,25 @@
"""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:
@@ -101,29 +91,7 @@ class BasicVmMacroStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/scenario/basic_vm_macro_stability.py b/src/onaptests/scenario/basic_vm_macro_stability.py
index 98464a4..7ad08e6 100644
--- a/src/onaptests/scenario/basic_vm_macro_stability.py
+++ b/src/onaptests/scenario/basic_vm_macro_stability.py
@@ -1,41 +1,13 @@
"""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()
diff --git a/src/onaptests/scenario/cds_blueprint_enrichment.py b/src/onaptests/scenario/cds_blueprint_enrichment.py
deleted file mode 100644
index a16169b..0000000
--- a/src/onaptests/scenario/cds_blueprint_enrichment.py
+++ /dev/null
@@ -1,48 +0,0 @@
-#!/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()
diff --git a/src/onaptests/scenario/cds_resource_resolution.py b/src/onaptests/scenario/cds_resource_resolution.py
index 470f030..fcc3e93 100644
--- a/src/onaptests/scenario/cds_resource_resolution.py
+++ b/src/onaptests/scenario/cds_resource_resolution.py
@@ -1,36 +1,24 @@
#!/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:
@@ -55,39 +43,13 @@ class CDSResourceResolutionStep(BaseStep):
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()
diff --git a/src/onaptests/scenario/clearwater_ims.py b/src/onaptests/scenario/clearwater_ims.py
index ed16bc4..9f88201 100644
--- a/src/onaptests/scenario/clearwater_ims.py
+++ b/src/onaptests/scenario/clearwater_ims.py
@@ -1,60 +1,15 @@
#!/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()
diff --git a/src/onaptests/scenario/multi_vnf_macro.py b/src/onaptests/scenario/multi_vnf_macro.py
index 82a0511..96ade6d 100644
--- a/src/onaptests/scenario/multi_vnf_macro.py
+++ b/src/onaptests/scenario/multi_vnf_macro.py
@@ -1,36 +1,26 @@
"""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:
@@ -105,32 +95,7 @@ class MultiVnfUbuntuMacroStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/scenario/pnf_macro.py b/src/onaptests/scenario/pnf_macro.py
index 7ecf3c2..a07404f 100644
--- a/src/onaptests/scenario/pnf_macro.py
+++ b/src/onaptests/scenario/pnf_macro.py
@@ -1,40 +1,28 @@
"""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:
@@ -59,7 +47,7 @@ class PnfMacroScenarioStep(YamlTemplateBaseStep):
str: Component name
"""
- return "PythonSDK-tests"
+ return "TEST"
@property
def yaml_template(self) -> dict:
@@ -106,28 +94,7 @@ class PnfMacroScenarioStep(YamlTemplateBaseStep):
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()
diff --git a/src/onaptests/scenario/scenario_base.py b/src/onaptests/scenario/scenario_base.py
index 6fea811..179bea8 100644
--- a/src/onaptests/scenario/scenario_base.py
+++ b/src/onaptests/scenario/scenario_base.py
@@ -1,6 +1,10 @@
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
@@ -13,8 +17,68 @@ class ScenarioBase(testcase.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)
diff --git a/src/onaptests/scenario/status.py b/src/onaptests/scenario/status.py
index 881a840..dae6443 100644
--- a/src/onaptests/scenario/status.py
+++ b/src/onaptests/scenario/status.py
@@ -1,53 +1,11 @@
-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()