Split plan to multiple plans by --max tests

tests/run/max/data/.fmf/version

@@ -0,0 +1 @@
+1

tests/run/max/data/main.fmf

@@ -0,0 +1,18 @@
+/plan:
+    discover:
+        how: shell
+        tests:
+          - name: Test 1
+            test: echo "1"
+          - name: Test 2
+            test: echo "2"
+          - name: Test 3
+            test: echo "3"
+          - name: Test 4
+            test: echo "4"
+          - name: Test 5
+            test: echo "5"
+    provision:
+        how: local
+    execute:
+        how: tmt

tests/run/max/main.fmf

@@ -0,0 +1 @@
+summary: Verify --max option splits plans into smaller plans

tests/run/max/test.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+. /usr/share/beakerlib/beakerlib.sh || exit 1
+
+rlJournalStart
+    rlPhaseStartSetup
+        rlRun "run=\$(mktemp -d)" 0 "Create run directory"
+        rlRun "pushd data"
+    rlPhaseEnd
+
+    rlPhaseStartTest
+        rlRun -s "tmt run -vv --id $run --max 3"
+        rlAssertGrep "Splitting plan to batches of 3 tests." $rlRun_LOG
+        rlAssertGrep "3 tests selected" $rlRun_LOG
+        rlAssertGrep "summary: 3 tests passed" $rlRun_LOG
+        rlAssertGrep "2 tests selected" $rlRun_LOG
+        rlAssertGrep "summary: 2 tests passed" $rlRun_LOG
+    rlPhaseEnd
+
+    rlPhaseStartCleanup
+        rlRun "popd"
+        rlRun "rm -r $run" 0 "Remove run directory"
+    rlPhaseEnd
+rlJournalEnd

tmt/base.py

@@ -43,6 +43,7 @@
 import tmt.lint
 import tmt.log
 import tmt.plugins
+import tmt.plugins.plan_shapers
 import tmt.result
 import tmt.steps
 import tmt.steps.discover
@@ -78,6 +79,7 @@
 
 if TYPE_CHECKING:
     import tmt.cli
+    import tmt.steps.discover
     import tmt.steps.provision.local
 
 
@@ -713,9 +715,10 @@ def __init__(self,
                  tree: Optional['Tree'] = None,
                  parent: Optional[tmt.utils.Common] = None,
                  logger: tmt.log.Logger,
+                 name: Optional[str] = None,
                  **kwargs: Any) -> None:
         """ Initialize the node """
-        super().__init__(node=node, logger=logger, parent=parent, name=node.name, **kwargs)
+        super().__init__(node=node, logger=logger, parent=parent, name=name or node.name, **kwargs)
 
         self.node = node
         self.tree = tree
