ci(pytest): support multi-dut different app

tools/ci/build_pytest_apps.py

@@ -13,7 +13,7 @@ import sys
 from collections import defaultdict
 from typing import List
 
-from idf_ci_utils import IDF_PATH, get_pytest_cases
+from idf_ci_utils import IDF_PATH, PytestCase, get_pytest_cases
 
 try:
     from build_apps import build_apps
@@ -28,15 +28,16 @@ except ImportError:
 
 
 def main(args: argparse.Namespace) -> None:
-    pytest_cases = []
+    pytest_cases: List[PytestCase] = []
     for path in args.paths:
         pytest_cases += get_pytest_cases(path, args.target)
 
     paths = set()
     app_configs = defaultdict(set)
     for case in pytest_cases:
-        paths.add(case.app_path)
-        app_configs[case.app_path].add(case.config)
+        for app in case.apps:
+            paths.add(app.path)
+            app_configs[app.path].add(app.config)
 
     app_dirs = list(paths)
     if not app_dirs:

tools/ci/idf_ci_utils.py

@@ -1,19 +1,22 @@
 # internal use only for CI
 # some CI related util functions
 #
-# SPDX-FileCopyrightText: 2020-2021 Espressif Systems (Shanghai) CO LTD
+# SPDX-FileCopyrightText: 2020-2022 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Apache-2.0
 #
+
 import io
 import logging
 import os
 import subprocess
 import sys
 from contextlib import redirect_stdout
-from typing import TYPE_CHECKING, List
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, List, Set
 
 if TYPE_CHECKING:
-    from _pytest.nodes import Function
+    from _pytest.python import Function
+
 
 IDF_PATH = os.path.abspath(
     os.getenv('IDF_PATH', os.path.join(os.path.dirname(__file__), '..', '..'))
@@ -113,39 +116,82 @@ def is_in_directory(file_path: str, folder: str) -> bool:
     return os.path.realpath(file_path).startswith(os.path.realpath(folder) + os.sep)
 
 
+def to_list(s: Any) -> List[Any]:
+    if isinstance(s, set) or isinstance(s, tuple):
+        return list(s)
+    elif isinstance(s, list):
+        return s
+    else:
+        return [s]
+
+
+@dataclass
+class PytestApp:
+    path: str
+    target: str
+    config: str
+
+    def __hash__(self) -> int:
+        return hash((self.path, self.target, self.config))
+
+
+@dataclass
 class PytestCase:
-    def __init__(self, test_path: str, target: str, config: str, case: str):
-        self.app_path = os.path.dirname(test_path)
-        self.test_path = test_path
-        self.target = target
-        self.config = config
-        self.case = case
+    path: str
+    name: str
+    apps: Set[PytestApp]
 
-    def __repr__(self) -> str:
-        return f'{self.test_path}: {self.target}.{self.config}.{self.case}'
+    def __hash__(self) -> int:
+        return hash((self.path, self.name, self.apps))
 
 
 class PytestCollectPlugin:
     def __init__(self, target: str) -> None:
         self.target = target
-        self.nodes: List[PytestCase] = []
+        self.cases: List[PytestCase] = []
+
+    @staticmethod
+    def get_param(item: 'Function', key: str, default: Any = None) -> Any:
+        if not hasattr(item, 'callspec'):
+            raise ValueError(f'Function {item} does not have params')
+
+        return item.callspec.params.get(key, default) or default
 
     def pytest_collection_modifyitems(self, items: List['Function']) -> None:
-        for item in items:
-            try:
-                file_path = str(item.path)
-            except AttributeError:
-                # pytest 6.x
-                file_path = item.fspath
+        from pytest_embedded.plugin import parse_multi_dut_args
 
+        for item in items:
+            count = 1
+            case_path = str(item.path)
+            case_name = item.originalname
             target = self.target
+            # funcargs is not calculated while collection
             if hasattr(item, 'callspec'):
-                config = item.callspec.params.get('config', 'default')
+                count = item.callspec.params.get('count', 1)
+                app_paths = to_list(
+                    parse_multi_dut_args(
+                        count,
+                        self.get_param(item, 'app_path', os.path.dirname(case_path)),
+                    )
+                )
+                configs = to_list(
+                    parse_multi_dut_args(
+                        count, self.get_param(item, 'config', 'default')
+                    )
+                )
+                targets = to_list(
+                    parse_multi_dut_args(count, self.get_param(item, 'target', target))
+                )
             else:
-                config = 'default'
-            case_name = item.originalname
+                app_paths = [os.path.dirname(case_path)]
+                configs = ['default']
+                targets = [target]
 
-            self.nodes.append(PytestCase(file_path, target, config, case_name))
+            case_apps = set()
+            for i in range(count):
+                case_apps.add(PytestApp(app_paths[i], targets[i], configs[i]))
+
+            self.cases.append(PytestCase(case_path, case_name, case_apps))
 
 
 def get_pytest_cases(folder: str, target: str) -> List[PytestCase]:
@@ -156,18 +202,23 @@ def get_pytest_cases(folder: str, target: str) -> List[PytestCase]:
 
     with io.StringIO() as buf:
         with redirect_stdout(buf):
-            res = pytest.main(['--collect-only', folder, '-q', '--target', target], plugins=[collector])
+            res = pytest.main(
+                ['--collect-only', folder, '-q', '--target', target],
+                plugins=[collector],
+            )
         if res.value != ExitCode.OK:
             if res.value == ExitCode.NO_TESTS_COLLECTED:
-                print(f'WARNING: no pytest app found for target {target} under folder {folder}')
+                print(
+                    f'WARNING: no pytest app found for target {target} under folder {folder}'
+                )
             else:
                 print(buf.getvalue())
                 raise RuntimeError('pytest collection failed')
 
-    return collector.nodes
+    return collector.cases
 
 
-def get_pytest_app_paths(folder: str, target: str) -> List[str]:
-    nodes = get_pytest_cases(folder, target)
+def get_pytest_app_paths(folder: str, target: str) -> Set[str]:
+    cases = get_pytest_cases(folder, target)
 
-    return list({node.app_path for node in nodes})
+    return set({app.path for case in cases for app in case.apps})