@@ -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:
"""
@@ -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: