[RFC PATCH v1 2/5] dts: move test suite execution logic to DTSRunner
Juraj Linkeš
juraj.linkes at pantheon.tech
Wed Dec 20 11:33:28 CET 2023
Move the code responsible for running the test suite from the
TestSuite class to the DTSRunner class. This restructuring decision
was made to consolidate and unify the related logic into a single unit.
Signed-off-by: Juraj Linkeš <juraj.linkes at pantheon.tech>
---
dts/framework/runner.py | 156 +++++++++++++++++++++++++++++++++---
dts/framework/test_suite.py | 134 +------------------------------
2 files changed, 147 insertions(+), 143 deletions(-)
diff --git a/dts/framework/runner.py b/dts/framework/runner.py
index 5b077c5805..5e145a8066 100644
--- a/dts/framework/runner.py
+++ b/dts/framework/runner.py
@@ -5,6 +5,7 @@
import logging
import sys
+from types import MethodType
from .config import (
BuildTargetConfiguration,
@@ -12,10 +13,18 @@
ExecutionConfiguration,
TestSuiteConfig,
)
-from .exception import BlockingTestSuiteError
+from .exception import BlockingTestSuiteError, SSHTimeoutError, TestCaseVerifyError
from .logger import DTSLOG, getLogger
-from .test_result import BuildTargetResult, DTSResult, ExecutionResult, Result
-from .test_suite import get_test_suites
+from .settings import SETTINGS
+from .test_result import (
+ BuildTargetResult,
+ DTSResult,
+ ExecutionResult,
+ Result,
+ TestCaseResult,
+ TestSuiteResult,
+)
+from .test_suite import TestSuite, get_test_suites
from .testbed_model import SutNode, TGNode
from .utils import check_dts_python_version
@@ -148,7 +157,7 @@ def _run_build_target(
build_target_result.update_setup(Result.FAIL, e)
else:
- self._run_all_suites(sut_node, tg_node, execution, build_target_result)
+ self._run_test_suites(sut_node, tg_node, execution, build_target_result)
finally:
try:
@@ -158,7 +167,7 @@ def _run_build_target(
self._logger.exception("Build target teardown failed.")
build_target_result.update_teardown(Result.FAIL, e)
- def _run_all_suites(
+ def _run_test_suites(
self,
sut_node: SutNode,
tg_node: TGNode,
@@ -175,7 +184,7 @@ def _run_all_suites(
execution.test_suites[:0] = [TestSuiteConfig.from_dict("smoke_tests")]
for test_suite_config in execution.test_suites:
try:
- self._run_single_suite(
+ self._run_test_suite(
sut_node, tg_node, execution, build_target_result, test_suite_config
)
except BlockingTestSuiteError as e:
@@ -189,7 +198,7 @@ def _run_all_suites(
if end_build_target:
break
- def _run_single_suite(
+ def _run_test_suite(
self,
sut_node: SutNode,
tg_node: TGNode,
@@ -198,6 +207,9 @@ def _run_single_suite(
test_suite_config: TestSuiteConfig,
) -> None:
"""Runs a single test suite.
+ Setup, execute and teardown the whole suite.
+ Suite execution consists of running all test cases scheduled to be executed.
+ A test cast run consists of setup, execution and teardown of said test case.
Args:
sut_node: Node to run tests on.
@@ -222,13 +234,131 @@ def _run_single_suite(
else:
for test_suite_class in test_suite_classes:
test_suite = test_suite_class(
- sut_node,
- tg_node,
- test_suite_config.test_cases,
- execution.func,
- build_target_result,
+ sut_node, tg_node, test_suite_config.test_cases
+ )
+
+ test_suite_name = test_suite.__class__.__name__
+ test_suite_result = build_target_result.add_test_suite(test_suite_name)
+ try:
+ self._logger.info(f"Starting test suite setup: {test_suite_name}")
+ test_suite.set_up_suite()
+ test_suite_result.update_setup(Result.PASS)
+ self._logger.info(f"Test suite setup successful: {test_suite_name}")
+ except Exception as e:
+ self._logger.exception(f"Test suite setup ERROR: {test_suite_name}")
+ test_suite_result.update_setup(Result.ERROR, e)
+
+ else:
+ self._execute_test_suite(
+ execution.func, test_suite, test_suite_result
+ )
+
+ finally:
+ try:
+ test_suite.tear_down_suite()
+ sut_node.kill_cleanup_dpdk_apps()
+ test_suite_result.update_teardown(Result.PASS)
+ except Exception as e:
+ self._logger.exception(
+ f"Test suite teardown ERROR: {test_suite_name}"
+ )
+ self._logger.warning(
+ f"Test suite '{test_suite_name}' teardown failed, "
+ f"the next test suite may be affected."
+ )
+ test_suite_result.update_setup(Result.ERROR, e)
+ if (
+ len(test_suite_result.get_errors()) > 0
+ and test_suite.is_blocking
+ ):
+ raise BlockingTestSuiteError(test_suite_name)
+
+ def _execute_test_suite(
+ self, func: bool, test_suite: TestSuite, test_suite_result: TestSuiteResult
+ ) -> None:
+ """
+ Execute all test cases scheduled to be executed in this suite.
+ """
+ if func:
+ for test_case_method in test_suite._get_functional_test_cases():
+ test_case_name = test_case_method.__name__
+ test_case_result = test_suite_result.add_test_case(test_case_name)
+ all_attempts = SETTINGS.re_run + 1
+ attempt_nr = 1
+ self._run_test_case(test_suite, test_case_method, test_case_result)
+ while not test_case_result and attempt_nr < all_attempts:
+ attempt_nr += 1
+ self._logger.info(
+ f"Re-running FAILED test case '{test_case_name}'. "
+ f"Attempt number {attempt_nr} out of {all_attempts}."
+ )
+ self._run_test_case(test_suite, test_case_method, test_case_result)
+
+ def _run_test_case(
+ self,
+ test_suite: TestSuite,
+ test_case_method: MethodType,
+ test_case_result: TestCaseResult,
+ ) -> None:
+ """
+ Setup, execute and teardown a test case in this suite.
+ Exceptions are caught and recorded in logs and results.
+ """
+ test_case_name = test_case_method.__name__
+
+ try:
+ # run set_up function for each case
+ test_suite.set_up_test_case()
+ test_case_result.update_setup(Result.PASS)
+ except SSHTimeoutError as e:
+ self._logger.exception(f"Test case setup FAILED: {test_case_name}")
+ test_case_result.update_setup(Result.FAIL, e)
+ except Exception as e:
+ self._logger.exception(f"Test case setup ERROR: {test_case_name}")
+ test_case_result.update_setup(Result.ERROR, e)
+
+ else:
+ # run test case if setup was successful
+ self._execute_test_case(test_case_method, test_case_result)
+
+ finally:
+ try:
+ test_suite.tear_down_test_case()
+ test_case_result.update_teardown(Result.PASS)
+ except Exception as e:
+ self._logger.exception(f"Test case teardown ERROR: {test_case_name}")
+ self._logger.warning(
+ f"Test case '{test_case_name}' teardown failed, "
+ f"the next test case may be affected."
)
- test_suite.run()
+ test_case_result.update_teardown(Result.ERROR, e)
+ test_case_result.update(Result.ERROR)
+
+ def _execute_test_case(
+ self, test_case_method: MethodType, test_case_result: TestCaseResult
+ ) -> None:
+ """
+ Execute one test case and handle failures.
+ """
+ test_case_name = test_case_method.__name__
+ try:
+ self._logger.info(f"Starting test case execution: {test_case_name}")
+ test_case_method()
+ test_case_result.update(Result.PASS)
+ self._logger.info(f"Test case execution PASSED: {test_case_name}")
+
+ except TestCaseVerifyError as e:
+ self._logger.exception(f"Test case execution FAILED: {test_case_name}")
+ test_case_result.update(Result.FAIL, e)
+ except Exception as e:
+ self._logger.exception(f"Test case execution ERROR: {test_case_name}")
+ test_case_result.update(Result.ERROR, e)
+ except KeyboardInterrupt:
+ self._logger.error(
+ f"Test case execution INTERRUPTED by user: {test_case_name}"
+ )
+ test_case_result.update(Result.SKIP)
+ raise KeyboardInterrupt("Stop DTS")
def _exit_dts(self) -> None:
"""
diff --git a/dts/framework/test_suite.py b/dts/framework/test_suite.py
index 4a7907ec33..e96305deb0 100644
--- a/dts/framework/test_suite.py
+++ b/dts/framework/test_suite.py
@@ -17,15 +17,9 @@
from scapy.layers.l2 import Ether # type: ignore[import]
from scapy.packet import Packet, Padding # type: ignore[import]
-from .exception import (
- BlockingTestSuiteError,
- ConfigurationError,
- SSHTimeoutError,
- TestCaseVerifyError,
-)
+from .exception import ConfigurationError, TestCaseVerifyError
from .logger import DTSLOG, getLogger
from .settings import SETTINGS
-from .test_result import BuildTargetResult, Result, TestCaseResult, TestSuiteResult
from .testbed_model import SutNode, TGNode
from .testbed_model.hw.port import Port, PortLink
from .utils import get_packet_summaries
@@ -50,11 +44,10 @@ class TestSuite(object):
"""
sut_node: SutNode
+ tg_node: TGNode
is_blocking = False
_logger: DTSLOG
_test_cases_to_run: list[str]
- _func: bool
- _result: TestSuiteResult
_port_links: list[PortLink]
_sut_port_ingress: Port
_sut_port_egress: Port
@@ -69,17 +62,13 @@ def __init__(
self,
sut_node: SutNode,
tg_node: TGNode,
- test_cases: list[str],
- func: bool,
- build_target_result: BuildTargetResult,
+ test_cases_to_run: list[str],
):
self.sut_node = sut_node
self.tg_node = tg_node
self._logger = getLogger(self.__class__.__name__)
- self._test_cases_to_run = test_cases
+ self._test_cases_to_run = test_cases_to_run
self._test_cases_to_run.extend(SETTINGS.test_cases)
- self._func = func
- self._result = build_target_result.add_test_suite(self.__class__.__name__)
self._port_links = []
self._process_links()
self._sut_port_ingress, self._tg_port_egress = (
@@ -280,60 +269,6 @@ def _verify_l3_packet(self, received_packet: IP, expected_packet: IP) -> bool:
return False
return True
- def run(self) -> None:
- """
- Setup, execute and teardown the whole suite.
- Suite execution consists of running all test cases scheduled to be executed.
- A test cast run consists of setup, execution and teardown of said test case.
- """
- test_suite_name = self.__class__.__name__
-
- try:
- self._logger.info(f"Starting test suite setup: {test_suite_name}")
- self.set_up_suite()
- self._result.update_setup(Result.PASS)
- self._logger.info(f"Test suite setup successful: {test_suite_name}")
- except Exception as e:
- self._logger.exception(f"Test suite setup ERROR: {test_suite_name}")
- self._result.update_setup(Result.ERROR, e)
-
- else:
- self._execute_test_suite()
-
- finally:
- try:
- self.tear_down_suite()
- self.sut_node.kill_cleanup_dpdk_apps()
- self._result.update_teardown(Result.PASS)
- except Exception as e:
- self._logger.exception(f"Test suite teardown ERROR: {test_suite_name}")
- self._logger.warning(
- f"Test suite '{test_suite_name}' teardown failed, "
- f"the next test suite may be affected."
- )
- self._result.update_setup(Result.ERROR, e)
- if len(self._result.get_errors()) > 0 and self.is_blocking:
- raise BlockingTestSuiteError(test_suite_name)
-
- def _execute_test_suite(self) -> None:
- """
- Execute all test cases scheduled to be executed in this suite.
- """
- if self._func:
- for test_case_method in self._get_functional_test_cases():
- test_case_name = test_case_method.__name__
- test_case_result = self._result.add_test_case(test_case_name)
- all_attempts = SETTINGS.re_run + 1
- attempt_nr = 1
- self._run_test_case(test_case_method, test_case_result)
- while not test_case_result and attempt_nr < all_attempts:
- attempt_nr += 1
- self._logger.info(
- f"Re-running FAILED test case '{test_case_name}'. "
- f"Attempt number {attempt_nr} out of {all_attempts}."
- )
- self._run_test_case(test_case_method, test_case_result)
-
def _get_functional_test_cases(self) -> list[MethodType]:
"""
Get all functional test cases.
@@ -363,67 +298,6 @@ def _should_be_executed(self, test_case_name: str, test_case_regex: str) -> bool
return match
- def _run_test_case(
- self, test_case_method: MethodType, test_case_result: TestCaseResult
- ) -> None:
- """
- Setup, execute and teardown a test case in this suite.
- Exceptions are caught and recorded in logs and results.
- """
- test_case_name = test_case_method.__name__
-
- try:
- # run set_up function for each case
- self.set_up_test_case()
- test_case_result.update_setup(Result.PASS)
- except SSHTimeoutError as e:
- self._logger.exception(f"Test case setup FAILED: {test_case_name}")
- test_case_result.update_setup(Result.FAIL, e)
- except Exception as e:
- self._logger.exception(f"Test case setup ERROR: {test_case_name}")
- test_case_result.update_setup(Result.ERROR, e)
-
- else:
- # run test case if setup was successful
- self._execute_test_case(test_case_method, test_case_result)
-
- finally:
- try:
- self.tear_down_test_case()
- test_case_result.update_teardown(Result.PASS)
- except Exception as e:
- self._logger.exception(f"Test case teardown ERROR: {test_case_name}")
- self._logger.warning(
- f"Test case '{test_case_name}' teardown failed, "
- f"the next test case may be affected."
- )
- test_case_result.update_teardown(Result.ERROR, e)
- test_case_result.update(Result.ERROR)
-
- def _execute_test_case(
- self, test_case_method: MethodType, test_case_result: TestCaseResult
- ) -> None:
- """
- Execute one test case and handle failures.
- """
- test_case_name = test_case_method.__name__
- try:
- self._logger.info(f"Starting test case execution: {test_case_name}")
- test_case_method()
- test_case_result.update(Result.PASS)
- self._logger.info(f"Test case execution PASSED: {test_case_name}")
-
- except TestCaseVerifyError as e:
- self._logger.exception(f"Test case execution FAILED: {test_case_name}")
- test_case_result.update(Result.FAIL, e)
- except Exception as e:
- self._logger.exception(f"Test case execution ERROR: {test_case_name}")
- test_case_result.update(Result.ERROR, e)
- except KeyboardInterrupt:
- self._logger.error(f"Test case execution INTERRUPTED by user: {test_case_name}")
- test_case_result.update(Result.SKIP)
- raise KeyboardInterrupt("Stop DTS")
-
def get_test_suites(testsuite_module_path: str) -> list[type[TestSuite]]:
def is_test_suite(object) -> bool:
--
2.34.1
More information about the dev
mailing list