+import functools
+import itertools
import logging
import logging.config
import time
from abc import ABC, abstractmethod
-from typing import List
+from typing import Iterator, List
from onapsdk.aai.business import Customer
from onapsdk.configuration import settings
-from onapsdk.exceptions import SettingsError
+from onapsdk.exceptions import SDKException, SettingsError
-from .reports_collection import Report, ReportsCollection, ReportStepStatus
+from onaptests.steps.reports_collection import Report, ReportsCollection, ReportStepStatus
+from onaptests.utils.exceptions import OnapTestException, SubstepExecutionException
class BaseStep(ABC):
self._cleanup: bool = cleanup
self._parent: "BaseStep" = None
self._reports_collection: ReportsCollection = None
- self._start_time: float = None
+ self._start_execution_time: float = None
+ self._start_cleanup_time: float = None
+ self._execution_report: ReportStepStatus = None
+ self._cleanup_report: ReportStepStatus = None
def add_step(self, step: "BaseStep") -> None:
"""Add substep.
return self.parent.reports_collection
if not self._reports_collection:
self._reports_collection = ReportsCollection()
+ for step_report in itertools.chain(self.execution_reports, self.cleanup_reports):
+ self._reports_collection.put(step_report)
return self._reports_collection
+ @property
+ def execution_reports(self) -> Iterator[ReportsCollection]:
+ """Execution reports generator.
+
+ Steps tree postorder traversal
+
+ Yields:
+ Iterator[ReportsCollection]: Step execution report
+
+ """
+ for step in self._steps:
+ yield from step.execution_reports
+ if self._execution_report:
+ yield self._execution_report
+
+ @property
+ def cleanup_reports(self) -> Iterator[ReportsCollection]:
+ """Cleanup reports generator.
+
+ Steps tree preorder traversal
+
+ Yields:
+ Iterator[ReportsCollection]: Step cleanup report
+
+ """
+ if self._cleanup:
+ if self._cleanup_report:
+ yield self._cleanup_report
+ for step in self._steps:
+ yield from step.cleanup_reports
+
@property
def name(self) -> str:
"""Step name."""
"""
@classmethod
- def store_state(cls, fun):
+ def store_state(cls, fun=None, *, cleanup=False):
+ if fun is None:
+ return functools.partial(cls.store_state, cleanup=cleanup)
+ @functools.wraps(fun)
def wrapper(self, *args, **kwargs):
try:
+ if cleanup:
+ self._start_cleanup_time = time.time()
ret = fun(self, *args, **kwargs)
execution_status: ReportStepStatus = ReportStepStatus.PASS
return ret
- except Exception:
+ except SubstepExecutionException:
+ execution_status: ReportStepStatus = ReportStepStatus.PASS if cleanup else ReportStepStatus.NOT_EXECUTED
+ raise
+ except (OnapTestException, SDKException):
execution_status: ReportStepStatus = ReportStepStatus.FAIL
raise
finally:
- if not self._start_time:
- self._logger.error("No execution start time saved for %s step. Fix it by call `super.execute()` "
- "in step class `execute()` method definition")
- self._start_time = time.time()
- self.reports_collection.put(
- Report(
+ if cleanup:
+ self._cleanup_report = Report(
+ step_description=f"[{self.component}] {self.name} cleanup: {self.description}",
+ step_execution_status=execution_status,
+ step_execution_duration=time.time() - self._start_cleanup_time
+ )
+ else:
+ if not self._start_execution_time:
+ 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,
- step_execution_duration=time.time() - self._start_time
+ step_execution_duration=time.time() - self._start_execution_time
)
- )
return wrapper
def execute(self) -> None:
"""
for step in self._steps:
- step.execute()
- self._start_time = time.time()
+ try:
+ step.execute()
+ except (OnapTestException, SDKException) as substep_err:
+ raise SubstepExecutionException from substep_err
+ self._start_execution_time = time.time()
def cleanup(self) -> None:
"""Step's cleanup.
"""
if self._cleanup:
for step in self._steps:
- step.cleanup()
+ try:
+ step.cleanup()
+ except (OnapTestException, SDKException) as substep_err:
+ raise SubstepExecutionException from substep_err
@classmethod
def set_proxy(cls, sock_http):
settings.CLOUD_REGION_CLOUD_OWNER,
open(settings.K8S_CONFIG, 'rb').read())
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup K8S Connectivity information."""
self._logger.info("Clean the k8s connectivity information")
####### Upload artifact for created profile ##############################
profile.upload_artifact(open(settings.K8S_PROFILE_ARTIFACT_PATH, 'rb').read())
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup K8S profiles.
"""
rb_name=settings.PNF_RB_NAME,
rb_version=settings.PNF_RB_VERSION)
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Delete instance."""
if self.instance:
service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name)
self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name)
-
+ @YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup Service.
service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name)
self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name)
-
+ @YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup Service.
if vf_module_instantiation.failed:
raise onap_test_exceptions.VfModuleInstantiateException
-
+ @YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup Vf module.
if net_instantiation.failed:
raise onap_test_exceptions.NetworkInstantiateException
+ @YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup VL.
if vnf_instantiation.failed:
raise onap_test_exceptions.VnfInstantiateException
+ @YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup VNF.
loop_name=loop_name,
operational_policies=operational_policies)
+ @YamlTemplateBaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup Service.
blueprint.enrich()
blueprint.save(settings.CDS_CBA_ENRICHED)
+ @BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""Cleanup enrichment step.
from onapsdk.configuration import settings
from onapsdk.exceptions import SettingsError
+
class ReportStepStatus(Enum):
"""Enum which stores steps execution statuses."""
PASS = "PASS"
FAIL = "FAIL"
+ NOT_EXECUTED = "NOT EXECUTED"
@dataclass
</thead>
<tbody>
{% for step_report in report.report %}
- <tr {% if step_report.step_execution_status.value == 'FAIL' %} class="has-background-danger" {% else %} class="has-background-success-light" {% endif %}>
+ <tr {% if step_report.step_execution_status.value == 'FAIL' %} class="has-background-danger" {% elif step_report.step_execution_status.value == 'PASS' %} class="has-background-success-light" {% else %} class="has-background-warning-light" {% endif %}>
<td>
{{ step_report.step_description }}
</td>
class EnvironmentPreparationException(OnapTestException):
"""Test environment preparation exception."""
error_message="Test can't be run properly due to preparation error"
+
+class SubstepExecutionException(OnapTestException):
+ """Exception raised if substep execution fails."""
import pytest
from onaptests.steps.base import BaseStep
+from onaptests.utils.exceptions import OnapTestException
def execute(self):
return super().execute()
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ return super().cleanup()
+
@property
def description(self):
return "Test pass step"
@BaseStep.store_state
def execute(self):
super().execute()
- raise Exception
+ raise OnapTestException
+
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self) -> None:
+ raise OnapTestException
@property
def description(self):
return "Test"
+class TestCleanupStepA(BaseStep):
+
+ @BaseStep.store_state
+ def execute(self):
+ return super().execute()
+
+ @BaseStep.store_state(cleanup=True)
+ def cleanup(self):
+ return super().cleanup()
+
+ @property
+ def description(self):
+ return "Test cleanup step A"
+
+ @property
+ def component(self) -> str:
+ return "Test"
+
+
+class TestCleanupStepB(TestCleanupStepA):
+
+ @property
+ def description(self):
+ return "Test cleanup step B"
+
+
+class TestCleanupStepC(TestCleanupStepA):
+
+ @property
+ def description(self):
+ return "Test cleanup step C"
+
+
+class TestCleanupStepD(TestCleanupStepA):
+
+ @property
+ def description(self):
+ return "Test cleanup step D"
+
+
def test_store_state():
ts = TestStep()
ts.execute()
assert rep_s.step_execution_status.value == "PASS"
assert rep_s.step_execution_duration != 0
+ ts = TestStep(cleanup=True)
+ ts.add_step(TestFailStep(cleanup=True))
+ with pytest.raises(Exception):
+ ts.execute()
+ with pytest.raises(Exception):
+ ts.cleanup()
+ assert len(ts.reports_collection.report) == 4
+ cln_rep_f, cln_rep_s, rep_s, rep_f = ts.reports_collection.report
+ assert rep_f.step_description == "[Test] TestFailStep: Test fail step"
+ assert rep_f.step_execution_status.value == "FAIL"
+ assert rep_f.step_execution_duration != 0
+
+ assert rep_s.step_description == "[Test] TestStep: Test pass step"
+ assert rep_s.step_execution_status.value == "NOT EXECUTED"
+ assert rep_s.step_execution_duration != 0
+
+ assert cln_rep_s.step_description == "[Test] TestStep cleanup: Test pass step"
+ assert cln_rep_s.step_execution_status.value == "PASS"
+ assert cln_rep_s.step_execution_duration != 0
+
+ assert cln_rep_f.step_description == "[Test] TestFailStep cleanup: Test fail step"
+ assert cln_rep_f.step_execution_status.value == "FAIL"
+ assert cln_rep_f.step_execution_duration != 0
+
+ ts = TestStep(cleanup=True)
+ tsf = TestFailStep(cleanup=True)
+ tsf.add_step(TestStep(cleanup=True))
+ ts.add_step(tsf)
+ ts.add_step(TestStep(cleanup=True))
+ with pytest.raises(Exception):
+ ts.execute()
+ with pytest.raises(Exception):
+ ts.cleanup()
+
+ assert len(ts.reports_collection.report) == 5
+ cln_2, cln_1, exec_3, exec_2, exec_1 = ts.reports_collection.report
+
+ assert exec_1.step_description == "[Test] TestStep: Test pass step"
+ assert exec_1.step_execution_status.value == "PASS"
+ assert exec_1.step_execution_duration != 0
+
+ assert exec_2.step_description == "[Test] TestFailStep: Test fail step"
+ assert exec_2.step_execution_status.value == "FAIL"
+ assert exec_2.step_execution_duration != 0
+
+ assert exec_3.step_description == "[Test] TestStep: Test pass step"
+ assert exec_3.step_execution_status.value == "NOT EXECUTED"
+ assert exec_3.step_execution_duration != 0
+
+ assert cln_1.step_description == "[Test] TestStep cleanup: Test pass step"
+ assert cln_1.step_execution_status.value == "PASS"
+ assert cln_1.step_execution_duration != 0
+
+ assert cln_2.step_description == "[Test] TestFailStep cleanup: Test fail step"
+ assert cln_2.step_execution_status.value == "FAIL"
+ assert cln_2.step_execution_duration != 0
+
def test_store_state_time_measurement():
assert len(ts.reports_collection.report) == 1
rep = ts.reports_collection.report[0]
assert rep.step_execution_duration < 1
+
+
+def test_store_state_with_cleanup():
+
+ ts = TestCleanupStepA(cleanup=True)
+ ts_b = TestCleanupStepB(cleanup=True)
+ ts_b.add_step(TestCleanupStepC(cleanup=True))
+ ts.add_step(ts_b)
+ ts.add_step(TestCleanupStepD(cleanup=True))
+ ts.execute()
+ ts.cleanup()
+ assert len(ts.reports_collection.report) == 8
+ (rep_cleanup_step_4, rep_cleanup_step_3, rep_cleanup_step_2, rep_cleanup_step_1,
+ rep_exec_step_4, rep_exec_step_3, rep_exec_step_2, rep_exec_step_1) = ts.reports_collection.report
+ assert rep_exec_step_1.step_description == "[Test] TestCleanupStepC: Test cleanup step C"
+ assert rep_exec_step_2.step_description == "[Test] TestCleanupStepB: Test cleanup step B"
+ assert rep_exec_step_3.step_description == "[Test] TestCleanupStepD: Test cleanup step D"
+ assert rep_exec_step_4.step_description == "[Test] TestCleanupStepA: Test cleanup step A"
+ assert rep_cleanup_step_1.step_description == "[Test] TestCleanupStepA cleanup: Test cleanup step A"
+ assert rep_cleanup_step_2.step_description == "[Test] TestCleanupStepB cleanup: Test cleanup step B"
+ assert rep_cleanup_step_3.step_description == "[Test] TestCleanupStepC cleanup: Test cleanup step C"
+ assert rep_cleanup_step_4.step_description == "[Test] TestCleanupStepD cleanup: Test cleanup step D"
Code Review / testsuite / pythonsdk-tests.git / commitdiff