Feature/adjust estimated multitest exec time based on testcase count

doc/en/introduction.rst

@@ -346,7 +346,8 @@ Command line
                                 "<Multitest>": {
                                     | "execution_time": 199.99,
                                     | "setup_time": 39.99,
-                                    | "teardown_time": 39.99, // optional
+                                    | "teardown_time": 39.99,
+                                    | "testcase_count": 10  // optional
                                 },
                             }
 

doc/en/pools.rst

@@ -390,7 +390,8 @@ enabled by providing runtime data like following via ``--runtime-data`` command
         "<Multitest>": {
             "execution_time": 199.99,
             "setup_time": 39.99,
-            "teardown_time": 0
+            "teardown_time": 0,
+            "testcase_count": 10
         },
         ......
     }

doc/newsfragments/3294_changed.testcase_count_adjusted_runtime.rst

@@ -0,0 +1 @@
+Runtime data JSON file specified through ``--runtime-data`` now accepts an optional ``testcase_count`` field, which should hold the total number of testcases from the previous run. If ``testcase_count`` is specified, estimated MultiTest execution time will be adjusted by the ratio of current run's testcase count to that of previous run, which should bring more accurate auto-partitioning and improved testing resource utilization.

examples/ExecutionPools/AutoPart/runtime_data.txt

@@ -2,6 +2,7 @@
   "Proj1-suite": {
     "execution_time": 199.99,
     "setup_time": 5,
-    "teardown_time": 0
+    "teardown_time": 0,
+    "testcase_count": 8
   }
 }

testplan/base.py

@@ -174,7 +174,7 @@ def __init__(
         parse_cmdline: bool = True,
         parser: Type[TestplanParser] = TestplanParser,
         interactive_port: Optional[int] = None,
-        abort_signals: Optional[List[int]] = None,
+        abort_signals: Optional[List[signal.Signals]] = None,
         logger_level: int = logger.USER_INFO,
         file_log_level: int = logger.DEBUG,
         runpath: Union[str, Callable] = path.default_runpath,
@@ -197,8 +197,8 @@ def __init__(
         browse: bool = False,
         ui_port: Optional[int] = None,
         web_server_startup_timeout: int = defaults.WEB_SERVER_TIMEOUT,
-        test_filter: Type[BaseFilter] = filtering.Filter(),
-        test_sorter: Type[BaseSorter] = ordering.NoopSorter(),
+        test_filter: BaseFilter = filtering.Filter(),
+        test_sorter: BaseSorter = ordering.NoopSorter(),
         test_lister: Optional[MetadataBasedLister] = None,
         test_lister_output: Optional[os.PathLike] = None,
         verbose: bool = False,

testplan/common/entity/base.py

@@ -906,7 +906,7 @@ class RunnableResult:
 
     def __init__(self):
         self.step_results = OrderedDict()
-        self.run = False
+        self.run: Union[bool, Exception] = False
 
     def __repr__(self):
         return f"{self.__class__.__name__}[{vars(self)}]"

testplan/common/utils/logger.py

@@ -106,7 +106,7 @@ def _custom_log(self, level, msg, *args, **kwargs):
             self._log(level, msg, args, **kwargs)
 
 
-def _initial_setup():
+def _initial_setup() -> tuple[TestplanLogger, logging.StreamHandler]:
     """
     Perform initial setup for the logger. Creates and adds a handler to log
     to stdout with default level USER_INFO.

testplan/parser.py

@@ -202,6 +202,7 @@ def generate_parser(self) -> HelpParser:
         "execution_time": 199.99,
         "setup_time": 39.99,
         "teardown_time": 39.99,
+        "testcase_count": 10
     },
     ......
 }""",
@@ -599,7 +600,8 @@ def _read_text_file(file: str) -> List[str]:
         str: {
             "execution_time": schema.Or(int, float),
             "setup_time": schema.Or(int, float),
-            schema.Optional("teardown_time"): schema.Or(int, float),
+            "teardown_time": schema.Or(int, float),
+            schema.Optional("testcase_count"): int,
         }
     }
 )

testplan/runnable/base.py

@@ -913,22 +913,29 @@ def auto_part(self, tasks: List[Task]) -> List[Task]:
         discovered: List[TaskInformation] = [
             _detach_task_info(task) for task in tasks
         ]
-        auto_part_runtime_limit = self._calculate_part_runtime(discovered)
-
+        runtime_data = self.cfg.runtime_data or {}
+        auto_part_runtime_limit = self._calculate_part_runtime(
+            discovered, runtime_data
+        )
         for task_info in discovered:
             partitioned.extend(
                 self._calculate_parts_and_weights(
-                    task_info, auto_part_runtime_limit
+                    task_info, auto_part_runtime_limit, runtime_data
                 )
             )
 
