Warn on duplicate commands in the same namespace

stephenfin · stephenfin · commit a0eee2b08db2 · 2025-11-28T15:44:35.000Z
We intentionally do not use stevedore's conflict resolution
functionality here because, unlikely stevedore, we don't care what
extension namespace a command is defined in: if it's got the same
command name then it's a duplicate regardless.

Change-Id: I20f4c75fc6538615e7d1be5fc4e91568d4e8b193
Signed-off-by: Stephen Finucane &lt;sfinucan@redhat.com&gt;
diff --git a/cliff/commandmanager.py b/cliff/commandmanager.py
@@ -53,6 +53,13 @@ def __init__(
     def load(self) -> type[command.Command]:
         return self.command_class
 
+    @property
+    def value(self) -> str:
+        # fake entry point target for logging purposes
+        return ':'.join(
+            [self.command_class.__module__, self.command_class.__name__]
+        )
+
 
 EntryPointT: TypeAlias = EntryPointWrapper | importlib.metadata.EntryPoint
 
@@ -64,15 +71,23 @@ class CommandManager:
         plugins to be loaded. For example, ``'cliff.formatter.list'``.
     :param convert_underscores: Whether cliff should convert underscores to
         spaces in entry_point commands.
+    :param ignored_modules: A list of module names to ignore when loading
+        commands. This will be matched partial, so be as specific as needed.
     """
 
     def __init__(
-        self, namespace: str, convert_underscores: bool = True
+        self,
+        namespace: str | None,
+        convert_underscores: bool = True,
+        *,
+        ignored_modules: collections.abc.Iterable[str] | None = None,
     ) -> None:
-        self.commands: dict[str, EntryPointT] = {}
-        self._legacy: dict[str, str] = {}
         self.namespace = namespace
         self.convert_underscores = convert_underscores
+        self.ignored_modules = ignored_modules or ()
+
+        self.commands: dict[str, EntryPointT] = {}
+        self._legacy: dict[str, str] = {}
         self.group_list: list[str] = []
         self._load_commands()
 
@@ -81,19 +96,79 @@ def _load_commands(self) -> None:
         if self.namespace:
             self.load_commands(self.namespace)
 
+    @staticmethod
+    def _is_module_ignored(
+        module_name: str, ignored_modules: collections.abc.Iterable[str]
+    ) -> bool:
+        # given module_name = 'foo.bar.baz:wow', we expect to match any of
+        # the following ignores: foo.bar.baz:wow, foo.bar.baz, foo.bar, foo
+        while True:
+            if module_name in ignored_modules:
+                return True
+
+            split_index = max(module_name.rfind(':'), module_name.rfind('.'))
+            if split_index == -1:
+                break
+
+            module_name = module_name[:split_index]
+
+        return False
+
     def load_commands(self, namespace: str) -> None:
-        """Load all the commands from an entrypoint"""
+        """Load all the commands from an entrypoint
+
+        :param namespace: The namespace to load commands from.
+        :returns: None
+        """
         self.group_list.append(namespace)
         em: stevedore.ExtensionManager[command.Command]
+        # note that we don't invoke stevedore's conflict resolver functionality
+        # because that is namespace specific and we care about conflicts
+        # regardless of the namespace
         em = stevedore.ExtensionManager(namespace)
-        for ep in em:
-            LOG.debug('found command %r', ep.name)
+        for ext in em:
+            LOG.debug('found command %r', ext.name)
+
+            if self._is_module_ignored(ext.module_name, self.ignored_modules):
+                LOG.debug(
+                    'extension found in ignored module %r: skipping',
+                    ext.module_name,
+                )
+                continue
+
             cmd_name = (
-                ep.name.replace('_', ' ')
+                ext.name.replace('_', ' ')
                 if self.convert_underscores
-                else ep.name
+                else ext.name
             )
-            self.commands[cmd_name] = ep.entry_point
+
+            if cmd_name in self.commands:
+                # Attention, programmers: If you arrived here attempting to
+                # resolve a warning in your application then you have a command
+                # with the same name either defined more than once in the same
+                # application (a typo?) or in multiple packages (for example,
+                # a package that adds plugins to your applications). The latter
+                # can often happen if you e.g. move a command from a plugin to
+                # the core application. In this situation, you should add the
+                # old location to 'ignored_modules' and the plugin will now be
+                # ignored.
+                LOG.warning(
+                    'found duplicate implementations of the %(name)r command '
+                    'in the following modules: %(modules)s: this is likely '
+                    'programmer error and should be reported as a bug to the '
+                    'relevant project(s)',
+                    {
+                        'name': cmd_name,
+                        'modules': ', '.join(
+                            [
+                                self.commands[cmd_name].value,
+                                ext.entry_point.value,
+                            ]
+                        ),
+                    },
+                )
+
+            self.commands[cmd_name] = ext.entry_point
 
     def __iter__(
         self,
diff --git a/cliff/tests/base.py b/cliff/tests/base.py
@@ -19,9 +19,13 @@
 class TestBase(testtools.TestCase):
     def setUp(self):
         super().setUp()
+
         self._stdout_fixture = fixtures.StringStream('stdout')
         self.stdout = self.useFixture(self._stdout_fixture).stream
         self.useFixture(fixtures.MonkeyPatch('sys.stdout', self.stdout))
+
         self._stderr_fixture = fixtures.StringStream('stderr')
         self.stderr = self.useFixture(self._stderr_fixture).stream
         self.useFixture(fixtures.MonkeyPatch('sys.stderr', self.stderr))
+
+        self.logger = self.useFixture(fixtures.FakeLogger('cliff'))
diff --git a/cliff/tests/test_commandmanager.py b/cliff/tests/test_commandmanager.py
@@ -10,9 +10,10 @@
 #  License for the specific language governing permissions and limitations
 #  under the License.
 
-import testscenarios
 from unittest import mock
 
+import testscenarios
+
 from cliff import command
 from cliff import commandmanager
 from cliff.tests import base
@@ -113,20 +114,23 @@ def test_intersected_commands(self):
 
 class TestLoad(base.TestBase):
     def test_load_commands(self):
-        testcmd = mock.Mock(name='testcmd')
-        testcmd.name.replace.return_value = 'test'
+        testcmd = mock.Mock()
+        testcmd.name = 'test_cmd'
+        testcmd.module_name = 'module.a'
         mock_get_group_all = mock.Mock(return_value=[testcmd])
         with mock.patch(
             'stevedore.ExtensionManager', mock_get_group_all
         ) as mock_manager:
             mgr = commandmanager.CommandManager('test')
-            mock_manager.assert_called_once_with('test')
-            names = [n for n, v in mgr]
-            self.assertEqual(['test'], names)
+
+        mock_manager.assert_called_once_with('test')
+        names = [n for n, v in mgr]
+        self.assertEqual(['test cmd'], names)
 
     def test_load_commands_keep_underscores(self):
         testcmd = mock.Mock()
         testcmd.name = 'test_cmd'
+        testcmd.module_name = 'module.a'
         mock_get_group_all = mock.Mock(return_value=[testcmd])
         with mock.patch(
             'stevedore.ExtensionManager', mock_get_group_all
@@ -135,24 +139,58 @@ def test_load_commands_keep_underscores(self):
                 'test',
                 convert_underscores=False,
             )
-            mock_manager.assert_called_once_with('test')
-            names = [n for n, v in mgr]
-            self.assertEqual(['test_cmd'], names)
 
-    def test_load_commands_replace_underscores(self):
-        testcmd = mock.Mock()
-        testcmd.name = 'test_cmd'
-        mock_get_group_all = mock.Mock(return_value=[testcmd])
+        mock_manager.assert_called_once_with('test')
+        names = [n for n, v in mgr]
+        self.assertEqual(['test_cmd'], names)
+
+    def test_load_commands_ignore_modules(self):
+        testcmd_normal = mock.Mock()
+        testcmd_normal.name = 'normal_cmd'
+        testcmd_normal.module_name = 'normal.module'
+        testcmd_ignored = mock.Mock()
+        testcmd_ignored.name = 'ignored_cmd'
+        testcmd_ignored.module_name = 'ignored.module'
+        mock_get_group_all = mock.Mock(
+            return_value=[testcmd_normal, testcmd_ignored]
+        )
         with mock.patch(
             'stevedore.ExtensionManager', mock_get_group_all
         ) as mock_manager:
             mgr = commandmanager.CommandManager(
                 'test',
-                convert_underscores=True,
+                ignored_modules=['ignored.module'],
             )
-            mock_manager.assert_called_once_with('test')
-            names = [n for n, v in mgr]
-            self.assertEqual(['test cmd'], names)
+
+        mock_manager.assert_called_once_with('test')
+        names = [n for n, v in mgr]
+        self.assertEqual(['normal cmd'], names)
+
+    def test_load_commands_duplicates(self):
+        testcmd_a = mock.Mock()
+        testcmd_a.name = 'test_cmd'
+        testcmd_a.module_name = 'module.a'
+        testcmd_a.entry_point.value = 'module.a:TestCmd'
+        testcmd_b = mock.Mock()
+        testcmd_b.name = 'test_cmd'
+        testcmd_b.module_name = 'module.b'
+        testcmd_b.entry_point.value = 'module.b:TestCmd'
+        mock_get_group_all = mock.Mock(return_value=[testcmd_a, testcmd_b])
+        with mock.patch(
+            'stevedore.ExtensionManager', mock_get_group_all
+        ) as mock_manager:
+            mgr = commandmanager.CommandManager('test')
+
+        mock_manager.assert_called_once_with('test')
+        names = [n for n, v in mgr]
+        commands = [v for _, v in mgr]
+        self.assertEqual(['test cmd'], names)
+        # we should have ended up with the second command
+        self.assertEqual([testcmd_b.entry_point], commands)
+        self.assertIn(
+            "found duplicate implementations of the 'test cmd' command",
+            self.logger.output,
+        )
 
 
 class FauxCommand(command.Command):
@@ -216,6 +254,34 @@ def test(self):
         self.assertFalse(remaining)
 
 
+class TestIsModuleIgnored(base.TestBase):
+    def test_match(self):
+        result = commandmanager.CommandManager._is_module_ignored(
+            'foo.bar.baz', ['foo.bar.baz', 'other.module']
+        )
+        self.assertTrue(result)
+        result = commandmanager.CommandManager._is_module_ignored(
+            'foo.bar.baz', ['foo.bar', 'other.module']
+        )
+        self.assertTrue(result)
+        result = commandmanager.CommandManager._is_module_ignored(
+            'foo.bar.baz', ['foo', 'other.module']
+        )
+        self.assertTrue(result)
+
+    def test_no_match(self):
+        result = commandmanager.CommandManager._is_module_ignored(
+            'foo.bar.baz', ['other.module', 'another.package']
+        )
+        self.assertFalse(result)
+
+    def test_no_ignores(self):
+        result = commandmanager.CommandManager._is_module_ignored(
+            'foo.bar.baz', []
+        )
+        self.assertFalse(result)
+
+
 class TestGetByPartialName(base.TestBase):
     def setUp(self):
         super().setUp()
@@ -334,8 +400,10 @@ def test_get_command_groups(self):
     def test_get_command_names(self):
         mock_cmd_one = mock.Mock()
         mock_cmd_one.name = 'one'
+        mock_cmd_one.module_name = 'module.a'
         mock_cmd_two = mock.Mock()
         mock_cmd_two.name = 'cmd two'
+        mock_cmd_two.module_name = 'module.b'
         mock_get_group_all = mock.Mock(
             return_value=[mock_cmd_one, mock_cmd_two],
         )
@@ -345,5 +413,9 @@ def test_get_command_names(self):
         ) as mock_manager:
             mgr = commandmanager.CommandManager('test')
             mock_manager.assert_called_once_with('test')
+
             cmds = mgr.get_command_names('test')
+            mock_manager.assert_has_calls(
+                [mock.call('test'), mock.call('test')]
+            )
             self.assertEqual(['one', 'cmd two'], cmds)
diff --git a/releasenotes/notes/conflict-resolution-21bcb5f9a6c19d7f.yaml b/releasenotes/notes/conflict-resolution-21bcb5f9a6c19d7f.yaml
@@ -0,0 +1,12 @@
+---
+features:
+  - |
+    ``cliff.commandmanager.CommandManager`` now accepts an optional
+    ``ignored_modules`` argument, which can be used to indicate modules that
+    should be skipped when loading commands. This can be useful if e.g. moving
+    a command from a plugin to the core application, to ensure the version in
+    the core application is always preferred.
+upgrade:
+  - |
+    cliff will now warn if duplicates command implementations are found, rather
+    than silently ignoring them.
