Allow non-exact matching for ids in cli subcommands (New)

checkbox-ng/checkbox_ng/launcher/subcommands.py

@@ -1075,6 +1075,11 @@
         parser.add_argument(
             "-m", "--message", help=_("submission description")
         )
+        parser.add_argument(
+            "--exact",
+            action="store_true",
+            help="only expand the test-plan fully qualified ID that exactly matches",
+        )
 
     @property
     def C(self):
@@ -1097,6 +1102,23 @@
             and not self.ctx.args.non_interactive
         )
 
+    def _get_relevant_units(self, patterns, exact=False):
+        if exact:
+            return patterns
+        providers = self.sa.get_selected_providers()
+        root = Explorer(providers).get_object_tree()
+        # here handle the patterns one by one to not change the order
+        matching_units = [
+            root.find_children_by_name([pattern]) for pattern in patterns
+        ]
+        all_ids = [
+            [pattern] if not matches else [match.name for match in matches]
+            for matching_unit in matching_units
+            for (pattern, matches) in matching_unit.items()
+        ]
+        all_ids = [id for all_id in all_ids for id in all_id]
+        return all_ids
+
     def invoked(self, ctx):
         try:
             self._C = Colorizer()
@@ -1110,7 +1132,9 @@
             )
             tps = self.sa.get_test_plans()
             self._configure_report()
-            selection = ctx.args.PATTERN
+            selection = self._get_relevant_units(
+                ctx.args.PATTERN, ctx.args.exact
+            )
             submission_message = self.ctx.args.message
             if len(selection) == 1 and selection[0] in tps:
                 self.ctx.sa.update_app_blob(
@@ -1133,7 +1157,7 @@
                 self._run_jobs(self.sa.get_dynamic_todo_list())
                 # there might have been new jobs instantiated
                 while True:
-                    self.sa.hand_pick_jobs(ctx.args.PATTERN)
+                    self.sa.hand_pick_jobs(selection)
                     todos = self.sa.get_dynamic_todo_list()
                     if not todos:
                         break
@@ -1325,14 +1349,21 @@
             "order. To see the execution order, please use the "
             "'list-bootstrapped' command instead."
         )
-        parser.add_argument("TEST_PLAN", help=_("test-plan id to expand"))
+        parser.add_argument(
+            "TEST_PLAN", help=_("test-plan ID or fully qualified ID to expand")
+        )
         parser.add_argument(
             "-f",
             "--format",
             type=str,
             default="text",
             help=_("output format: 'text' or 'json' (default: %(default)s)"),
         )