+        # here we replace the original overall runtime data with "partitioned" values
+        # XXX: testcase_count are still sum-up value from previous run, what to do?
+        self.cfg.set_local("runtime_data", runtime_data)
+
         return [_attach_task_info(task_info) for task_info in partitioned]
 
-    def _calculate_part_runtime(self, discovered: List[TaskInformation]):
+    def _calculate_part_runtime(
+        self, discovered: List[TaskInformation], runtime_data: dict
+    ) -> float:
         if self.cfg.auto_part_runtime_limit != "auto":
             return self.cfg.auto_part_runtime_limit
 
-        runtime_data = self.cfg.runtime_data or {}
         if not runtime_data:
             self.logger.warning(
                 "Cannot derive auto_part_runtime_limit without runtime data, "
@@ -984,17 +991,46 @@ def _calculate_part_runtime(self, discovered: List[TaskInformation]):
         return auto_part_runtime_limit
 
     def _calculate_parts_and_weights(
-        self, task_info: TaskInformation, auto_part_runtime_limit: float
+        self,
+        task_info: TaskInformation,
+        auto_part_runtime_limit: float,
+        runtime_data: dict,
     ):
         num_of_parts = (
             task_info.num_of_parts
         )  # @task_target(multitest_parts=...)
         uid = task_info.uid
-        runtime_data: dict = self.cfg.runtime_data or {}
-        time_info = runtime_data.get(uid, None)
+        time_info: Optional[dict] = runtime_data.get(uid, None)
 
         partitioned: List[TaskInformation] = []
 
+        adjusted_exec_time = prev_case_count = curr_case_count = 0
+        if time_info:
+            adjusted_exec_time = time_info["execution_time"]
+            if isinstance(task_info.materialized_test, MultiTest):
+                if prev_case_count := time_info.get("testcase_count", 0):
+                    # XXX: cache dry_run result somewhere?
+                    # NOTE: get_metadata won't work here since filters not applied
+                    if (
+                        curr_case_count
+                        := task_info.materialized_test.dry_run().report.counter[
+                            "total"
+                        ]
+                    ):
+                        # XXX: define lb & ub of testcase-count factor?
+                        adjusted_exec_time *= curr_case_count / prev_case_count
+                        self.logger.user_info(
+                            "%s: estimated total execution time %f -> %f "
+                            "(prev total tc: %d, curr total tc: %d)",
+                            uid,
+                            time_info["execution_time"],
+                            adjusted_exec_time,
+                            prev_case_count,
+                            curr_case_count,
+                        )
+                        time_info["execution_time"] = adjusted_exec_time
+                    # XXX: shoutout if curr_case_count is 0?
+
         if num_of_parts:
             if not isinstance(task_info.materialized_test, MultiTest):
                 raise TypeError(
@@ -1010,25 +1046,30 @@ def _calculate_parts_and_weights(
                     )
                     num_of_parts = 1
                 else:
-                    # the setup time shall take no more than 50% of runtime
-                    cap = math.ceil(
-                        time_info["execution_time"]
-                        / auto_part_runtime_limit
-                        * 2
-                    )
+                    if prev_case_count and curr_case_count:
+                        adjust_formula_part = f"""
+                * curr_case_count {curr_case_count}
+                / prev_case_count {prev_case_count}"""
+                    else:
+                        adjust_formula_part = ""
                     formula = f"""
             num_of_parts = math.ceil(
-                time_info["execution_time"] {time_info["execution_time"]}
+                time_info["execution_time"] {time_info["execution_time"]}{adjust_formula_part}
                 / (
                     self.cfg.auto_part_runtime_limit {auto_part_runtime_limit}
                     - time_info["setup_time"] {time_info["setup_time"]}
                     - time_info["teardown_time"] {time_info["teardown_time"]}
                 )
             )
-        """
+"""
+
+                    # the setup time shall take no more than 50% of runtime
+                    cap = math.ceil(
+                        adjusted_exec_time / auto_part_runtime_limit * 2
+                    )
                     try:
                         num_of_parts = math.ceil(
-                            time_info["execution_time"]
+                            adjusted_exec_time
                             / (
                                 auto_part_runtime_limit
                                 - time_info["setup_time"]
@@ -1058,12 +1099,12 @@ def _calculate_parts_and_weights(
             if "weight" not in task_arguments:
                 task_arguments["weight"] = (
                     math.ceil(
-                        (time_info["execution_time"] / num_of_parts)
+                        (adjusted_exec_time / num_of_parts)
                         + time_info["setup_time"]
                         + time_info["teardown_time"]
                     )
                     if time_info
-                    else auto_part_runtime_limit
+                    else int(auto_part_runtime_limit)
                 )
             self.logger.user_info(
                 "%s: parts=%d, weight=%d",
@@ -1085,7 +1126,7 @@ def _calculate_parts_and_weights(
         else:
             if time_info and not task_info.target.weight:
                 task_info.target.weight = math.ceil(
-                    time_info["execution_time"]
+                    adjusted_exec_time
                     + time_info["setup_time"]
                     + time_info["teardown_time"]
                 )

testplan/runnable/interactive/reloader.py

@@ -14,15 +14,19 @@
 from testplan.common.utils import path as path_utils
 from testplan.common.utils import logger
 from testplan.common.utils import strings
+from testplan.common.utils.logger import TESTPLAN_LOGGER
 from testplan.common.utils.package import import_tmp_module
 from testplan.testing.multitest import suite, MultiTest
 
 
-def _patched_find_module(name, path=None):
+def _patched_find_module(name, path=None, logger=None):
     """
     modified from <3.13 branches, to handle namespace packages
     """
 
+    if logger is None:
+        logger = TESTPLAN_LOGGER
+
     # Old imp constants:
     _SEARCH_ERROR = 0
     _PY_SOURCE = 1
@@ -64,6 +68,7 @@ def _patched_find_module(name, path=None):
         ):
             # ModuleFinder.find_module is designed to only return one package dir,
             # while namespace packages can have multiple
+            logger.debug("Skipping namespace package %s under %s", name, path)
             return None, None, ("", "", _NAMESPACE_IGNORED)
 
     if spec.loader is importlib.machinery.BuiltinImporter:
@@ -150,7 +155,7 @@ def __init__(self, extra_deps=None, scheduled_modules=None):
         ) = self._build_dependencies()
 
         # Last recorded reload time for watched modules.
-        self._last_reload_time = {}  # type: Dict[str, float]
+        self._last_reload_time: dict[str, float] = {}
         self._init_time = time.time()
 
     def reload(self, tests, rebuild_dependencies=False):
@@ -663,7 +668,7 @@ def find_module(self, name, path, parent=None):
 
             path = self.path
 
-        return _patched_find_module(name, path)
+        return _patched_find_module(name, path, logger=self.logger)
 
 
 class _ModuleNode:

testplan/testing/multitest/base.py

@@ -1,7 +1,7 @@
 """MultiTest test execution framework."""
 
 import collections.abc
-import concurrent
+import concurrent.futures
 import functools
 import itertools
 import warnings

tests/functional/testplan/runners/pools/test_auto_part.py

@@ -1,5 +1,7 @@
 import os
+import math
 import tempfile
+from itertools import count
 from pathlib import Path
 
 import pytest
@@ -379,3 +381,64 @@ def test_auto_plan_runtime_target():
             assert task.weight == 900
         mockplan.run()
         assert pool.size == 6
+
+
+@pytest.mark.parametrize(
+    "tc, adjusted_exec_t, exp_weight, exp_parts",
+    [
+        (10, 60, 35, 2),
+        (5, 120, 35, 4),
+        (20, 30, 35, 1),
+        (8, 75, 30, 3),
+    ],
+    ids=count(0),
+)
+def test_multitest_weight_adjusted_by_relative_testcase_count(
+    tc, adjusted_exec_t, exp_weight, exp_parts
+):
+    with tempfile.TemporaryDirectory() as runpath:
+        mockplan = TestplanMock(
+            "plan",
+            runpath=runpath,
+            merge_scheduled_parts=True,
+            auto_part_runtime_limit=40,
+            plan_runtime_target=80,
+            runtime_data={
+                "Proj1-suite": {
+                    "execution_time": 60,
+                    "setup_time": 5,
+                    "teardown_time": 0,
+                    "testcase_count": tc,
+                }
+            },
+        )
+        pool = ProcessPool(name="MyPool", size="auto")
+        mockplan.add_resource(pool)
+        current_folder = Path(__file__).resolve().parent
+        assert pool.cfg.runtime_data == {
+            "Proj1-suite": {
+                "execution_time": 60,
+                "setup_time": 5,
+                "teardown_time": 0,
+                "testcase_count": tc,
+            }
+        }
+        # curr tc: 10
+        mockplan.schedule_all(
+            path=current_folder / "discover_tasks",
+            name_pattern=r".*auto_parts_tasks\.py$",
+            resource="MyPool",
+        )
+        assert pool.cfg.runtime_data == {
+            "Proj1-suite": {
+                "execution_time": adjusted_exec_t,
+                "setup_time": 5,
+                "teardown_time": 0,
+                "testcase_count": tc,
+            }
+        }
+        assert len(pool.added_items) == exp_parts
+        for task in pool.added_items.values():
+            assert task.weight == exp_weight
+        mockplan.run()
+        assert pool.size == math.ceil(exp_weight * exp_parts / 80)