pytest: use case_tester for multi-device tests

components/esp_wifi/test_apps/wifi_connect/pytest_wifi_connect.py

@@ -1,34 +1,12 @@
 # SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Unlicense OR CC0-1.0
 
-from typing import Tuple
-
 import pytest
-from pytest_embedded_idf.dut import IdfDut
-
-
-def multi_device_case_run(dut: Tuple[IdfDut, IdfDut], test_case: str, sub_cases: list, timeout: int) -> None:
-    dut_num = len(sub_cases)
-    # start
-    for i in range(dut_num):
-        dut[i].write(f'"{test_case}"')
-    for i in range(dut_num):
-        dut[i].expect(r'\t\(2\)\s+\".+\"')
-    for i in range(dut_num):
-        dut[i].write(f'{i+1}')
-    # end
-    for i in range(dut_num):
-        dut[i].expect_unity_test_output(timeout=timeout)
-        dut[i].expect_exact("Enter next test, or 'enter' to see menu")
 
 
 @pytest.mark.esp32
 @pytest.mark.esp32c3
 @pytest.mark.wifi_two_dut
 @pytest.mark.parametrize('count', [2], indirect=True)
-def test_wifi_connect_cases(dut: Tuple[IdfDut, IdfDut]) -> None:
-    test_menu = dut[0].parse_test_menu()
-    dut[1].expect_exact('Press ENTER to see the list of tests')
-    for test_case in test_menu:
-        timeout = int(test_case.attributes.get('timeout', 60))
-        multi_device_case_run(dut, test_case.name, test_case.subcases, timeout)
+def test_wifi_connect_cases(case_tester) -> None:  # type: ignore
+    case_tester.run_all_cases()

conftest.py

@@ -35,7 +35,7 @@
 from pytest_embedded_idf.dut import IdfDut
 
 try:
-    from tools.ci.idf_unity_tester import CaseTester
+    from idf_unity_tester import CaseTester
 except ImportError:
     sys.path.append(os.path.join(os.path.dirname(__file__), 'tools', 'ci'))
     from idf_unity_tester import CaseTester

tools/ci/idf_unity_tester.py

@@ -206,7 +206,7 @@ def __init__(self, dut: Union[Dut, List[Dut]], **kwargs) -> None:     # type: ig
             dev_res = MultiDevResource(dut)
             self.group.append(dev_res)
 
-    def _wait_multi_dev_case_finish(self, timeout: int = 90) -> None:
+    def _wait_multi_dev_case_finish(self, timeout: int = 60) -> None:
         """
         Wait until all the sub-cases of this multi_device case finished
         """
@@ -216,7 +216,7 @@ def _wait_multi_dev_case_finish(self, timeout: int = 90) -> None:
             else:
                 raise TimeoutError('Wait case to finish timeout')
 
-    def _start_sub_case_thread(self, dev_res: MultiDevResource, case: UnittestMenuCase, sub_case_index: int) -> None:
+    def _start_sub_case_thread(self, dev_res: MultiDevResource, case: UnittestMenuCase, sub_case_index: int, timeout: int = 60) -> None:
         """
         Start the thread monitoring on the corresponding dut of the sub-case
         """
@@ -226,6 +226,7 @@ def _start_sub_case_thread(self, dev_res: MultiDevResource, case: UnittestMenuCa
         _kwargs['dev_res'] = dev_res
         _kwargs['case'] = case
         _kwargs['sub_case_index'] = sub_case_index
+        _kwargs['timeout'] = timeout
 
         # Create the thread of the sub-case
         dev_res.thread = Thread(target=self._run, kwargs=_kwargs, daemon=True)
@@ -254,6 +255,7 @@ def _run(self, **kwargs) -> None:  # type: ignore
         dev_res = kwargs['dev_res']
         case = kwargs['case']
         sub_case_index = kwargs['sub_case_index']
+        timeout = kwargs['timeout']
         # Start the case
         dut.expect(self.ready_pattern_list)
         # Retry at most 30 times if not write successfully
@@ -271,7 +273,7 @@ def _run(self, **kwargs) -> None:  # type: ignore
 
         # Wait for the specific patterns, only exist when the sub-case finished
         while True:
-            pat = dut.expect(signal_pattern_list, timeout=60)
+            pat = dut.expect(signal_pattern_list, timeout=timeout)
             if pat is not None:
                 match_str = pat.group().decode('utf-8')
 
@@ -302,7 +304,7 @@ def _run(self, **kwargs) -> None:  # type: ignore
         # The case finished, release the semaphore to unblock the '_wait_multi_dev_case_finish'
         dev_res.sem.release()
 
-    def run_all_multi_dev_cases(self, reset: bool = False, timeout: int = 90) -> None:
+    def run_all_multi_dev_cases(self, reset: bool = False, timeout: int = 60) -> None:
         """
         Run only multi_device cases
 
@@ -314,7 +316,7 @@ def run_all_multi_dev_cases(self, reset: bool = False, timeout: int = 90) -> Non
             # Run multi_device case on every device
             self.run_multi_dev_case(case, reset, timeout)
 
-    def run_multi_dev_case(self, case: UnittestMenuCase, reset: bool = False, timeout: int = 90) -> None:
+    def run_multi_dev_case(self, case: UnittestMenuCase, reset: bool = False, timeout: int = 60) -> None:
         """
         Run a specific multi_device case
 
@@ -335,8 +337,8 @@ def run_multi_dev_case(self, case: UnittestMenuCase, reset: bool = False, timeou
                     index = int(sub_case['index'], 10)
                 else:
                     index = sub_case['index']
-                self._start_sub_case_thread(dev_res=self.group[index - 1],
-                                            case=case, sub_case_index=index)
+                self._start_sub_case_thread(dev_res=self.group[index - 1], case=case,
+                                            sub_case_index=index, timeout=timeout)
             # Waiting all the devices to finish their test cases
             self._wait_multi_dev_case_finish(timeout=timeout)
 
@@ -351,7 +353,7 @@ class CaseTester(NormalCaseTester, MultiStageCaseTester, MultiDevCaseTester):
         test_menu (List[UnittestMenuCase]): The list of the cases
     """
 
-    def run_all_cases(self, reset: bool = False, timeout: int = 90) -> None:
+    def run_all_cases(self, reset: bool = False, timeout: int = 60) -> None:
         """
         Run all cases
 
@@ -362,18 +364,19 @@ def run_all_cases(self, reset: bool = False, timeout: int = 90) -> None:
         for case in self.test_menu:
             self.run_case(case, reset, timeout=timeout)
 
-    def run_case(self, case: UnittestMenuCase, reset: bool = False, timeout: int = 90) -> None:
+    def run_case(self, case: UnittestMenuCase, reset: bool = False, timeout: int = 60) -> None:
         """
         Run a specific case
 
         Args:
             case: the specific case that parsed in test menu
             reset: whether do a hardware reset before running the case
-            timeout: timeout in second
+            timeout: timeout in second, the case's timeout attribute has a higher priority than this param.
         """
+        _timeout = int(case.attributes.get('timeout', timeout))
         if case.type == 'normal':
-            self.run_normal_case(case, reset, timeout=timeout)
+            self.run_normal_case(case, reset, timeout=_timeout)
         elif case.type == 'multi_stage':
-            self.run_multi_stage_case(case, reset, timeout=timeout)
+            self.run_multi_stage_case(case, reset, timeout=_timeout)
         elif case.type == 'multi_device':
-            self.run_multi_dev_case(case, reset, timeout=timeout)
+            self.run_multi_dev_case(case, reset, timeout=_timeout)