+        parser.add_argument(
+            "--exact",
+            action="store_true",
+            help="only expand the test-plan that exactly matches the fully qualified ID",
+        )
 
     def _get_relevant_manifest_units(self, jobs_and_templates_list):
         """
@@ -1372,9 +1403,12 @@
         session_title = "checkbox-expand-{}".format(ctx.args.TEST_PLAN)
         self.sa.start_new_session(session_title)
         tps = self.sa.get_test_plans()
-        if ctx.args.TEST_PLAN not in tps:
+        testplan_id = get_testplan_id_by_id(
+            tps, ctx.args.TEST_PLAN, self.sa, ctx.args.exact
+        )
+        if testplan_id not in tps:
             raise SystemExit("Test plan not found")
-        self.sa.select_test_plan(ctx.args.TEST_PLAN)
+        self.sa.select_test_plan(testplan_id)
         all_jobs_and_templates = [
             unit
             for unit in self.sa._context.state.unit_list
@@ -1447,6 +1481,11 @@
         return self.ctx.sa
 
     def register_arguments(self, parser):
+        parser.add_argument(
+            "--exact",
+            action="store_true",
+            help="only bootstrap test-plan that exactly match fully qualified ID",
+        )
         parser.add_argument("TEST_PLAN", help=_("test-plan id to bootstrap"))
         parser.add_argument(
             "-f",
@@ -1464,10 +1503,14 @@
     def invoked(self, ctx):
         self.ctx = ctx
         self.sa.start_new_session("checkbox-listing-ephemeral")
+
         tps = self.sa.get_test_plans()
-        if ctx.args.TEST_PLAN not in tps:
+        testplan_id = get_testplan_id_by_id(
+            tps, ctx.args.TEST_PLAN, self.sa, ctx.args.exact
+        )
+        if testplan_id not in tps:
             raise SystemExit("Test plan not found")
-        self.sa.select_test_plan(ctx.args.TEST_PLAN)
+        self.sa.select_test_plan(testplan_id)
         self.sa.bootstrap()
         jobs = []
         for job in self.sa.get_static_todo_list():
@@ -1539,6 +1582,35 @@
         print(path)
 
 
+def get_testplan_id_by_id(tps, testplan_id, sa, exact=False):
+    """
+    Searches for a testplan that matches the given testplan_id
+
+    The input id may not match the testplan id because it is missing the
+    namespace. When the search is not exact, this searches any test plan that
+    has the same id ignoring the namespace.
+    """
+    if exact:
+        return testplan_id
+    if testplan_id in tps:
+        # no need to search for the testplan id
+        return testplan_id
+    providers = sa.get_selected_providers()
+    root = Explorer(providers).get_object_tree()
+
+    relevant = root.find_children_by_name([testplan_id], exact).values()
+    relevant = [unit for units in relevant for unit in units]
+    if len(relevant) > 1:
+        raise SystemExit(
+            "More than one testplan match the id {}. Use either:\n- {}".format(
+                testplan_id, "\n- ".join(unit.name for unit in relevant)
+            )
+        )
+    if relevant:
+        return relevant[0].name
+    return testplan_id  # parent will fail
+
+
 def get_all_jobs(sa):
     providers = sa.get_selected_providers()
     root = Explorer(providers).get_object_tree()
@@ -1609,18 +1681,28 @@
         parser.add_argument(
             "IDs", nargs="+", help=_("Show the definitions of objects")
         )
+        parser.add_argument(
+            "--exact",
+            action="store_true",
+            help=_(
+                "Only show units that exactly match the fully qualified ID"
+            ),
+        )
 
     def invoked(self, ctx):
         providers = ctx.sa.get_selected_providers()
         self._searched_names = ctx.args.IDs
         root = Explorer(providers).get_object_tree()
-        self._traverse_obj_tree(root)
+        relevant = root.find_children_by_name(ctx.args.IDs, ctx.args.exact)
 
-    def _traverse_obj_tree(self, obj):
-        if obj.name in self._searched_names:
-            self._print_obj(obj)
-        for child in obj.children:
-            self._traverse_obj_tree(child)
+        failed = [id for id, founds in relevant.items() if not founds]
+        to_prints = [unit for units in relevant.values() for unit in units]
+
+        for to_print in to_prints:
+            self._print_obj(to_print)
+
+        if failed:
+            raise SystemExit("Failed to find: {}".format(", ".join(failed)))
 
     def _print_obj(self, obj):
         if "origin" in obj.attrs:

checkbox-ng/checkbox_ng/launcher/test_subcommands.py

@@ -30,6 +30,7 @@
 from plainbox.impl.unit.template import TemplateUnit
 
 from checkbox_ng.launcher.subcommands import (
+    Run,
     Expand,
     List,
     Launcher,
@@ -39,6 +40,7 @@
     IJobResult,
     request_comment,
     generate_resume_candidate_description,
+    get_testplan_id_by_id,
     print_objs,
 )
 
@@ -1098,3 +1100,80 @@ def test_generate_resume_candidate_description(self):
         self.assertIn("123", description)
         self.assertIn("Title", description)
         self.assertIn("Test", description)
+
+    @patch("checkbox_ng.launcher.subcommands.Explorer")
+    def test_multiple_relevant_found_raises_system_exit(
+        self, mock_explorer_class
+    ):
+        mock_unit1 = MagicMock()
+        mock_unit1.name = "namespace1::some"
+        mock_unit2 = MagicMock()
+        mock_unit2.name = "namespace2::some"
+
+        mock_root = MagicMock()
+        mock_root.find_children_by_name.return_value = {
+            "key": [mock_unit1, mock_unit2]
+        }
+        mock_explorer_instance = mock_explorer_class()
+        mock_explorer_instance.get_object_tree.return_value = mock_root
+
+        with self.assertRaises(SystemExit):
+            get_testplan_id_by_id(
+                ["namespace1::some", "namespace2::some"],
+                "some",
+                MagicMock(),
+                exact=False,
+            )
+
+    @patch("checkbox_ng.launcher.subcommands.Explorer")
+    def test_single_relevant_found_returns_name(self, mock_explorer_class):
+        mock_unit = MagicMock()
+        mock_unit.name = "namespace1::some"
+
+        mock_root = MagicMock()
+        mock_root.find_children_by_name.return_value = {"some": [mock_unit]}
+        mock_explorer_instance = mock_explorer_class()
+        mock_explorer_instance.get_object_tree.return_value = mock_root
+
+        result = get_testplan_id_by_id(
+            ["namespace1::some"], "some", MagicMock(), exact=False
+        )
+        self.assertEqual(result, "namespace1::some")
+
+    @patch("checkbox_ng.launcher.subcommands.Explorer")
+    def test_no_relevant_found_returns_original_id(self, mock_explorer_class):
+        mock_root = MagicMock()
+        mock_root.find_children_by_name.return_value = {}
+        mock_explorer_instance = mock_explorer_class()
+        mock_explorer_instance.get_object_tree.return_value = mock_root
+
+        result = get_testplan_id_by_id([], "some", MagicMock(), exact=False)
+        self.assertEqual(result, "some")
+
+
+class TestRun(TestCase):
+    @patch("checkbox_ng.launcher.subcommands.Explorer")
+    def test__get_relevant_units(self, explorer_mock):
+        self_mock = MagicMock()
+        root = explorer_mock().get_object_tree()
+        should_find = [
+            "com.canonical.certification::some",
+            "2021.com.canonica.certification::some",
+        ]
+
+        def find_children_by_name(pattern):
+            if pattern == ["some"]:
+                to_r = [MagicMock(), MagicMock()]
+                to_r[0].name = should_find[0]
+                to_r[1].name = should_find[1]
+                return {"some": to_r}
+            return {x: [] for x in pattern}
+
+        root.find_children_by_name = find_children_by_name
+        found_ids = Run._get_relevant_units(
+            self_mock, ["other2.*", "some", "other1.*"], exact=False
+        )
+
+        # we expect the relevant unit function to leave unfound values the same
+        # and all in the same order
+        self.assertEqual(found_ids, ["other2.*", *should_find, "other1.*"])

checkbox-ng/plainbox/impl/highlevel.py

@@ -23,6 +23,7 @@
 """
 
 from collections import OrderedDict