@@ -1670,11 +1673,17 @@ class Plan(
     # Optional Login instance attached to the plan for easy login in tmt try
     login: Optional[tmt.steps.Login] = None
 
-    # When fetching remote plans we store links between the original
-    # plan with the fmf id and the imported plan with the content.
-    _imported_plan: Optional['Plan'] = field(default=None, internal=True)
+    # When fetching remote plans or splitting plans, we store links
+    # between the original plan with the fmf id and the imported or
+    # derived plans with the content.
     _original_plan: Optional['Plan'] = field(default=None, internal=True)
-    _remote_plan_fmf_id: Optional[FmfId] = field(default=None, internal=True)
+    _original_plan_fmf_id: Optional[FmfId] = field(default=None, internal=True)
+
+    _imported_plan_fmf_id: Optional[FmfId] = field(default=None, internal=True)
+    _imported_plan: Optional['Plan'] = field(default=None, internal=True)
+
+    _derived_plans: list['Plan'] = field(default_factory=list, internal=True)
+    derived_id: Optional[int] = field(default=None, internal=True)
 
     #: Used by steps to mark invocations that have been already applied to
     #: this plan's phases. Needed to avoid the second evaluation in
@@ -1716,11 +1725,12 @@ def __init__(
         # set, incorrect default value is generated, and the field ends up being
         # set to `None`. See https://github.com/teemtee/tmt/issues/2630.
         self._applied_cli_invocations = []
+        self._derived_plans = []
 
         # Check for possible remote plan reference first
         reference = self.node.get(['plan', 'import'])
         if reference is not None:
-            self._remote_plan_fmf_id = FmfId.from_spec(reference)
+            self._imported_plan_fmf_id = FmfId.from_spec(reference)
 
         # Save the run, prepare worktree and plan data directory
         self.my_run = run
@@ -2137,13 +2147,14 @@ def show(self) -> None:
             self._show_additional_keys()
 
         # Show fmf id of the remote plan in verbose mode
-        if (self._original_plan or self._remote_plan_fmf_id) and self.verbosity_level:
+        if (self._original_plan or self._imported_plan_fmf_id) and self.verbosity_level:
             # Pick fmf id from the original plan by default, use the
             # current plan in shallow mode when no plans are fetched.
+
             if self._original_plan is not None:
-                fmf_id = self._original_plan._remote_plan_fmf_id
+                fmf_id = self._original_plan._imported_plan_fmf_id
             else:
-                fmf_id = self._remote_plan_fmf_id
+                fmf_id = self._imported_plan_fmf_id
 
             echo(tmt.utils.format('import', '', key_color='blue'))
             assert fmf_id is not None  # narrow type
@@ -2413,14 +2424,21 @@ def go(self) -> None:
         try:
             for step in self.steps(skip=['finish']):
                 step.go()
-                # Finish plan if no tests found (except dry mode)
-                if (isinstance(step, tmt.steps.discover.Discover) and not step.tests()
-                        and not self.is_dry_run and not step.extract_tests_later):
-                    step.info(
-                        'warning', 'No tests found, finishing plan.',
-                        color='yellow', shift=1)
-                    abort = True
-                    return
+
+                if isinstance(step, tmt.steps.discover.Discover):
+                    tests = step.tests()
+
+                    # Finish plan if no tests found (except dry mode)
+                    if not tests and not self.is_dry_run and not step.extract_tests_later:
+                        step.info(
+                            'warning', 'No tests found, finishing plan.',
+                            color='yellow', shift=1)
+                        abort = True
+                        return
+
+                    if self.my_run and self.reshape(tests):
+                        return
+
                 # Source the plan environment file after prepare and execute step
                 if isinstance(step, (tmt.steps.prepare.Prepare, tmt.steps.execute.Execute)):
                     self._source_plan_environment_file()
@@ -2456,7 +2474,7 @@ def _export(
     @property
     def is_remote_plan_reference(self) -> bool:
         """ Check whether the plan is a remote plan reference """
-        return self._remote_plan_fmf_id is not None
+        return self._imported_plan_fmf_id is not None
 
     def import_plan(self) -> Optional['Plan']:
         """ Import plan from a remote repository, return a Plan instance """
@@ -2466,8 +2484,8 @@ def import_plan(self) -> Optional['Plan']:
         if self._imported_plan:
             return self._imported_plan
 
-        assert self._remote_plan_fmf_id is not None  # narrow type
-        plan_id = self._remote_plan_fmf_id
+        assert self._imported_plan_fmf_id is not None  # narrow type
+        plan_id = self._imported_plan_fmf_id
         self.debug(f"Import remote plan '{plan_id.name}' from '{plan_id.url}'.", level=3)
 
         # Clone the whole git repository if executing tests (run is attached)
@@ -2559,12 +2577,37 @@ def import_plan(self) -> Optional['Plan']:
         # Create the plan object, save links between both plans
         self._imported_plan = Plan(node=node, run=self.my_run, logger=self._logger)
         self._imported_plan._original_plan = self
+        self._imported_plan._original_plan_fmf_id = self.fmf_id
 
         with self.environment.as_environ():
             expand_node_data(node.data, self._fmf_context)
 
         return self._imported_plan
 
+    def derive_plan(self, derived_id: int, tests: dict[str, list[Test]]) -> 'Plan':
+        derived_plan = Plan(
+            node=self.node,
+            run=self.my_run,
+            logger=self._logger,
+            name=f'{self.name}.{derived_id}')
+
+        derived_plan._original_plan = self
+        derived_plan._original_plan_fmf_id = self.fmf_id
+        self._derived_plans.append(derived_plan)
+
+        derived_plan.discover._tests = tests
+        derived_plan.discover.status('done')
+
+        assert self.discover.workdir is not None
+        assert derived_plan.discover.workdir is not None
+
+        shutil.copytree(self.discover.workdir, derived_plan.discover.workdir, dirs_exist_ok=True)
+
+        for step_name in tmt.steps.STEPS:
+            getattr(derived_plan, step_name).save()
+
+        return derived_plan
+
     def prune(self) -> None:
         """ Remove all uninteresting files from the plan workdir """
 
@@ -2582,6 +2625,23 @@ def prune(self) -> None:
         for step in self.steps(enabled_only=False):
             step.prune(logger=step._logger)
 
+    def reshape(self, tests: list['tmt.steps.discover.TestAddress']) -> bool:
+        for shaper_id in tmt.plugins.plan_shapers._PLAN_SHAPER_PLUGIN_REGISTRY.iter_plugin_ids():
+            shaper = tmt.plugins.plan_shapers._PLAN_SHAPER_PLUGIN_REGISTRY.get_plugin(shaper_id)
+
+            assert shaper is not None  # narrow type
+
+            if not shaper.check(self, tests):
+                self.debug(f"Plan shaper '{shaper_id}' not applicable.")
+                continue
+
+            if self.my_run:
+                self.my_run.swap_plans(self, *shaper.apply(self, tests))
+
+            return True
+
+        return False
+
 
 class StoryPriority(enum.Enum):
     MUST_HAVE = 'must have'
@@ -3549,14 +3609,45 @@ def load(self) -> None:
         self.remove = self.remove or data.remove
         self.debug(f"Remove workdir when finished: {self.remove}", level=3)
 
-    @property
-    def plans(self) -> list[Plan]:
+    @functools.cached_property
+    def plans(self) -> Sequence[Plan]:
         """ Test plans for execution """
         if self._plans is None:
             assert self.tree is not None  # narrow type
             self._plans = self.tree.plans(run=self, filters=['enabled:true'])
         return self._plans
 
+    @functools.cached_property
+    def plan_queue(self) -> Sequence[Plan]:
+        """
+        A list of plans remaining to be executed.
+
+        It is being populated via :py:attr:`plans`, but eventually,
+        :py:meth:`go` will remove plans from it as they get processed.
+        :py:attr:`plans` will remain untouched and will represent all
+        plans collected.
+        """
+
+        return self.plans[:]
+
+    def swap_plans(self, plan: Plan, *others: Plan) -> None:
+        """
+        Replace given plan with one or more plans.
+
+        :param plan: a plan to remove.
+        :param others: plans to put into the queue instead of ``plans``.
+        """
+
+        plans = cast(list[Plan], self.plans)
+        plan_queue = cast(list[Plan], self.plan_queue)
+
+        if plan in plan_queue:
+            plan_queue.remove(plan)
+            plans.remove(plan)
+
+        plan_queue.extend(others)
+        plans.extend(others)
+
     def finish(self) -> None:
         """ Check overall results, return appropriate exit code """
         # We get interesting results only if execute or prepare step is enabled
@@ -3720,7 +3811,9 @@ def go(self) -> None:
         # Iterate over plans
         crashed_plans: list[tuple[Plan, Exception]] = []
 
-        for plan in self.plans:
+        while self.plan_queue:
+            plan = cast(list[Plan], self.plan_queue).pop(0)
+
             try:
                 plan.go()
 

tmt/cli.py

@@ -23,6 +23,7 @@
 import tmt.log
 import tmt.options
 import tmt.plugins
+import tmt.plugins.plan_shapers
 import tmt.steps
 import tmt.templates
 import tmt.trying
@@ -461,6 +462,10 @@ def run(context: Context, id_: Optional[str], **kwargs: Any) -> None:
     context.obj.run = run
 
 
+for plugin_class in tmt.plugins.plan_shapers._PLAN_SHAPER_PLUGIN_REGISTRY.iter_plugins():
+    run = create_options_decorator(plugin_class.run_options())(run)
+
+
 # Steps options
 run.add_command(tmt.steps.discover.DiscoverPlugin.command())
 run.add_command(tmt.steps.provision.ProvisionPlugin.command())

tmt/plugins/__init__.py

@@ -73,6 +73,7 @@ def _discover_packages() -> list[tuple[str, Path]]:
         ('tmt.frameworks', Path('frameworks')),
         ('tmt.checks', Path('checks')),
         ('tmt.package_managers', Path('package_managers')),
+        ('tmt.plugins.plan_shapers', Path('plugins/plan_shapers')),
         ]
 
 

tmt/plugins/plan_shapers/__init__.py

@@ -0,0 +1,67 @@
+from collections.abc import Iterator
+from typing import TYPE_CHECKING, Any, Callable
+
+import tmt.log
+import tmt.utils
+from tmt.plugins import PluginRegistry
+
+if TYPE_CHECKING:
+    from tmt.base import Plan, Test
+    from tmt.options import ClickOptionDecoratorType
+
+
+PlanShaperClass = type['PlanShaper']
+
+
+_PLAN_SHAPER_PLUGIN_REGISTRY: PluginRegistry[PlanShaperClass] = PluginRegistry()
+
+
+def provides_plan_shaper(
+        shaper: str) -> Callable[[PlanShaperClass], PlanShaperClass]:
+    """
+    A decorator for registering plan shaper plugins.
+
+    Decorate a plan shaper plugin class to register a plan shaper.
+    """
+
+    def _provides_plan_shaper(plan_shaper_cls: PlanShaperClass) -> PlanShaperClass:
+        _PLAN_SHAPER_PLUGIN_REGISTRY.register_plugin(
+            plugin_id=shaper,
+            plugin=plan_shaper_cls,
+            logger=tmt.log.Logger.get_bootstrap_logger())
+
+        return plan_shaper_cls
+
+    return _provides_plan_shaper
+
+
+class PlanShaper(tmt.utils.Common):
+    """ A base class for plan shaper plugins """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+
+    @classmethod
+    def run_options(cls) -> list['ClickOptionDecoratorType']:
+        """ Return additional options for ``tmt run`` """
+
+        raise NotImplementedError
+
+    @classmethod
+    def check(cls, plan: 'Plan', tests: list[tuple[str, 'Test']]) -> bool:
+        """ Check whether this shaper should be applied to the given plan """
+
+        raise NotImplementedError
+
+    @classmethod
+    def apply(
+            cls,
+            plan: 'Plan',
+            tests: list[tuple[str, 'Test']]) -> Iterator['Plan']:
+        """
+        Apply the shaper to a given plan and a set of tests.
+
+        :returns: a sequence of plans replacing the original plan.
+        """
+
+        raise NotImplementedError