Create a decorator to collect step execution result and store them in storage class.
Storage class prepare a dictionary with step class name and execution result.
Issue-ID: INT-1733
Change-Id: I9c4030a0740085a9acca461c1581683c469ecbcf
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
from onapsdk.configuration import settings
from onapsdk.aai.business import Customer
+from .reports_collection import ReportsCollection
+
+
class BaseStep(ABC):
"""Base step class."""
self._steps: List["BaseStep"] = []
self._cleanup: bool = cleanup
self._parent: "BaseStep" = None
+ self._reports_collection: ReportsCollection = None
def add_step(self, step: "BaseStep") -> None:
"""Add substep.
"""
return self._parent is None
+ @property
+ def reports_collection(self) -> ReportsCollection:
+ """Collection to store step reports.
+
+ Store there if step result is "PASS" or "FAIL"
+
+ Returns:
+ Queue: Thread safe collection to store reports
+
+ """
+ if not self.is_root:
+ return self.parent.reports_collection
+ if not self._reports_collection:
+ self._reports_collection = ReportsCollection()
+ return self._reports_collection
+
+ @property
+ def name(self) -> str:
+ """Step name."""
+ return self.__class__.__name__
+
+ @classmethod
+ def store_state(cls, fun):
+ def wrapper(self, *args, **kwargs):
+ try:
+ ret = fun(self, *args, **kwargs)
+ self.reports_collection.put({self.name: "PASS"})
+ return ret
+ except Exception:
+ self.reports_collection.put({self.name: "FAIL"})
+ raise
+ return wrapper
+
def execute(self) -> None:
"""Step's action.
class ComplexCreateStep(BaseStep):
"""Complex creation step."""
+ @BaseStep.store_state
def execute(self):
"""Create complex.
self.add_step(LinkCloudRegionToComplexStep(cleanup=cleanup))
self.add_step(CustomerServiceSubscriptionCreateStep(cleanup=cleanup))
+ @BaseStep.store_state
def execute(self):
"""Connect service subsription to cloud region and tenant.
class CustomerCreateStep(BaseStep):
"""Customer creation step."""
+ @BaseStep.store_state
def execute(self):
"""Create cutomer.
super().__init__(cleanup=cleanup)
self.add_step(CustomerCreateStep(cleanup=cleanup))
+ @BaseStep.store_state
def execute(self):
"""Create customer service subsription.
super().__init__(cleanup=cleanup)
self.add_step(ComplexCreateStep(cleanup=cleanup))
+ @BaseStep.store_state
def execute(self):
"""Link cloud region to complex.
class RegisterCloudRegionStep(BaseStep):
"""Cloud region registration step."""
+ @BaseStep.store_state
def execute(self):
"""Register cloud region
self.add_step(ServiceOnboardStep(cleanup))
self.add_step(ConnectServiceSubToCloudRegionStep(cleanup))
+ @BaseStep.store_state
def execute(self):
"""Instantiate service.
return self._service_instance_name
return self.parent.service_instance_name
+ @YamlTemplateBaseStep.store_state
def execute(self):
"""Instantiate service.
value=vnf_parameter["value"]
)
+ @YamlTemplateBaseStep.store_state
def execute(self) -> None:
"""Instantiate Vf module.
return self._service_instance_name
return self.parent.service_instance_name
+ @YamlTemplateBaseStep.store_state
def execute(self):
"""Instantiate vnf.
super().__init__(cleanup=cleanup)
self.add_step(VfOnboardStep(cleanup=cleanup))
+ @BaseStep.store_state
def execute(self):
"""Onboard service.
else:
return self.parent.service_name
+ @YamlTemplateBaseStep.store_state
def execute(self):
"""Onboard service."""
super().execute()
class VendorOnboardStep(BaseStep):
"""Vendor onboard step."""
+ @BaseStep.store_state
def execute(self):
"""Onboard vendor.
super().__init__(cleanup=cleanup)
self.add_step(VspOnboardStep(cleanup=cleanup))
+ @BaseStep.store_state
def execute(self):
"""Onboard Vf.
"""
return self.parent.yaml_template[self.parent.service_name]
+ @YamlTemplateBaseStep.store_state
def execute(self):
"""Onboard Vfs from YAML template."""
super().execute()
super().__init__(cleanup=cleanup)
self.add_step(VendorOnboardStep(cleanup=cleanup))
+ @BaseStep.store_state
def execute(self):
"""Onboard Vsp.
"""
return self.parent.yaml_template
+ @YamlTemplateBaseStep.store_state
def execute(self):
"""Onboard Vsps from YAML template.
--- /dev/null
+from typing import Dict
+
+
+class ReportsCollection:
+ """Collection to store steps execution statuses."""
+
+ def __init__(self) -> None:
+ """Initialize collection."""
+ self._collection: list = []
+
+ def put(self, item: Dict[str, str]) -> None:
+ """Put execution status dictionary.
+
+ Args:
+ item (Dict[str, str]): Step name with status dictionary
+
+ """
+ self._collection.append(item)
+
+ @property
+ def report(self) -> Dict[str, str]:
+ """Get report.
+
+ Build a dictionary with execution statuses.
+
+ Returns:
+ Dict[str, str]: Steps name with status dictionary
+
+ """
+ report: Dict[str, str] = {}
+ for element in self._collection[::-1]:
+ print(element)
+ print(type(element))
+ report.update(element)
+ return report
--- /dev/null
+
+from onaptests.steps.reports_collection import ReportsCollection
+
+
+def test_reports_collection():
+ rc = ReportsCollection()
+ assert rc.report == {}
+
+ rc.put({"a": "b"})
+ assert rc.report == {"a": "b"}
--- /dev/null
+import pytest
+from onaptests.steps.base import BaseStep
+
+
+class TestStep(BaseStep):
+
+ @BaseStep.store_state
+ def execute(self):
+ return super().execute()
+
+
+class TestFailStep(BaseStep):
+
+ @BaseStep.store_state
+ def execute(self):
+ super().execute()
+ raise Exception
+
+
+def test_store_state():
+ ts = TestStep()
+ ts.execute()
+ assert ts.reports_collection.report == {"TestStep": "PASS"}
+
+ fs = TestFailStep()
+ fs.add_step(TestStep())
+ with pytest.raises(Exception):
+ fs.execute()
+ fs.reports_collection.report == {"TestFailStep": "FAIL", "TestStep": "PASS"}