self._cleanup: bool = cleanup
self._parent: "BaseStep" = None
self._reports_collection: ReportsCollection = None
+ self._start_time: float = None
def add_step(self, step: "BaseStep") -> None:
"""Add substep.
def store_state(cls, fun):
def wrapper(self, *args, **kwargs):
try:
- start_time: float = time.time()
ret = fun(self, *args, **kwargs)
execution_status: ReportStepStatus = ReportStepStatus.PASS
return ret
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(
step_description=f"[{self.component}] {self.name}: {self.description}",
step_execution_status=execution_status,
- step_execution_duration=time.time() - start_time
+ step_execution_duration=time.time() - self._start_time
)
)
return wrapper
def execute(self) -> None:
- """Step's action.
+ """Step's action execution.
Run all substeps action before it's own action.
- Override this method and remember to call `super().action()` before.
+ Override this method and remember to call `super().execute()` before.
"""
for step in self._steps:
step.execute()
+ self._start_time = time.time()
def cleanup(self) -> None:
"""Step's cleanup.
+from time import sleep
+
import pytest
from onaptests.steps.base import BaseStep
return "Test"
+class TestOneSecStep(BaseStep):
+
+ @BaseStep.store_state
+ def execute(self):
+ super().execute()
+ sleep(1)
+
+ @property
+ def description(self):
+ return "One second test step"
+
+ @property
+ def component(self) -> str:
+ return "Test"
+
+
+class TestStepNoSuperExecute(BaseStep):
+
+ @BaseStep.store_state
+ def execute(self):
+ sleep(1)
+
+ @property
+ def description(self):
+ return "One second test step - no super execute call"
+
+ @property
+ def component(self) -> str:
+ return "Test"
+
+
def test_store_state():
ts = TestStep()
ts.execute()
assert rep_s.step_description == "[Test] TestStep: Test pass step"
assert rep_s.step_execution_status.value == "PASS"
assert rep_s.step_execution_duration != 0
+
+
+def test_store_state_time_measurement():
+
+ ts = TestOneSecStep()
+ ts.execute()
+ assert len(ts.reports_collection.report) == 1
+ rep = ts.reports_collection.report[0]
+ assert rep.step_execution_duration > 1
+
+ ts = TestOneSecStep()
+ ts.add_step(TestOneSecStep())
+ ts.execute()
+ assert len(ts.reports_collection.report) == 2
+ rep_one, rep_two = ts.reports_collection.report
+ assert rep_one.step_execution_duration > 1 and rep_one.step_execution_duration < 2
+ assert rep_two.step_execution_duration > 1 and rep_two.step_execution_duration < 2
+
+ ts = TestStepNoSuperExecute()
+ ts.execute()
+ assert len(ts.reports_collection.report) == 1
+ rep = ts.reports_collection.report[0]
+ assert rep.step_execution_duration < 1