+import operator
 import logging
 
 from plainbox.impl.session.storage import WellKnownDirsHelper
@@ -116,6 +117,26 @@ def attrs(self):
         """
         return self._attrs
 
+    def find_children_by_name(self, names: list, exact=False):
+        to_explore = [self]
+        name_matches = {name: [] for name in names}
+
+        def match_f(a, b):
+            # a is namespace::id, b may not contain namespace
+            return a == b or a.split("::", 1)[-1] == b
+
+        if exact:
+            match_f = operator.eq
+
+        while to_explore:
+            node = to_explore.pop()
+            matching = [name for name in names if match_f(node.name, name)]
+            for match in matching:
+                name_matches[match].append(node)
+            to_explore += node.children
+
+        return name_matches
+
 
 class Explorer:
     """

checkbox-ng/plainbox/impl/test_highlevel.py

@@ -18,7 +18,7 @@
 
 from unittest import TestCase
 
-from plainbox.impl.highlevel import Explorer
+from plainbox.impl.highlevel import Explorer, PlainBoxObject
 from plainbox.impl.unit.template import TemplateUnit
 
 
@@ -42,3 +42,51 @@ def test_template_to_obj__with_template_id(self):
         explorer = Explorer()
         obj = explorer._template_to_obj(template)
         self.assertEqual(obj.name, "template-id")
+
+
+class TestPlainBoxObject(TestCase):
+    def test_exact_match(self):
+        pbo = PlainBoxObject(None, name="com.canonical.certification::abc")
+        found = pbo.find_children_by_name(["abc"], exact=True)
+        self.assertEqual(list(found.values()), [[]])
+
+        found = pbo.find_children_by_name(
+            ["com.canonical.certification::abc"], exact=True
+        )
+        self.assertEqual(list(found.values()), [[pbo]])
+
+    def test_non_exact_match(self):
+        pbo = PlainBoxObject(None, name="com.canonical.certification::abc")
+        found = pbo.find_children_by_name(["abc"], exact=False)
+        self.assertEqual(list(found.values()), [[pbo]])
+
+        found = pbo.find_children_by_name(
+            ["com.canonical.certification::abc"], exact=False
+        )
+        self.assertEqual(list(found.values()), [[pbo]])
+
+    def test_explore_tree(self):
+        target = PlainBoxObject(
+            None, name="com.canonical.certification::target"
+        )
+        target1 = PlainBoxObject(
+            None, name="com.canonical.certification::target1"
+        )
+        sibiling_unrelated = PlainBoxObject(
+            None, name="com.canonical.certification::not-target"
+        )
+        parent = PlainBoxObject(
+            None,
+            name="com.canonical.certification::parent",
+            children=[target, sibiling_unrelated],
+        )
+        gparent = PlainBoxObject(
+            None,
+            name="com.canonical.certification::gparent",
+            children=[parent, target1],
+        )
+
+        found = gparent.find_children_by_name(["target", "target1"])
+
+        self.assertEqual(found["target"], [target])
+        self.assertEqual(found["target1"], [target1])