Merge main and resolve upstream integration conflicts

tests/common/test_cli_plugins.py

@@ -1,11 +1,15 @@
+import shlex
 from types import SimpleNamespace
 
 import click
 from click.testing import CliRunner
 
+from mvt.cli import cli as mvt_cli
 from mvt.common.cli_plugins import (
     ANDROID_CLI_PLUGIN_GROUP,
     IOS_CLI_PLUGIN_GROUP,
+    MVT_CUSTOM_COMMANDS_ENV,
+    NEUTRAL_CLI_PLUGIN_GROUP,
     BrokenPluginCommand,
     load_cli_commands_option,
     register_cli_commands_from_path,
@@ -14,6 +18,9 @@ from mvt.common.cli_plugins import (
 )
 
 
+# Keep the banner of the mvt group callback from checking for updates online.
+OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
+
 COMMAND_TEMPLATE = """
 import click
 
@@ -104,7 +111,7 @@ def test_load_command_option_supports_folders_and_repeated_paths(tmp_path):
 def test_loaded_command_participates_in_shell_completion(tmp_path):
     command_path = _write_command(tmp_path / "hello.py", "hello")
     group = _make_group()
-    words = f"group --load-command {command_path} he"
+    words = f"group --load-command {shlex.quote(str(command_path))} he"
 
     result = CliRunner().invoke(
         group,
@@ -345,7 +352,11 @@ def test_platform_entry_point_groups_and_environment_paths_are_separate(
     def entry_points(*, group):
         if group == IOS_CLI_PLUGIN_GROUP:
             return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
-        return [_entry_point("android-package", "android_plugin:cli", android_package)]
+        if group == ANDROID_CLI_PLUGIN_GROUP:
+            return [
+                _entry_point("android-package", "android_plugin:cli", android_package)
+            ]
+        return []
 
     monkeypatch.setattr(
         "mvt.common.cli_plugins.importlib.metadata.entry_points",
@@ -369,3 +380,187 @@ def test_platform_entry_point_groups_and_environment_paths_are_separate(
 
     assert set(ios_group.commands) == {"ios-file", "ios-package"}
     assert set(android_group.commands) == {"android-file", "android-package"}
+
+
+def test_neutral_entry_point_group_is_not_registered_on_the_platform_clis(
+    monkeypatch,
+):
+    @click.command()
+    def neutral_package():
+        pass
+
+    def entry_points(*, group):
+        if group == NEUTRAL_CLI_PLUGIN_GROUP:
+            return [
+                _entry_point("neutral-package", "neutral_plugin:cli", neutral_package)
+            ]
+        return []
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        entry_points,
+    )
+    ios_group = click.Group()
+    android_group = click.Group()
+
+    register_cli_plugins(
+        ios_group,
+        entry_point_group=IOS_CLI_PLUGIN_GROUP,
+        environment_variable="TEST_IOS_COMMANDS",
+    )
+    register_cli_plugins(
+        android_group,
+        entry_point_group=ANDROID_CLI_PLUGIN_GROUP,
+        environment_variable="TEST_ANDROID_COMMANDS",
+    )
+
+    assert not ios_group.commands
+    assert not android_group.commands
+
+
+def test_environment_command_wins_collision_with_installed_command(
+    tmp_path, monkeypatch, caplog
+):
+    command_path = _write_command(
+        tmp_path / "duplicate.py",
+        "duplicate",
+        message="environment command ran",
+    )
+
+    @click.command()
+    def installed_command():
+        pass
+
+    def entry_points(*, group):
+        if group == IOS_CLI_PLUGIN_GROUP:
+            return [
+                _entry_point(
+                    "duplicate",
+                    "ios_plugin:cli",
+                    installed_command,
+                    distribution="ios-plugin",
+                )
+            ]
+        return []
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        entry_points,
+    )
+    monkeypatch.setenv("TEST_IOS_COMMANDS", str(command_path))
+    group = click.Group()
+
+    register_cli_plugins(
+        group,
+        entry_point_group=IOS_CLI_PLUGIN_GROUP,
+        environment_variable="TEST_IOS_COMMANDS",
+    )
+
+    assert group.commands["duplicate"] is not installed_command
+    result = CliRunner().invoke(group, ["duplicate"])
+    assert result.exit_code == 0
+    assert "environment command ran" in result.output
+    assert "the command name is already registered" in caplog.text
+    assert "ios-plugin 1.0 (ios_plugin:cli)" in caplog.text
+
+
+def test_the_mvt_cli_gets_the_neutral_commands_and_no_platform_command(
+    monkeypatch, restore_cli_commands
+):
+    @click.command()
+    def shared_package():
+        click.echo("shared command ran")
+
+    @click.command()
+    def ios_package():
+        pass
+
+    def entry_points(*, group):
+        if group == NEUTRAL_CLI_PLUGIN_GROUP:
+            return [_entry_point("shared-package", "shared_plugin:cli", shared_package)]
+        if group == IOS_CLI_PLUGIN_GROUP:
+            return [_entry_point("ios-package", "ios_plugin:cli", ios_package)]
+        return []
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        entry_points,
+    )
+
+    register_cli_plugins(
+        mvt_cli,
+        entry_point_group=NEUTRAL_CLI_PLUGIN_GROUP,
+        environment_variable=MVT_CUSTOM_COMMANDS_ENV,
+    )
+
+    assert "ios-package" not in mvt_cli.commands
+    result = CliRunner().invoke(mvt_cli, [*OFFLINE, "shared-package"])
+    assert result.exit_code == 0
+    assert "shared command ran" in result.output
+
+
+def test_builtin_mvt_command_wins_collision_with_neutral_command(
+    monkeypatch, caplog, restore_cli_commands
+):
+    @click.command()
+    def neutral_version():
+        pass
+
+    def entry_points(*, group):
+        if group == NEUTRAL_CLI_PLUGIN_GROUP:
+            return [
+                _entry_point(
+                    "version",
+                    "neutral_plugin:cli",
+                    neutral_version,
+                    distribution="neutral-plugin",
+                )
+            ]
+        return []
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        entry_points,
+    )
+    builtin_version = mvt_cli.commands["version"]
+
+    register_cli_plugins(
+        mvt_cli,
+        entry_point_group=NEUTRAL_CLI_PLUGIN_GROUP,
+        environment_variable=MVT_CUSTOM_COMMANDS_ENV,
+    )
+
+    assert mvt_cli.commands["version"] is builtin_version
+    assert "the command name is already registered" in caplog.text
+    assert "neutral-plugin 1.0 (neutral_plugin:cli)" in caplog.text
+
+
+def test_broken_neutral_plugin_does_not_break_the_mvt_cli(
+    monkeypatch, restore_cli_commands
+):
+    def entry_points(*, group):
+        if group == NEUTRAL_CLI_PLUGIN_GROUP:
+            return [
+                _entry_point(
+                    "broken",
+                    "broken_plugin:cli",
+                    exception=RuntimeError("missing dependency"),
+                    distribution="broken-plugin",
+                )
+            ]
+        return []
+
+    monkeypatch.setattr(
+        "mvt.common.cli_plugins.importlib.metadata.entry_points",
+        entry_points,
+    )
+
+    register_cli_plugins(
+        mvt_cli,
+        entry_point_group=NEUTRAL_CLI_PLUGIN_GROUP,
+        environment_variable=MVT_CUSTOM_COMMANDS_ENV,
+    )
+
+    assert isinstance(mvt_cli.commands["broken"], BrokenPluginCommand)
+    result = CliRunner().invoke(mvt_cli, [*OFFLINE, "version"])
+    assert result.exit_code == 0

tests/common/test_cmd_check_iocs.py

@@ -0,0 +1,225 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+import json
+import logging
+
+import pytest
+from click.testing import CliRunner
+
+from mvt.android.cli import cli as android_cli
+from mvt.android.command_modules import ANDROID_CHECK_IOCS_MODULES
+from mvt.common.cmd_check_iocs import CmdCheckIOCS
+from mvt.common.module import MVTModule
+from mvt.ios.cli import cli as ios_cli
+from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES
+from mvt.ios.modules.backup.manifest import Manifest
+
+# Keep the banner of the group callback from checking for updates online.
+OFFLINE = ["--disable-update-check", "--disable-indicator-update-check"]
+
+
+class CustomResultsModule(MVTModule):
+    """A custom module which declares the check-iocs pair of both platforms."""
+
+    slug = "custom_results"
+    supported_commands = (
+        ("ios", "check-backup"),
+        ("ios", "check-iocs"),
+        ("android", "check-iocs"),
+    )
+
+    checked: list = []
+
+    def run(self) -> None:
+        pass
+
+    def check_indicators(self) -> None:
+        self.checked.append(list(self.results))
+
+
+class BackupCheckerModule(MVTModule):
+    """An iOS module which implements check_indicators() without declaring check-iocs."""
+
+    slug = "backup_checker"
+    supported_commands = (("ios", "check-backup"),)
+
+    checked: list = []
+
+    def run(self) -> None:
+        pass
+
+    def check_indicators(self) -> None:
+        self.checked.append(list(self.results))
+
+
+class BugReportCheckerModule(MVTModule):
+    """The same for Android."""
+
+    slug = "bugreport_checker"
+    supported_commands = (("android", "check-bugreport"),)
+
+    checked: list = []
+
+    def run(self) -> None:
+        pass
+
+    def check_indicators(self) -> None:
+        self.checked.append(list(self.results))
+
+
+class BackupOnlyModule(MVTModule):
+    """A custom module which does not implement check_indicators()."""
+
+    slug = "backup_only"
+    supported_commands = (("ios", "check-backup"),)
+
+    def run(self) -> None:
+        pass
+
+
+@pytest.mark.parametrize(
+    "platform, builtin_modules, checker_module",
+    [
+        ("ios", IOS_CHECK_IOCS_MODULES, BackupCheckerModule),
+        ("android", ANDROID_CHECK_IOCS_MODULES, BugReportCheckerModule),
+    ],
+)
+def test_check_iocs_rechecks_the_stored_results_of_custom_modules(
+    platform, builtin_modules, checker_module, tmp_path, caplog
+):
+    # check-iocs matches every <slug>.json in the results folder to the module
+    # with that slug, custom modules included, and runs its check_indicators()
+    # again over the stored results.
+    results = [{"domain": "example.org"}]
+    (tmp_path / "custom_results.json").write_text(json.dumps(results))
+    (tmp_path / f"{checker_module.slug}.json").write_text(json.dumps(results))
+    (tmp_path / "backup_only.json").write_text(json.dumps(results))
+    CustomResultsModule.checked.clear()
+    checker_module.checked.clear()
+
+    cmd = CmdCheckIOCS(
+        target_path=str(tmp_path),
+        custom_modules=[CustomResultsModule, checker_module, BackupOnlyModule],
+        platform=platform,
+    )
+    cmd.modules = builtin_modules
+
+    with caplog.at_level(logging.INFO):
+        cmd.run()
+
+    # A module which declares the check-iocs pair is re-checked.
+    assert CustomResultsModule.checked == [results]
+    assert (
+        'Loading results from "custom_results.json" with module CustomResultsModule'
+        in caplog.text
+    )
+    # So is a module which only implements check_indicators().
+    assert checker_module.checked == [results]
+    # A module which does neither is not part of check-iocs.
+    assert "backup_only.json" not in caplog.text
+
+
+@pytest.mark.parametrize(
+    "platform, builtin_modules, listed, not_listed",
+    [
+        (
+            "ios",
+            IOS_CHECK_IOCS_MODULES,
+            "BackupCheckerModule",
+            "BugReportCheckerModule",
+        ),
+        (
+            "android",
+            ANDROID_CHECK_IOCS_MODULES,
+            "BugReportCheckerModule",
+            "BackupCheckerModule",
+        ),
+    ],
+)
+def test_check_iocs_lists_the_custom_modules_it_runs(
+    platform, builtin_modules, listed, not_listed, caplog
+):
+    cmd = CmdCheckIOCS(
+        custom_modules=[
+            CustomResultsModule,
+            BackupCheckerModule,
+            BugReportCheckerModule,
+            BackupOnlyModule,
+        ],
+        platform=platform,
+    )
+    cmd.modules = builtin_modules
+
+    with caplog.at_level(logging.INFO):
+        cmd.list_modules()
+
+    assert "CustomResultsModule" in caplog.text
+    # The module which implements check_indicators() for this platform is listed.
+    assert listed in caplog.text
+    # The one for the other platform is not, and neither is BackupOnlyModule.
+    assert not_listed not in caplog.text
+    assert "BackupOnlyModule" not in caplog.text
+
+
+class ReplacementManifest(Manifest):
+    """A replacement for a built-in module which does not declare check-iocs."""
+
+    supported_commands = (("ios", "check-backup"),)
+    replaces = Manifest
+
+
+def test_check_iocs_uses_a_replacement_of_a_built_in_module():
+    # A replacement which subclasses a built-in module inherits its
+    # check_indicators(). check-iocs then runs it in place of that module.
+    cmd = CmdCheckIOCS(custom_modules=[ReplacementManifest], platform="ios")
+    cmd.modules = IOS_CHECK_IOCS_MODULES
+
+    available = cmd._available_modules()
+
+    assert ReplacementManifest in available
+    assert Manifest not in available
+
+
+LOADED_MODULE = '''
+from mvt.common.module import MVTModule
+
+
+class LoadedResultsModule(MVTModule):
+    """A module loaded from a file with --load-module."""
+
+    slug = "loaded_results"
+    supported_commands = (("ios", "check-iocs"), ("android", "check-iocs"))
+
+    def run(self) -> None:
+        pass
+
+    def check_indicators(self) -> None:
+        self.log.warning("loaded module checked %d results", len(self.results))
+'''
+
+
+@pytest.mark.parametrize("cli", [ios_cli, android_cli], ids=["mvt-ios", "mvt-android"])
+def test_check_iocs_loads_custom_modules_from_a_file_on_each_cli(cli, tmp_path, caplog):
+    module_path = tmp_path / "loaded_module.py"
+    module_path.write_text(LOADED_MODULE)
+    results_folder = tmp_path / "results"
+    results_folder.mkdir()
+    (results_folder / "loaded_results.json").write_text(json.dumps([{"a": 1}]))
+
+    with caplog.at_level(logging.INFO):
+        result = CliRunner().invoke(
+            cli,
+            [
+                *OFFLINE,
+                "check-iocs",
+                "--load-module",
+                str(module_path),
+                str(results_folder),
+            ],
+        )
+
+    assert result.exit_code == 0, result.output
+    assert "loaded module checked 1 results" in caplog.text

tests/common/test_cmd_plugins.py

@@ -0,0 +1,240 @@
+import json
+import re
+from types import SimpleNamespace
+
+import pytest
+from click.testing import CliRunner
+
+from mvt.android.cli import cli as android_cli
+from mvt.cli import cli as mvt_cli
+from mvt.common.cli_plugins import (
+    ANDROID_CLI_PLUGIN_GROUP,
+    IOS_CLI_PLUGIN_GROUP,
+    NEUTRAL_CLI_PLUGIN_GROUP,
+)
+from mvt.common.cmd_plugins import plugins
+from mvt.common.module import MVTModule
+from mvt.common.module_loader import MODULES_ENTRY_POINT_GROUP
+from mvt.common.updates import PluginUpdates
+from mvt.ios.cli import cli as ios_cli
+
+
+class ExampleModule(MVTModule):
+    pass
+
+
+class AnotherModule(MVTModule):
+    pass
+
+
+class FakeDistribution:
+    def __init__(self, name, version="1.0.0", direct_url=None):
+        self.name = name
+        self.version = version
+        self.direct_url = direct_url
+
+    def read_text(self, file_name):
+        if file_name == "direct_url.json" and self.direct_url is not None:
+            return json.dumps(self.direct_url)
+        return None
+
+
+def _entry_point(name, distribution, modules=None, exception=None):
+    def load():
+        if exception is not None:
+            raise exception
+        return modules
+
+    return SimpleNamespace(
+        name=name, value="example_plugin:modules", dist=distribution, load=load
+    )
+
+
+def _run(command, arguments):
+    # Keep rich from wrapping the table while its content is being asserted.
+    return CliRunner().invoke(command, arguments, env={"COLUMNS": "200"})
+
+
+def _table_lines(output):
+    """Return the cells of each line of the table, header first."""
+    return [
+        [cell.strip() for cell in re.split("[│┃]", line.strip().strip("│┃"))]
+        for line in output.splitlines()
+        if "│" in line or "┃" in line
+    ]
+
+
+def _table_rows(output):
+    """Return the content of the table rows, without the header and the box."""
+    return _table_lines(output)[1:]
+
+
+def _table_header(output):
+    lines = _table_lines(output)
+    return lines[0] if lines else []
+
+
+def _install(monkeypatch, distributions, entry_points):
+    monkeypatch.setattr(
+        "mvt.common.cmd_plugins.installed_plugin_distributions",
+        lambda: distributions,
+    )
+    monkeypatch.setattr(
+        "mvt.common.cmd_plugins.importlib.metadata.entry_points",
+        lambda *, group: entry_points.get(group, []),
+    )
+
+
+def test_plugins_is_a_builtin_command_of_the_mvt_cli_only():
+    assert mvt_cli.commands["plugins"] is plugins
+    assert "plugins" not in ios_cli.commands
+    assert "plugins" not in android_cli.commands
+
+
+def test_list_shows_what_every_plugin_contributes(monkeypatch):
+    index_plugin = FakeDistribution("example-plugin", version="1.2.0")
+    repository_plugin = FakeDistribution(
+        "repository-plugin",
+        version="0.1.0",
+        direct_url={
+            "url": "https://example.org/plugin.git",
+            "vcs_info": {"vcs": "git", "commit_id": "b" * 40},
+        },
+    )
+    local_plugin = FakeDistribution(
+        "local-plugin",
+        direct_url={"url": "file:///plugins", "dir_info": {"editable": True}},
+    )
+    _install(
+        monkeypatch,
+        [index_plugin, local_plugin, repository_plugin],
+        {
+            MODULES_ENTRY_POINT_GROUP: [
+                _entry_point(
+                    "example", index_plugin, modules=[ExampleModule, AnotherModule]
+                ),
+                _entry_point("local", local_plugin, modules=lambda: [ExampleModule]),
+            ],
+            IOS_CLI_PLUGIN_GROUP: [_entry_point("summarize", repository_plugin)],
+            ANDROID_CLI_PLUGIN_GROUP: [_entry_point("triage", local_plugin)],
+            NEUTRAL_CLI_PLUGIN_GROUP: [_entry_point("report", repository_plugin)],
+        },
+    )
+
+    result = _run(plugins, ["list"])
+
+    assert result.exit_code == 0
+    # Plugins are listed by name, with the modules and the commands each of
+    # them contributes.
+    assert _table_header(result.output) == [
+        "Name",
+        "Version",
+        "Origin",
+        "Modules",
+        "Commands",
+    ]
+    assert _table_rows(result.output) == [
+        ["example-plugin", "1.2.0", "pypi", "2", "-"],
+        ["local-plugin", "1.0.0", "local", "1", "triage"],
+        ["repository-plugin", "0.1.0", "git+bbbbbbbb", "0", "report, summarize"],
+    ]
+
+
+def test_list_reports_a_broken_module_entry_point(monkeypatch):
+    plugin = FakeDistribution("broken-plugin")
+    _install(
+        monkeypatch,
+        [plugin],
+        {
+            MODULES_ENTRY_POINT_GROUP: [
+                _entry_point(
+                    "broken", plugin, exception=ImportError("missing dependency")
+                )
+            ]
+        },
+    )
+
+    result = _run(plugins, ["list"])
+
+    assert result.exit_code == 0
+    assert _table_rows(result.output) == [
+        ["broken-plugin", "1.0.0", "pypi", "error", "-"]
+    ]
+
+
+def test_list_without_plugins(monkeypatch):
+    _install(monkeypatch, [], {})
+
+    result = _run(plugins, ["list"])
+
+    assert result.exit_code == 0
+    assert result.output == "No MVT plugins are installed.\n"
+
+
+def test_check_updates_prints_the_findings_and_ignores_the_throttle(monkeypatch):
+    findings = [
+        {
+            "name": "example-plugin",
+            "installed": "1.0.0",
+            "latest": "1.2.0",
+            "origin": "pypi",
+            "upgrade_command": "pip install -U example-plugin",
+        }
+    ]
+    _install(monkeypatch, [FakeDistribution("example-plugin")], {})
+    monkeypatch.setattr(PluginUpdates, "check", lambda self: findings)
+    monkeypatch.setattr(
+        PluginUpdates,
+        "should_check",
+        lambda self: pytest.fail("an explicit check must not be throttled"),
+    )
+
+    result = _run(plugins, ["check-updates"])
+
+    assert result.exit_code == 0
+    assert "Plugin updates available:" in result.output
+    assert "example-plugin  1.0.0 → 1.2.0" in result.output
+    assert "Upgrade with: pip install -U example-plugin" in result.output
+    assert "MVT does not install plugin updates." in result.output
+
+
+def test_check_updates_without_available_updates(monkeypatch):
+    _install(monkeypatch, [FakeDistribution("example-plugin")], {})
+    monkeypatch.setattr(PluginUpdates, "check", lambda self: [])
+
+    result = _run(plugins, ["check-updates"])
+
+    assert result.exit_code == 0
+    assert "All plugins are up to date." in result.output
+
+
+def test_check_updates_without_plugins(monkeypatch):
+    _install(monkeypatch, [], {})
+    monkeypatch.setattr(
+        PluginUpdates,
+        "check",
+        lambda self: pytest.fail("nothing must be checked without plugins"),
+    )
+
+    result = _run(plugins, ["check-updates"])
+
+    assert result.exit_code == 0
+    assert "No MVT plugins are installed." in result.output
+
+
+def test_check_updates_without_network_access(monkeypatch):
+    monkeypatch.setattr("mvt.common.cmd_plugins.settings.NETWORK_ACCESS_ALLOWED", False)
+    monkeypatch.setattr(
+        "mvt.common.cmd_plugins.installed_plugin_distributions",
+        lambda: pytest.fail("plugins must not be listed without network access"),
+    )
+    monkeypatch.setattr(
+        PluginUpdates,
+        "check",
+        lambda self: pytest.fail("nothing must be checked without network access"),
+    )
+
+    result = _run(plugins, ["check-updates"])
+
+    assert result.exit_code == 0
+    assert "Network access is disabled" in result.output

tests/common/test_command.py

@@ -5,8 +5,13 @@
 
 import json
 import logging
+from unittest.mock import patch
+
+import pytest
+from click.testing import CliRunner
 
 from mvt.common.command import Command
+from mvt.common.indicators import Indicators
 from mvt.common.module import MVTModule
 
 
@@ -68,6 +73,123 @@ class CustomDependsOnBuiltin(RecordingModule):
     dependencies = (FirstModule,)
 
 
+class ReplacementModule(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+
+    def run(self):
+        super().run()
+        self.results = ["replacement"]
+
+
+class OtherReplacementModule(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+
+
+class UnrelatedReplacementModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = IndependentModule
+
+
+class ReplacementOfReplacementModule(ReplacementModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = ReplacementModule
+
+    def run(self):
+        super().run()
+        self.results = ["replacement of replacement"]
+
+
+class ReplacesOwnDependencyModule(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+    dependencies = (FirstModule,)
+
+
+class DisabledReplacementModule(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+    enabled = False
+
+
+class MutualReplacementOne(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+
+
+class MutualReplacementTwo(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = MutualReplacementOne
+
+
+MutualReplacementOne.replaces = MutualReplacementTwo
+
+
+class CycleOneModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+
+
+class CycleTwoModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = CycleOneModule
+
+
+class CycleThreeModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = CycleTwoModule
+
+
+CycleOneModule.replaces = CycleThreeModule
+
+
+class UnavailableDependencyModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+
+
+class ReplacementMissingDependency(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+    dependencies = (UnavailableDependencyModule,)
+
+
+class SharedSlugModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    slug = "shared_slug"
+
+
+class OtherSharedSlugModule(RecordingModule):
+    supported_commands = (("ios", "check-backup"),)
+    slug = "shared_slug"
+
+
+class ReplacementKeepingTheSlug(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+    slug = "first_module"
+
+
+class SameNameReplacementModule(FirstModule):
+    supported_commands = (("ios", "check-backup"),)
+    replaces = FirstModule
+
+    def run(self):
+        super().run()
+        self.results = ["replacement"]
+
+
+# Modules replacing a built-in one usually keep its class name, which is the
+# name `--module` matches on.
+SameNameReplacementModule.__name__ = "FirstModule"
+
+
+def logged_substitutions(caplog) -> list[str]:
+    return [
+        record.getMessage()
+        for record in caplog.records
+        if record.levelno == logging.INFO and "replaces module" in record.getMessage()
+    ]
+
+
 class RecordingCommand(Command):
     def init(self):
         self.initialized = True
@@ -80,6 +202,86 @@ class RecordingCommand(Command):
 
 
 class TestCommand:
+    def test_listing_modules_does_not_load_indicators(self):
+        with patch("mvt.common.command.Indicators.load_indicators_files") as load:
+            cmd = RecordingCommand()
+            cmd.list_modules()
+            load.assert_not_called()
+
+    def test_output_folder_is_created_by_a_run_not_by_listing(self, tmp_path):
+        output_path = tmp_path / "out"
+        with patch("mvt.common.command.Indicators.load_indicators_files"):
+            cmd = RecordingCommand(results_path=str(output_path))
+            cmd.list_modules()
+            assert not output_path.exists()
+            cmd.run()
+        assert (output_path / "command.log").is_file()
+
+    def test_indicators_load_once_and_are_shared(self, indicator_file, monkeypatch):
+        from mvt.common.config import settings
+
+        monkeypatch.setattr(settings, "STIX2", "")
+        monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None)
+        original = Indicators.load_indicators_files
+        with patch.object(
+            Indicators, "load_indicators_files", autospec=True, side_effect=original
+        ) as load:
+            cmd = RecordingCommand(ioc_files=[indicator_file])
+            load.assert_not_called()
+            indicators = cmd.iocs
+            assert indicators.total_ioc_count == 9
+            assert len(indicators.ioc_collections) == 1
+            assert cmd.iocs is indicators
+            child = RecordingCommand(iocs=indicators)
+            assert child.iocs is indicators
+            load.assert_called_once_with(indicators, [indicator_file])
+
+    @pytest.mark.parametrize("assign", [False, True])
+    def test_supplied_empty_indicators_are_not_loaded(self, assign):
+        indicators = Indicators(logging.getLogger(__name__))
+        with patch.object(Indicators, "load_indicators_files") as load:
+            cmd = RecordingCommand(iocs=None if assign else indicators)
+            if assign:
+                cmd.iocs = indicators
+            assert cmd.iocs is indicators
+            assert cmd.iocs.total_ioc_count == 0
+            load.assert_not_called()
+
+    @pytest.mark.parametrize("list_modules", [False, True])
+    def test_backup_cli_does_not_load_indicators_before_analysis(
+        self, tmp_path, list_modules
+    ):
+        from mvt.ios.cli import check_backup
+
+        args = [str(tmp_path)]
+        if list_modules:
+            args.insert(0, "--list-modules")
+        with patch.object(Indicators, "load_indicators_files") as load:
+            result = CliRunner().invoke(check_backup, args)
+            assert result.exit_code == (0 if list_modules else 1)
+            load.assert_not_called()
+
+    def test_run_checks_synthetic_indicators(self, indicator_file, monkeypatch):
+        from mvt.common.config import settings
+
+        monkeypatch.setattr(settings, "STIX2", "")
+        monkeypatch.setattr(Indicators, "_load_downloaded_indicators", lambda self: None)
+
+        class MatchingModule(RecordingModule):
+            def run(self):
+                self.results = ["https://example.org/test"]
+
+            def check_indicators(self):
+                self.detected = [
+                    url for url in self.results if self.indicators.check_domain(url)
+                ]
+
+        cmd = RecordingCommand(ioc_files=[indicator_file])
+        cmd.modules = [MatchingModule]
+        cmd.run()
+        assert cmd.executed[0].detected == ["https://example.org/test"]
+        assert cmd.executed[0].indicators is cmd.iocs
+
     def setup_method(self):
         RecordingModule.run_order = []
 
@@ -157,7 +359,7 @@ class TestCommand:
         assert not hasattr(cmd, "initialized")
         assert "Circular module dependency detected" in caplog.text
 
-    def test_unavailable_dependency_warns_and_stops(self, caplog):
+    def test_unavailable_dependency_only_skips_the_dependent_module(self, caplog):
         class UnavailableModule(RecordingModule):
             pass
 
@@ -165,14 +367,97 @@ class TestCommand:
             dependencies = (UnavailableModule,)
 
         cmd = RecordingCommand()
-        cmd.modules = [DependentModule]
+        cmd.modules = [DependentModule, IndependentModule, FirstModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["IndependentModule", "FirstModule"]
+        assert cmd.initialized
+        assert "Module DependentModule will be SKIPPED" in caplog.text
+        assert "depends on module UnavailableModule" in caplog.text
+
+    def test_modules_depending_on_a_skipped_module_are_skipped_too(self, caplog):
+        class UnavailableModule(RecordingModule):
+            pass
+
+        class SkippedModule(RecordingModule):
+            dependencies = (UnavailableModule,)
+
+        class DependsOnSkippedModule(RecordingModule):
+            dependencies = (SkippedModule,)
+
+        class DependsOnTheChain(RecordingModule):
+            dependencies = (DependsOnSkippedModule,)
+
+        cmd = RecordingCommand()
+        cmd.modules = [
+            DependsOnTheChain,
+            DependsOnSkippedModule,
+            SkippedModule,
+            IndependentModule,
+        ]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["IndependentModule"]
+        skip_warnings = [
+            record.getMessage()
+            for record in caplog.records
+            if "will be SKIPPED" in record.getMessage()
+        ]
+        assert len(skip_warnings) == 3
+        assert [warning.split()[1] for warning in skip_warnings] == [
+            "SkippedModule",
+            "DependsOnSkippedModule",
+            "DependsOnTheChain",
+        ]
+        # Every warning names the root cause: the module missing a dependency
+        # and the dependency it is missing.
+        assert all("UnavailableModule" in warning for warning in skip_warnings)
+        assert all("module SkippedModule" in warning for warning in skip_warnings[1:])
+
+    def test_explicitly_selected_module_with_missing_dependency_runs_nothing(
+        self, caplog
+    ):
+        class UnavailableModule(RecordingModule):
+            pass
+
+        class DependentModule(RecordingModule):
+            dependencies = (UnavailableModule,)
+
+        cmd = RecordingCommand(module_name="DependentModule")
+        cmd.modules = [DependentModule, IndependentModule]
 
         with caplog.at_level(logging.WARNING):
             cmd.run()
 
         assert RecordingModule.run_order == []
-        assert not hasattr(cmd, "initialized")
-        assert "depends on unavailable module UnavailableModule" in caplog.text
+        assert "Module DependentModule will be SKIPPED" in caplog.text
+        assert "No modules will be run" in caplog.text
+        # Nothing else was selected, so the warning must not promise that the
+        # analysis continues right before saying that it does not.
+        assert "The rest of the analysis will still run" not in caplog.text
+
+    def test_unaffected_dependency_chains_keep_their_order(self, caplog):
+        class UnavailableModule(RecordingModule):
+            pass
+
+        class SkippedModule(RecordingModule):
+            dependencies = (UnavailableModule,)
+
+        cmd = RecordingCommand()
+        cmd.modules = [ThirdModule, SkippedModule, SecondModule, FirstModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == [
+            "FirstModule",
+            "SecondModule",
+            "ThirdModule",
+        ]
 
     def test_custom_modules_are_filtered_before_ordering(self):
         cmd = RecordingCommand()
@@ -220,3 +505,376 @@ class TestCommand:
         cmd.run()
 
         assert RecordingModule.run_order == ["FirstModule", "CustomDependsOnBuiltin"]
+
+    def test_custom_module_replaces_builtin(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule, IndependentModule]
+        cmd.custom_modules = [ReplacementModule]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["IndependentModule", "ReplacementModule"]
+        assert cmd.module_replacements == {FirstModule: ReplacementModule}
+        assert (
+            "Module ReplacementModule from" in caplog.text
+            and "replaces module FirstModule" in caplog.text
+        )
+
+    def test_modules_without_replaces_all_run(self):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [CustomIOSBackupModule]
+
+        cmd.run()
+
+        assert RecordingModule.run_order == ["FirstModule", "CustomIOSBackupModule"]
+        assert cmd.module_replacements == {}
+
+    def test_unavailable_replaced_module_is_ignored(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [IndependentModule]
+        cmd.custom_modules = [ReplacementModule]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["IndependentModule", "ReplacementModule"]
+        assert cmd.module_replacements == {}
+        assert "replaces module FirstModule" not in caplog.text
+
+    def test_dependencies_are_resolved_to_the_replacement(self):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [SecondModule, FirstModule]
+        cmd.custom_modules = [ReplacementModule]
+
+        cmd.run()
+
+        assert RecordingModule.run_order == ["ReplacementModule", "SecondModule"]
+        second = next(
+            module for module in cmd.executed if isinstance(module, SecondModule)
+        )
+        assert isinstance(second.dependency_modules[FirstModule], ReplacementModule)
+        assert second.results == ["replacement", "second"]
+
+    def test_replacement_with_unavailable_dependency_skips_its_dependents(
+        self, caplog
+    ):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule, SecondModule, IndependentModule]
+        cmd.custom_modules = [ReplacementMissingDependency]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        # The replacement cannot run, and the module it replaced was dropped
+        # from the run by the replacement, so neither of them produces
+        # results and the module depending on the replaced one is skipped.
+        assert RecordingModule.run_order == ["IndependentModule"]
+        assert (
+            "Module ReplacementMissingDependency will be SKIPPED: it depends "
+            "on module UnavailableDependencyModule, which is not available in "
+            "this command." in caplog.text
+        )
+        # The skipped dependent is told which module it actually depends on,
+        # and the module class its author declared.
+        assert (
+            "Module SecondModule will be SKIPPED: it depends on module "
+            "ReplacementMissingDependency (replacing module FirstModule), "
+            "itself skipped for depending on unavailable module "
+            "UnavailableDependencyModule." in caplog.text
+        )
+
+    def test_selected_replacement_with_unavailable_dependency_runs_nothing(
+        self, caplog, tmp_path
+    ):
+        cmd = RecordingCommand(module_name="FirstModule", results_path=str(tmp_path))
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule, IndependentModule]
+        cmd.custom_modules = [ReplacementMissingDependency]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == []
+        assert not hasattr(cmd, "initialized")
+        assert (
+            "Module FirstModule was replaced by module "
+            "ReplacementMissingDependency, which is run in its place."
+            in caplog.text
+        )
+        assert "Module ReplacementMissingDependency will be SKIPPED" in caplog.text
+        assert "No modules will be run" in caplog.text
+        # Nothing else was selected, so the warnings must not promise that the
+        # analysis continues right before saying that it does not.
+        assert "The rest of the analysis will still run" not in caplog.text
+        # No module ran, so no results were stored next to the command log.
+        assert [path.name for path in tmp_path.iterdir()] == ["command.log"]
+
+    def test_chained_replacements_are_resolved(self):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [SecondModule, FirstModule]
+        cmd.custom_modules = [ReplacementModule, ReplacementOfReplacementModule]
+
+        cmd.run()
+
+        assert RecordingModule.run_order == [
+            "ReplacementOfReplacementModule",
+            "SecondModule",
+        ]
+        second = next(
+            module for module in cmd.executed if isinstance(module, SecondModule)
+        )
+        assert second.results == ["replacement of replacement", "second"]
+
+    def test_replacement_which_is_not_a_subclass_warns(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [IndependentModule]
+        cmd.custom_modules = [UnrelatedReplacementModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["UnrelatedReplacementModule"]
+        assert "is not a subclass of it" in caplog.text
+
+    def test_multiple_modules_replacing_the_same_module_warn(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [ReplacementModule, OtherReplacementModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == [
+            "ReplacementModule",
+            "OtherReplacementModule",
+        ]
+        assert cmd.module_replacements == {FirstModule: ReplacementModule}
+        assert (
+            "Modules ReplacementModule and OtherReplacementModule both replace "
+            "module FirstModule. Both of them will run, FirstModule will not, "
+            "and modules depending on FirstModule will use the results of "
+            "ReplacementModule." in caplog.text
+        )
+        assert "overwrite each other's results file" in caplog.text
+
+    def test_module_replacing_its_own_dependency_runs(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [ReplacesOwnDependencyModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["ReplacesOwnDependencyModule"]
+        assert (
+            "Module ReplacesOwnDependencyModule depends on module FirstModule, "
+            "which it also replaces" in caplog.text
+        )
+
+    def test_literal_self_dependency_is_still_circular(self, caplog):
+        class SelfDependentModule(RecordingModule):
+            pass
+
+        SelfDependentModule.dependencies = (SelfDependentModule,)
+
+        cmd = RecordingCommand()
+        cmd.modules = [SelfDependentModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == []
+        assert not hasattr(cmd, "initialized")
+        assert "Circular module dependency detected" in caplog.text
+
+    def test_disabled_replacement_keeps_the_replaced_module(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [DisabledReplacementModule]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["FirstModule"]
+        assert cmd.module_replacements == {}
+        assert logged_substitutions(caplog) == []
+
+    def test_modules_replacing_each_other_all_run(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.custom_modules = [MutualReplacementOne, MutualReplacementTwo]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == [
+            "MutualReplacementOne",
+            "MutualReplacementTwo",
+        ]
+        assert cmd.module_replacements == {}
+        assert (
+            "Modules MutualReplacementOne, MutualReplacementTwo replace each "
+            "other in a cycle" in caplog.text
+        )
+        assert logged_substitutions(caplog) == []
+
+    def test_replacement_cycle_of_three_modules_all_run(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.custom_modules = [CycleOneModule, CycleTwoModule, CycleThreeModule]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == [
+            "CycleOneModule",
+            "CycleTwoModule",
+            "CycleThreeModule",
+        ]
+        assert cmd.module_replacements == {}
+        assert (
+            "Modules CycleOneModule, CycleThreeModule, CycleTwoModule replace "
+            "each other in a cycle" in caplog.text
+        )
+        assert logged_substitutions(caplog) == []
+
+    def test_selected_replaced_module_name_runs_the_replacement(self, caplog):
+        cmd = RecordingCommand(module_name="FirstModule")
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [ReplacementModule]
+
+        with caplog.at_level(logging.INFO):
+            cmd.run()
+
+        assert RecordingModule.run_order == ["ReplacementModule"]
+        assert (
+            "Module FirstModule was replaced by module ReplacementModule"
+            in caplog.text
+        )
+
+    def test_unknown_selected_module_warns_and_stops(self, caplog):
+        cmd = RecordingCommand(module_name="NoSuchModule")
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == []
+        assert not hasattr(cmd, "initialized")
+        assert (
+            "No module named NoSuchModule is available for the check-backup "
+            "command" in caplog.text
+        )
+
+    def test_selected_module_name_matches_the_replacement(self):
+        cmd = RecordingCommand(module_name="FirstModule")
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [SameNameReplacementModule]
+
+        cmd.run()
+
+        assert len(cmd.executed) == 1
+        assert isinstance(cmd.executed[0], SameNameReplacementModule)
+        assert cmd.executed[0].results == ["replacement"]
+
+    def test_list_modules_reflects_replacements(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule, IndependentModule]
+        cmd.custom_modules = [ReplacementModule]
+
+        with caplog.at_level(logging.INFO):
+            cmd.list_modules()
+
+        listed = [
+            record.getMessage()
+            for record in caplog.records
+            if "Modules from" in record.getMessage()
+        ]
+        assert any("ReplacementModule" in message for message in listed)
+        assert not any("FirstModule" in message for message in listed)
+
+    def test_modules_sharing_a_slug_are_reported(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.custom_modules = [SharedSlugModule, OtherSharedSlugModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert RecordingModule.run_order == [
+            "SharedSlugModule",
+            "OtherSharedSlugModule",
+        ]
+        collisions = [
+            record.getMessage()
+            for record in caplog.records
+            if "both use the slug" in record.getMessage()
+        ]
+        assert len(collisions) == 1
+        assert "Modules SharedSlugModule from" in collisions[0]
+        assert "and OtherSharedSlugModule from" in collisions[0]
+        assert "both use the slug shared_slug" in collisions[0]
+        assert "overwrites the results of the other in shared_slug.json" in (
+            collisions[0]
+        )
+
+    def test_replacement_keeping_the_replaced_slug_is_not_reported(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule]
+        cmd.custom_modules = [ReplacementKeepingTheSlug]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        # The replaced module is no longer part of the run, so taking over its
+        # slug is what the replacement is for, not a collision.
+        assert RecordingModule.run_order == ["ReplacementKeepingTheSlug"]
+        assert ReplacementKeepingTheSlug.get_slug() == FirstModule.get_slug()
+        assert "both use the slug" not in caplog.text
+
+    def test_modules_with_distinct_slugs_are_not_reported(self, caplog):
+        cmd = RecordingCommand()
+        cmd.platform = "ios"
+        cmd.name = "check-backup"
+        cmd.modules = [FirstModule, IndependentModule]
+        cmd.custom_modules = [CustomIOSBackupModule]
+
+        with caplog.at_level(logging.WARNING):
+            cmd.run()
+
+        assert "both use the slug" not in caplog.text

tests/common/test_command_modules.py

@@ -0,0 +1,30 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+from mvt.android.command_modules import ANDROID_CHECK_IOCS_MODULES
+from mvt.android.modules.androidqf import ANDROIDQF_MODULES
+from mvt.android.modules.backup import BACKUP_MODULES as ANDROID_BACKUP_MODULES
+from mvt.android.modules.bugreport import BUGREPORT_MODULES
+from mvt.android.modules.intrusion_logs import INTRUSION_LOGS_MODULES
+from mvt.ios.command_modules import IOS_CHECK_IOCS_MODULES
+from mvt.ios.modules.backup import BACKUP_MODULES as IOS_BACKUP_MODULES
+from mvt.ios.modules.fs import FS_MODULES
+from mvt.ios.modules.mixed import MIXED_MODULES
+from mvt.ios.modules.sysdiagnose import SYSDIAGNOSE_MODULES
+
+
+def test_the_check_iocs_lists_are_the_families_of_their_platform():
+    # The CLI reads these same lists, so nothing composing one elsewhere can
+    # drift from what the command runs. This pins what the lists are composed
+    # of.
+    assert IOS_CHECK_IOCS_MODULES == (
+        IOS_BACKUP_MODULES + FS_MODULES + MIXED_MODULES + SYSDIAGNOSE_MODULES
+    )
+    assert ANDROID_CHECK_IOCS_MODULES == (
+        ANDROID_BACKUP_MODULES
+        + BUGREPORT_MODULES
+        + ANDROIDQF_MODULES
+        + INTRUSION_LOGS_MODULES
+    )

tests/common/test_config.py

@@ -0,0 +1,30 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+import os
+
+import yaml
+
+from mvt.common import config
+from mvt.common.config import MVTSettings
+
+
+def test_env_variables_are_not_persisted_to_config_file(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.yaml"
+    monkeypatch.setattr(config, "MVT_CONFIG_FOLDER", str(tmp_path))
+    monkeypatch.setattr(config, "MVT_CONFIG_PATH", str(config_path))
+    monkeypatch.setenv("MVT_NETWORK_ACCESS_ALLOWED", "false")
+    monkeypatch.setenv("MVT_IOS_BACKUP_PASSWORD", "env-only-password")
+
+    settings = MVTSettings.initialise()
+
+    assert os.path.isfile(config_path)
+    saved = yaml.safe_load(config_path.read_text()) or {}
+    assert "NETWORK_ACCESS_ALLOWED" not in saved
+    assert "IOS_BACKUP_PASSWORD" not in saved
+
+    # The environment must still apply to the settings in use.
+    assert settings.NETWORK_ACCESS_ALLOWED is False
+    assert settings.IOS_BACKUP_PASSWORD == "env-only-password"

tests/common/test_indicators.py

@@ -311,3 +311,31 @@ class TestIndicators:
         ind = Indicators(log=logging)
         ind.load_indicators_files([], load_default=False)
         assert ind.total_ioc_count == 9
+
+    def test_check_url_matches_w_prefixed_subdomain(self, tmp_path):
+        import json
+
+        stix_file = tmp_path / "w-domain.stix2"
+        stix_file.write_text(
+            json.dumps(
+                {
+                    "objects": [
+                        {
+                            "type": "indicator",
+                            "pattern": "[domain-name:value = 'web.evil.com']",
+                        }
+                    ]
+                }
+            ),
+            encoding="utf-8",
+        )
+        ind = Indicators(log=logging)
+        ind.load_indicators_files([str(stix_file)], load_default=False)
+
+        for url in (
+            "https://web.evil.com/path",
+            "https://www.web.evil.com/path",
+        ):
+            match = ind.check_url(url)
+            assert match is not None
+            assert match.ioc.value == "web.evil.com"

tests/common/test_module_loader.py

@@ -1,9 +1,14 @@
+from pathlib import Path
+
 import pytest
 
+from mvt.common.cli_plugins import _module_name_for_path as _command_module_name
 from mvt.common.module import MVTModule
 from mvt.common.module_loader import (
     CustomModuleLoadError,
+    _module_name_for_path,
     get_module_logger,
+    get_plugin_logger,
     load_custom_modules,
     load_custom_modules_from_path,
     module_supports_command,
@@ -168,9 +173,9 @@ def test_get_module_logger_strips_the_plugin_package_prefix():
     class PluginModule(MVTModule):
         pass
 
-    PluginModule.__module__ = "mvt_plugin_amnesty_custom.ios.custom"
+    PluginModule.__module__ = "mvt_plugin_example_org.ios.custom"
 
-    assert get_module_logger(PluginModule).name == "mvt.ext.amnesty_custom.ios.custom"
+    assert get_module_logger(PluginModule).name == "mvt.ext.example_org.ios.custom"
 
 
 def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
@@ -187,3 +192,25 @@ def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
     module = load_custom_modules_from_path(str(module_path))[0]
 
     assert get_module_logger(module).name == "mvt.ext.my_custom_module"
+
+
+def test_get_plugin_logger_uses_the_same_namespace_as_modules():
+    assert (
+        get_plugin_logger("mvt_plugin_example_org.commands.summarize").name
+        == "mvt.ext.example_org.commands.summarize"
+    )
+    assert get_plugin_logger("example_plugin.cli").name == "mvt.ext.example_plugin.cli"
+
+
+def test_get_plugin_logger_keeps_builtin_names():
+    assert get_plugin_logger("mvt.ios.cli").name == "mvt.ios.cli"
+
+
+def test_get_plugin_logger_names_loaded_files_after_the_file():
+    # A file loaded with --load-command or --load-module is imported under a
+    # mangled name. The log names the file instead.
+    command_name = _command_module_name(Path("/tmp/case_summary.py"))
+    module_name = _module_name_for_path(Path("/tmp/my_custom_module.py"))
+
+    assert get_plugin_logger(command_name).name == "mvt.ext.case_summary"
+    assert get_plugin_logger(module_name).name == "mvt.ext.my_custom_module"

tests/common/test_plugin.py

@@ -0,0 +1,34 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+import mvt.plugin
+from mvt.android.modules.backup.base import BackupModule
+from mvt.common.config import settings
+
+from ..plugin_fixtures import run_isolated_python
+
+
+def test_the_exported_names_are_the_public_names():
+    public = {name for name in vars(mvt.plugin) if not name.startswith("_")}
+
+    assert public == set(mvt.plugin.__all__)
+    assert mvt.plugin.settings is settings
+    assert mvt.plugin.AndroidBackupModule is BackupModule
+
+
+def test_the_surface_imports_before_anything_else_of_mvt(tmp_path):
+    # A plugin can import the surface as its first import of MVT. The
+    # subprocess gets a temporary home because importing MVT writes its
+    # configuration file.
+    result = run_isolated_python(
+        "from mvt.plugin import IOSExtraction, MVT_VERSION, settings\n"
+        "assert MVT_VERSION\n"
+        "assert settings.NETWORK_TIMEOUT > 0\n"
+        "assert IOSExtraction.__name__ == 'IOSExtraction'\n",
+        home=tmp_path,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stderr == ""

tests/common/test_plugin_config.py

@@ -0,0 +1,391 @@
+# Mobile Verification Toolkit (MVT)
+# Copyright (c) 2021-2026 The MVT Authors.
+# Use of this software is governed by the MVT License 1.1 that can be found at
+#   https://license.mvt.re/1.1/
+
+import os
+import stat
+import sys
+from typing import Optional
+
+import pytest
+import yaml
+
+from mvt.common.plugin_config import (
+    MVTPluginSettings,
+    PluginConfigLoadError,
+    plugin_config_folder,
+    plugin_config_path,
+    plugin_data_folder,
+    plugin_env_prefix,
+)
+
+
+class ExamplePluginSettings(MVTPluginSettings):
+    plugin_name = "example-plugin"
+
+    API_KEY: Optional[str] = None
+    CACHE_FOLDER: str = "cache"
+    MAX_RESULTS: int = 25
+
+
+class OtherPluginSettings(MVTPluginSettings):
+    plugin_name = "other-plugin"
+
+    API_KEY: Optional[str] = None
+
+
+@pytest.fixture
+def config_folder(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        "mvt.common.plugin_config.user_config_dir",
+        lambda *args, **kwargs: str(tmp_path),
+    )
+    return tmp_path
+
+
+@pytest.fixture
+def data_folder(tmp_path, monkeypatch):
+    folder = tmp_path / "data"
+    monkeypatch.setattr(
+        "mvt.common.plugin_config.user_data_dir",
+        lambda *args, **kwargs: str(folder),
+    )
+    return folder
+
+
+def _write_plugin_file(plugin_name, values):
+    config_path = plugin_config_path(plugin_name)
+    os.makedirs(os.path.dirname(config_path), exist_ok=True)
+    content = values if isinstance(values, str) else yaml.dump(values)
+    with open(config_path, "w") as config_file:
+        config_file.write(content)
+    return config_path
+
+
+def test_plugin_paths_and_prefixes_are_namespaced(config_folder):
+    assert plugin_config_folder() == str(config_folder / "plugins")
+    assert plugin_config_path("example-plugin") == str(
+        config_folder / "plugins" / "example-plugin.yaml"
+    )
+    assert plugin_env_prefix("example-plugin") == "MVT_PLUGIN_EXAMPLE_PLUGIN_"
+    assert plugin_env_prefix("other-plugin") == "MVT_PLUGIN_OTHER_PLUGIN_"
+
+
+def test_defaults_are_used_without_file_or_environment(config_folder):
+    settings = ExamplePluginSettings.load()
+
+    assert settings.API_KEY is None
+    assert settings.CACHE_FOLDER == "cache"
+    assert settings.MAX_RESULTS == 25
+    assert not os.path.exists(plugin_config_path("example-plugin"))
+
+
+def test_values_are_loaded_from_the_plugin_file(config_folder):
+    _write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
+
+    settings = ExamplePluginSettings.load()
+
+    assert settings.API_KEY == "from-file"
+    assert settings.MAX_RESULTS == 5
+    assert settings.CACHE_FOLDER == "cache"
+
+
+def test_environment_overrides_the_plugin_file(config_folder, monkeypatch):
+    _write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
+
+    settings = ExamplePluginSettings.load()
+
+    assert settings.API_KEY == "from-environment"
+    assert settings.MAX_RESULTS == 5
+
+
+def test_arguments_override_the_environment_and_the_plugin_file(
+    config_folder, monkeypatch
+):
+    _write_plugin_file("example-plugin", {"API_KEY": "from-file", "MAX_RESULTS": 5})
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "from-environment")
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "10")
+
+    settings = ExamplePluginSettings(API_KEY="from-argument")
+
+    assert settings.API_KEY == "from-argument"
+    assert settings.MAX_RESULTS == 10
+
+
+def test_save_and_load_round_trip(config_folder):
+    settings = ExamplePluginSettings.load()
+    settings.API_KEY = "saved-key"
+    settings.MAX_RESULTS = 100
+
+    settings.save()
+
+    config_path = plugin_config_path("example-plugin")
+    assert os.path.isfile(config_path)
+    with open(config_path) as config_file:
+        assert yaml.safe_load(config_file) == {
+            "API_KEY": "saved-key",
+            "MAX_RESULTS": 100,
+        }
+
+    reloaded = ExamplePluginSettings.load()
+    assert reloaded.API_KEY == "saved-key"
+    assert reloaded.MAX_RESULTS == 100
+    assert reloaded.CACHE_FOLDER == "cache"
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="POSIX file permissions are not available"
+)
+def test_saved_file_is_only_readable_by_the_user(config_folder):
+    settings = ExamplePluginSettings.load()
+    settings.API_KEY = "saved-key"
+
+    settings.save()
+
+    config_path = plugin_config_path("example-plugin")
+    assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
+    folder_mode = stat.S_IMODE(os.stat(plugin_config_folder()).st_mode)
+    assert folder_mode & 0o077 == 0
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="POSIX file permissions are not available"
+)
+def test_save_restricts_the_permissions_of_an_existing_file(config_folder):
+    config_path = _write_plugin_file("example-plugin", {"API_KEY": "from-file"})
+    os.chmod(config_path, 0o644)
+
+    settings = ExamplePluginSettings.load()
+    settings.MAX_RESULTS = 100
+    settings.save()
+
+    assert stat.S_IMODE(os.stat(config_path).st_mode) == 0o600
+    assert os.listdir(plugin_config_folder()) == ["example-plugin.yaml"]
+
+
+def test_save_only_persists_non_default_values(config_folder):
+    settings = ExamplePluginSettings.load()
+    settings.CACHE_FOLDER = "another-cache"
+
+    settings.save()
+
+    with open(plugin_config_path("example-plugin")) as config_file:
+        assert yaml.safe_load(config_file) == {"CACHE_FOLDER": "another-cache"}
+
+
+def test_save_does_not_persist_values_coming_from_the_environment(
+    config_folder, monkeypatch
+):
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
+
+    settings = ExamplePluginSettings.load()
+    assert settings.API_KEY == "environment-secret"
+    settings.MAX_RESULTS = 100
+    settings.save()
+
+    with open(plugin_config_path("example-plugin")) as config_file:
+        assert yaml.safe_load(config_file) == {"MAX_RESULTS": 100}
+
+
+def test_an_invalid_environment_variable_still_protects_the_other_values(
+    config_folder, monkeypatch
+):
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_MAX_RESULTS", "not-an-int")
+
+    settings = ExamplePluginSettings(MAX_RESULTS=100)
+    assert settings.API_KEY == "environment-secret"
+    settings.save()
+
+    with open(plugin_config_path("example-plugin")) as config_file:
+        saved_values = yaml.safe_load(config_file)
+    assert saved_values == {"MAX_RESULTS": 100}
+    assert "API_KEY" not in saved_values
+
+
+def test_save_persists_values_which_differ_from_the_environment(
+    config_folder, monkeypatch
+):
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "environment-secret")
+
+    settings = ExamplePluginSettings.load()
+    settings.API_KEY = "chosen-key"
+    settings.save()
+
+    with open(plugin_config_path("example-plugin")) as config_file:
+        assert yaml.safe_load(config_file) == {"API_KEY": "chosen-key"}
+
+
+def test_save_does_not_write_the_mvt_configuration_file(config_folder):
+    settings = ExamplePluginSettings.load()
+    settings.API_KEY = "saved-key"
+
+    settings.save()
+
+    assert os.listdir(config_folder) == ["plugins"]
+
+
+def test_unknown_keys_in_the_plugin_file_are_ignored(config_folder):
+    _write_plugin_file(
+        "example-plugin",
+        {"API_KEY": "from-file", "UNKNOWN_SETTING": "ignored"},
+    )
+
+    settings = ExamplePluginSettings.load()
+
+    assert settings.API_KEY == "from-file"
+    assert not hasattr(settings, "UNKNOWN_SETTING")
+
+
+def test_unparsable_plugin_file_is_reported_with_its_path(config_folder):
+    config_path = _write_plugin_file("example-plugin", "API_KEY: [unclosed\n")
+
+    with pytest.raises(PluginConfigLoadError) as raised:
+        ExamplePluginSettings.load()
+
+    assert config_path in str(raised.value)
+
+
+def test_plugin_file_which_is_not_a_mapping_is_reported_with_its_path(config_folder):
+    config_path = _write_plugin_file("example-plugin", "- one\n- two\n")
+
+    with pytest.raises(PluginConfigLoadError) as raised:
+        ExamplePluginSettings.load()
+
+    assert config_path in str(raised.value)
+    assert "mapping of setting names" in str(raised.value)
+
+
+def test_plugins_do_not_interfere_with_each_other(config_folder, monkeypatch):
+    _write_plugin_file("other-plugin", {"API_KEY": "other-file-key"})
+    monkeypatch.setenv("MVT_PLUGIN_EXAMPLE_PLUGIN_API_KEY", "example-environment-key")
+
+    example_settings = ExamplePluginSettings.load()
+    other_settings = OtherPluginSettings.load()
+
+    assert example_settings.API_KEY == "example-environment-key"
+    assert other_settings.API_KEY == "other-file-key"
+
+    example_settings.MAX_RESULTS = 100
+    example_settings.save()
+    assert sorted(os.listdir(plugin_config_folder())) == [
+        "example-plugin.yaml",
+        "other-plugin.yaml",
+    ]
+    with open(plugin_config_path("other-plugin")) as config_file:
+        assert yaml.safe_load(config_file) == {"API_KEY": "other-file-key"}
+
+
+def test_subclass_without_plugin_name_is_rejected():
+    with pytest.raises(TypeError, match="plugin_name"):
+
+        class MissingNameSettings(MVTPluginSettings):
+            API_KEY: Optional[str] = None
+
+
+def test_subclass_with_invalid_plugin_name_is_rejected():
+    with pytest.raises(ValueError, match="Invalid plugin name"):
+
+        class InvalidNameSettings(MVTPluginSettings):
+            plugin_name = "Bad/Name"
+
+
+def test_underscores_are_not_allowed_in_plugin_names():
+    # Underscores are replaced by dashes in the environment prefix, so allowing
+    # both would let two plugin names share one environment namespace.
+    with pytest.raises(ValueError, match="Invalid plugin name"):
+
+        class UnderscoreNameSettings(MVTPluginSettings):
+            plugin_name = "under_score"
+
+    with pytest.raises(ValueError, match="Invalid plugin name"):
+        plugin_config_path("under_score")
+    with pytest.raises(ValueError, match="Invalid plugin name"):
+        plugin_env_prefix("under_score")
+
+
+@pytest.mark.parametrize(
+    "plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
+)
+def test_unsafe_plugin_names_have_no_configuration_path(plugin_name):
+    with pytest.raises(ValueError, match="Invalid plugin name"):
+        plugin_config_path(plugin_name)
+
+
+def test_data_folder_is_namespaced_and_created(data_folder):
+    folder = plugin_data_folder("example-plugin")
+
+    assert folder == str(data_folder / "plugin-data" / "example-plugin")
+    assert os.path.isdir(folder)
+
+
+def test_data_folder_can_be_requested_repeatedly(data_folder):
+    folder = plugin_data_folder("example-plugin")
+    with open(os.path.join(folder, "kept.json"), "w") as data_file:
+        data_file.write("{}")
+
+    assert plugin_data_folder("example-plugin") == folder
+    assert os.listdir(folder) == ["kept.json"]
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="POSIX file permissions are not available"
+)
+def test_data_folder_is_only_accessible_by_the_user(data_folder):
+    folder = plugin_data_folder("example-plugin")
+
+    assert stat.S_IMODE(os.stat(folder).st_mode) & 0o077 == 0
+    parent_mode = stat.S_IMODE(os.stat(os.path.dirname(folder)).st_mode)
+    assert parent_mode & 0o077 == 0
+
+
+def test_plugins_get_their_own_data_folder(data_folder):
+    example_folder = plugin_data_folder("example-plugin")
+    other_folder = plugin_data_folder("other-plugin")
+
+    assert example_folder != other_folder
+    assert sorted(os.listdir(data_folder / "plugin-data")) == [
+        "example-plugin",
+        "other-plugin",
+    ]
+
+
+def test_data_folder_does_not_touch_the_configuration_folder(
+    config_folder, data_folder
+):
+    plugin_data_folder("example-plugin")
+
+    assert not os.path.exists(plugin_config_folder())
+
+
+@pytest.mark.parametrize(
+    "plugin_name", ["../escape", "folder/name", "UPPER", "-dash", ""]
+)
+def test_unsafe_plugin_names_have_no_data_folder(data_folder, plugin_name):
+    with pytest.raises(ValueError, match="Invalid plugin name"):
+        plugin_data_folder(plugin_name)
+
+    assert not os.path.exists(data_folder)
+
+
+def test_settings_class_knows_its_data_folder(data_folder):
+    folder = ExamplePluginSettings.data_folder()
+
+    assert folder == plugin_data_folder("example-plugin")
+    assert os.path.isdir(folder)
+    assert OtherPluginSettings.data_folder() != folder
+
+
+def test_settings_instance_uses_the_same_data_folder(config_folder, data_folder):
+    settings = ExamplePluginSettings.load()
+
+    assert settings.data_folder() == ExamplePluginSettings.data_folder()
+
+
+def test_subclass_without_its_own_name_shares_the_data_folder(data_folder):
+    class InheritingSettings(ExamplePluginSettings):
+        pass
+
+    assert InheritingSettings.data_folder() == ExamplePluginSettings.data_folder()

tests/common/test_plugin_updates.py

@@ -0,0 +1,711 @@
+import json
+import shlex
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+
+from mvt.common import logo
+from mvt.common.cli_plugins import (
+    ANDROID_CLI_PLUGIN_GROUP,
+    IOS_CLI_PLUGIN_GROUP,
+    NEUTRAL_CLI_PLUGIN_GROUP,
+)
+from mvt.common.module_loader import MODULES_ENTRY_POINT_GROUP
+from mvt.common.updates import (
+    MVTUpdates,
+    PluginUpdates,
+    installed_plugin_distributions,
+)
+
+
+REPOSITORY_URL = "https://example.org/plugin.git"
+INSTALLED_COMMIT = "a" * 40
+REMOTE_COMMIT = "b" * 40
+
+
+class FakeDistribution:
+    def __init__(self, name, version="1.0.0", direct_url=None):
+        self.name = name
+        self.version = version
+        self.direct_url = direct_url
+
+    def read_text(self, file_name):
+        if file_name == "direct_url.json" and self.direct_url is not None:
+            return json.dumps(self.direct_url)
+        return None
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, payload=None):
+        self.status_code = status_code
+        self.payload = payload or {}
+
+    def json(self):
+        return self.payload
+
+
+def _entry_point(name, distribution, value="plugin:modules"):
+    return SimpleNamespace(name=name, value=value, dist=distribution)
+
+
+def _git_distribution(requested_revision=None, commit=INSTALLED_COMMIT):
+    vcs_info = {"vcs": "git", "commit_id": commit}
+    if requested_revision:
+        vcs_info["requested_revision"] = requested_revision
+
+    return FakeDistribution(
+        "example-plugin",
+        direct_url={"url": REPOSITORY_URL, "vcs_info": vcs_info},
+    )
+
+
+def _fake_git(stdout="", returncode=0, exception=None, calls=None):
+    def run(command, **kwargs):
+        if calls is not None:
+            calls.append((command, kwargs))
+        if exception is not None:
+            raise exception
+        return SimpleNamespace(returncode=returncode, stdout=stdout, stderr="")
+
+    return run
+
+
+@pytest.fixture
+def data_folder(tmp_path, monkeypatch):
+    folder = tmp_path / "mvt-data"
+    monkeypatch.setattr("mvt.common.updates.MVT_DATA_FOLDER", str(folder))
+    return folder
+
+
+@pytest.fixture
+def one_plugin(monkeypatch):
+    def install(distribution):
+        monkeypatch.setattr(
+            "mvt.common.updates.installed_plugin_distributions",
+            lambda: [distribution],
+        )
+        return distribution
+
+    return install
+
+
+def test_installed_plugin_distributions_covers_every_plugin_group(monkeypatch):
+    zeta = FakeDistribution("zeta-plugin")
+    alpha = FakeDistribution("alpha-plugin")
+    neutral = FakeDistribution("neutral-plugin")
+
+    def entry_points(*, group):
+        if group == MODULES_ENTRY_POINT_GROUP:
+            return [_entry_point("zeta", zeta), _entry_point("alpha", alpha)]
+        if group == IOS_CLI_PLUGIN_GROUP:
+            return [_entry_point("zeta-ios", zeta)]
+        if group == ANDROID_CLI_PLUGIN_GROUP:
+            return [_entry_point("alpha-android", alpha)]
+        if group == NEUTRAL_CLI_PLUGIN_GROUP:
+            return [_entry_point("shared", neutral)]
+        return []
+
+    monkeypatch.setattr(
+        "mvt.common.updates.importlib.metadata.entry_points", entry_points
+    )
+
+    distributions = installed_plugin_distributions()
+
+    assert [distribution.name for distribution in distributions] == [
+        "alpha-plugin",
+        "neutral-plugin",
+        "zeta-plugin",
+    ]
+
+
+def test_installed_plugin_distributions_skips_mvt_and_orphan_entry_points(monkeypatch):
+    entry_points = [
+        _entry_point("builtin", FakeDistribution("mvt")),
+        SimpleNamespace(name="orphan", value="plugin:modules", dist=None),
+        _entry_point("plugin", FakeDistribution("example-plugin")),
+    ]
+    monkeypatch.setattr(
+        "mvt.common.updates.importlib.metadata.entry_points",
+        lambda **kwargs: entry_points,
+    )
+
+    distributions = installed_plugin_distributions()
+
+    assert [distribution.name for distribution in distributions] == ["example-plugin"]
+
+
+def test_installed_plugin_distributions_survives_broken_metadata(monkeypatch, caplog):
+    def entry_points(*, group):
+        raise RuntimeError("invalid package metadata")
+
+    monkeypatch.setattr(
+        "mvt.common.updates.importlib.metadata.entry_points", entry_points
+    )
+
+    assert installed_plugin_distributions() == []
+    assert "Unable to discover installed plugin packages" in caplog.text
+
+
+def test_index_plugin_update_is_reported(monkeypatch, data_folder, one_plugin):
+    one_plugin(FakeDistribution("example-plugin", version="1.0.0"))
+    monkeypatch.setattr(
+        "mvt.common.updates.requests.get",
+        lambda url, **kwargs: FakeResponse(payload={"info": {"version": "1.2.0"}}),
+    )
+
+    findings = PluginUpdates().check()
+
+    assert findings == [
+        {
+            "name": "example-plugin",
+            "installed": "1.0.0",
+            "latest": "1.2.0",
+            "origin": "pypi",
+            "upgrade_command": "pip install -U example-plugin",
+        }
+    ]
+
+
+def test_index_plugin_queries_the_package_index_with_the_configured_timeout(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(FakeDistribution("example-plugin"))
+    requests_made = []
+
+    def get(url, **kwargs):
+        requests_made.append((url, kwargs))
+        return FakeResponse(payload={"info": {"version": "1.0.0"}})
+
+    monkeypatch.setattr("mvt.common.updates.requests.get", get)
+    monkeypatch.setattr("mvt.common.updates.settings.NETWORK_TIMEOUT", 3)
+
+    PluginUpdates().check()
+
+    assert requests_made == [
+        ("https://pypi.org/pypi/example-plugin/json", {"timeout": 3})
+    ]
+
+
+def test_up_to_date_index_plugin_is_not_reported(monkeypatch, data_folder, one_plugin):
+    one_plugin(FakeDistribution("example-plugin", version="1.2.0"))
+    monkeypatch.setattr(
+        "mvt.common.updates.requests.get",
+        lambda url, **kwargs: FakeResponse(payload={"info": {"version": "1.2.0"}}),
+    )
+
+    assert PluginUpdates().check() == []
+
+
+def test_unpublished_plugin_is_skipped_silently(monkeypatch, data_folder, one_plugin):
+    one_plugin(FakeDistribution("private-plugin"))
+    monkeypatch.setattr(
+        "mvt.common.updates.requests.get",
+        lambda url, **kwargs: FakeResponse(status_code=404),
+    )
+
+    assert PluginUpdates().check() == []
+
+
+def test_repository_plugin_following_a_branch_is_reported(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision="main"))
+    calls = []
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout=f"{REMOTE_COMMIT}\trefs/heads/main\n", calls=calls),
+    )
+    monkeypatch.delenv("GIT_SSH_COMMAND", raising=False)
+
+    findings = PluginUpdates().check()
+
+    assert findings == [
+        {
+            "name": "example-plugin",
+            "installed": "aaaaaaaa",
+            "latest": "bbbbbbbb",
+            "origin": "git",
+            "upgrade_command": (
+                f"pip install -U 'example-plugin @ git+{REPOSITORY_URL}@main'"
+            ),
+        }
+    ]
+    command, options = calls[0]
+    assert command == ["git", "ls-remote", REPOSITORY_URL, "main"]
+    assert options["env"]["GIT_TERMINAL_PROMPT"] == "0"
+    # ssh asks the terminal for a passphrase or a host key unless it is told
+    # not to, which git itself cannot prevent.
+    assert options["env"]["GIT_SSH_COMMAND"] == (
+        "ssh -o BatchMode=yes -o ConnectTimeout=10"
+    )
+
+
+def test_batch_mode_options_come_before_the_configured_ssh_options(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision="main"))
+    calls = []
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout=f"{REMOTE_COMMIT}\trefs/heads/main\n", calls=calls),
+    )
+    monkeypatch.setenv("GIT_SSH_COMMAND", "ssh -o BatchMode=no -i /home/analyst/key")
+
+    PluginUpdates().check()
+
+    # ssh uses the first value it is given for a keyword, so an analyst asking
+    # for prompts cannot bring them back, while their other options still
+    # apply.
+    assert calls[0][1]["env"]["GIT_SSH_COMMAND"] == (
+        "ssh -o BatchMode=yes -o ConnectTimeout=10 -o BatchMode=no -i /home/analyst/key"
+    )
+
+
+def test_repository_plugin_without_a_revision_follows_the_default_branch(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution())
+    calls = []
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout=f"{REMOTE_COMMIT}\tHEAD\n", calls=calls),
+    )
+
+    findings = PluginUpdates().check()
+
+    assert calls[0][0] == ["git", "ls-remote", REPOSITORY_URL, "HEAD"]
+    assert findings[0]["upgrade_command"] == (
+        f"pip install -U 'example-plugin @ git+{REPOSITORY_URL}'"
+    )
+
+
+def test_hostile_revision_cannot_inject_into_the_upgrade_command(
+    monkeypatch, data_folder, one_plugin
+):
+    revision = "main$(id)`id`;id"
+    one_plugin(_git_distribution(requested_revision=revision))
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout=f"{REMOTE_COMMIT}\trefs/heads/{revision}\n"),
+    )
+
+    upgrade_command = PluginUpdates().check()[0]["upgrade_command"]
+
+    # Single quotes are the only quoting a shell does not expand anything in.
+    assert upgrade_command == (
+        f"pip install -U 'example-plugin @ git+{REPOSITORY_URL}@{revision}'"
+    )
+    assert shlex.split(upgrade_command) == [
+        "pip",
+        "install",
+        "-U",
+        f"example-plugin @ git+{REPOSITORY_URL}@{revision}",
+    ]
+
+
+def test_repository_plugin_with_an_option_like_url_is_skipped(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(
+        FakeDistribution(
+            "example-plugin",
+            direct_url={
+                "url": "--upload-pack=touch /tmp/mvt",
+                "vcs_info": {"vcs": "git", "commit_id": INSTALLED_COMMIT},
+            },
+        )
+    )
+    calls = []
+    monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
+
+    assert PluginUpdates().check() == []
+    assert calls == []
+
+
+def test_repository_plugin_with_an_option_like_revision_is_skipped(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision="--upload-pack=touch /tmp/mvt"))
+    calls = []
+    monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
+
+    assert PluginUpdates().check() == []
+    assert calls == []
+
+
+def test_repository_plugin_at_the_latest_commit_is_not_reported(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision="main"))
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout=f"{INSTALLED_COMMIT}\trefs/heads/main\n"),
+    )
+
+    assert PluginUpdates().check() == []
+
+
+def test_commit_pinned_repository_plugin_is_never_outdated(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision=INSTALLED_COMMIT))
+    calls = []
+    monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
+
+    assert PluginUpdates().check() == []
+    assert calls == []
+
+
+def test_short_commit_pinned_repository_plugin_is_never_outdated(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision=INSTALLED_COMMIT[:10]))
+    calls = []
+    monkeypatch.setattr("mvt.common.updates.subprocess.run", _fake_git(calls=calls))
+
+    assert PluginUpdates().check() == []
+    assert calls == []
+
+
+def test_tag_pinned_repository_plugin_is_never_outdated(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision="v1.0.0"))
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout=f"{REMOTE_COMMIT}\trefs/tags/v1.0.0\n"),
+    )
+
+    assert PluginUpdates().check() == []
+
+
+def test_repository_plugin_is_skipped_without_git(monkeypatch, data_folder, one_plugin):
+    one_plugin(_git_distribution(requested_revision="main"))
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(exception=FileNotFoundError("git")),
+    )
+
+    assert PluginUpdates().check() == []
+
+
+def test_repository_plugin_is_skipped_when_git_fails(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(_git_distribution(requested_revision="main"))
+    monkeypatch.setattr(
+        "mvt.common.updates.subprocess.run",
+        _fake_git(stdout="", returncode=128),
+    )
+
+    assert PluginUpdates().check() == []
+
+
+def test_local_plugin_install_is_skipped(monkeypatch, data_folder, one_plugin):
+    one_plugin(
+        FakeDistribution(
+            "example-plugin",
+            direct_url={
+                "url": "file:///home/analyst/example-plugin",
+                "dir_info": {"editable": True},
+            },
+        )
+    )
+
+    def fail(*args, **kwargs):
+        raise AssertionError("a local plugin install must not be checked")
+
+    monkeypatch.setattr("mvt.common.updates.requests.get", fail)
+    monkeypatch.setattr("mvt.common.updates.subprocess.run", fail)
+
+    assert PluginUpdates().check() == []
+
+
+def test_check_stores_the_findings_and_the_check_timestamp(
+    monkeypatch, data_folder, one_plugin
+):
+    one_plugin(FakeDistribution("example-plugin", version="1.0.0"))
+    monkeypatch.setattr(
+        "mvt.common.updates.requests.get",
+        lambda url, **kwargs: FakeResponse(payload={"info": {"version": "1.2.0"}}),
+    )
+    plugin_updates = PluginUpdates()
+
+    findings = plugin_updates.check()
+
+    assert json.loads((data_folder / "plugin_updates.json").read_text()) == findings
+    assert (data_folder / "latest_plugins_check").read_text().isdigit()
+    assert PluginUpdates().get_findings() == findings
+
+
+def test_findings_are_empty_before_the_first_check(data_folder):
+    assert PluginUpdates().get_findings() == []
+
+
+def test_malformed_cached_findings_are_dropped(data_folder):
+    plugin_updates = PluginUpdates()
+    usable = {
+        "name": "example-plugin",
+        "installed": "1.0.0",
+        "latest": "1.2.0",
+        "origin": "pypi",
+        "upgrade_command": "pip install -U example-plugin",
+    }
+    data_folder.mkdir(parents=True, exist_ok=True)
+    (data_folder / "plugin_updates.json").write_text(
+        json.dumps(
+            [
+                {"oops": 1},
+                "not a finding",
+                {"name": "half-plugin", "installed": "1.0.0"},
+                {**usable, "latest": None},
+                usable,
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    assert plugin_updates.get_findings() == [usable]
+
+
+def test_corrupt_cached_findings_are_ignored(data_folder):
+    data_folder.mkdir(parents=True, exist_ok=True)
+    (data_folder / "plugin_updates.json").write_text("{ not json", encoding="utf-8")
+
+    assert PluginUpdates().get_findings() == []
+
+
+def test_cached_findings_of_upgraded_and_removed_plugins_are_dropped(
+    monkeypatch, data_folder
+):
+    findings = [
+        {
+            "name": "upgraded-plugin",
+            "installed": "1.0.0",
+            "latest": "1.2.0",
+            "origin": "pypi",
+            "upgrade_command": "pip install -U upgraded-plugin",
+        },
+        {
+            "name": "removed-plugin",
+            "installed": "1.0.0",
+            "latest": "1.2.0",
+            "origin": "pypi",
+            "upgrade_command": "pip install -U removed-plugin",
+        },
+        {
+            "name": "example-plugin",
+            "installed": "1.0.0",
+            "latest": "1.2.0",
+            "origin": "pypi",
+            "upgrade_command": "pip install -U example-plugin",
+        },
+    ]
+    plugin_updates = PluginUpdates()
+    plugin_updates.set_findings(findings)
+    monkeypatch.setattr(
+        "mvt.common.updates.installed_plugin_distributions",
+        lambda: [
+            # The analyst upgraded this plugin since the latest check.
+            FakeDistribution("upgraded-plugin", version="1.2.0"),
+            FakeDistribution("example-plugin", version="1.0.0"),
+        ],
+    )
+
+    assert plugin_updates.current_findings() == [findings[2]]
+
+
+def test_cached_findings_of_updated_repository_plugins_are_dropped(data_folder):
+    findings = [
+        {
+            "name": "example-plugin",
+            "installed": "aaaaaaaa",
+            "latest": "bbbbbbbb",
+            "origin": "git",
+            "upgrade_command": "pip install -U example-plugin",
+        }
+    ]
+    plugin_updates = PluginUpdates()
+    plugin_updates.set_findings(findings)
+
+    assert plugin_updates.current_findings([_git_distribution()]) == findings
+    assert (
+        plugin_updates.current_findings([_git_distribution(commit=REMOTE_COMMIT)]) == []
+    )
+
+
+def test_corrupt_check_timestamp_does_not_raise(data_folder):
+    plugin_updates = PluginUpdates()
+    data_folder.mkdir(parents=True, exist_ok=True)
+    (data_folder / "latest_plugins_check").write_text("truncated", encoding="utf-8")
+
+    assert plugin_updates.get_latest_check() == 0
+    assert plugin_updates.should_check() == (True, 0)
+
+
+def test_should_check_is_throttled_for_twelve_hours(data_folder):
+    plugin_updates = PluginUpdates()
+    plugin_updates.set_findings([])
+
+    recent = datetime.now() - timedelta(hours=4)
+    with open(plugin_updates.latest_check_path, "w", encoding="utf-8") as handle:
+        handle.write(str(int(recent.timestamp())))
+
+    should_check, hours = plugin_updates.should_check()
+    assert not should_check
+    assert hours == 8
+
+    old = datetime.now() - timedelta(hours=13)
+    with open(plugin_updates.latest_check_path, "w", encoding="utf-8") as handle:
+        handle.write(str(int(old.timestamp())))
+
+    assert plugin_updates.should_check() == (True, 0)
+
+
+def test_should_check_without_a_previous_check(data_folder):
+    assert PluginUpdates().should_check() == (True, 0)
+
+
+@pytest.fixture
+def no_version_check(monkeypatch):
+    monkeypatch.setattr(MVTUpdates, "check", lambda self: "")
+    # Keep rich from wrapping the plugin lines while they are being asserted.
+    monkeypatch.setenv("COLUMNS", "200")
+
+
+@pytest.fixture
+def throttled_cache(monkeypatch, data_folder):
+    """Fill the findings cache and put the check inside its throttle window."""
+
+    def fill(findings, distributions):
+        PluginUpdates().set_findings(findings)
+        monkeypatch.setattr(
+            logo, "installed_plugin_distributions", lambda: distributions
+        )
+        monkeypatch.setattr(PluginUpdates, "should_check", lambda self: (False, 8))
+        monkeypatch.setattr(
+            PluginUpdates,
+            "check",
+            lambda self: pytest.fail("the check must be throttled"),
+        )
+
+    return fill
+
+
+def test_logo_prints_the_cached_plugin_updates(
+    capsys, no_version_check, throttled_cache
+):
+    throttled_cache(
+        [
+            {
+                "name": "example-plugin",
+                "installed": "1.0.0",
+                "latest": "1.2.0",
+                "origin": "pypi",
+                "upgrade_command": "pip install -U example-plugin",
+            }
+        ],
+        [FakeDistribution("example-plugin", version="1.0.0")],
+    )
+
+    logo.check_updates(disable_indicator_check=True)
+
+    output = capsys.readouterr().out
+    assert "Plugin updates available:" in output
+    assert "example-plugin  1.0.0 → 1.2.0   (pip install -U example-plugin)" in output
+
+
+def test_logo_does_not_print_a_cached_update_of_an_upgraded_plugin(
+    capsys, no_version_check, throttled_cache
+):
+    throttled_cache(
+        [
+            {
+                "name": "example-plugin",
+                "installed": "1.0.0",
+                "latest": "1.2.0",
+                "origin": "pypi",
+                "upgrade_command": "pip install -U example-plugin",
+            }
+        ],
+        # The analyst already upgraded the plugin the cached finding is about.
+        [FakeDistribution("example-plugin", version="1.2.0")],
+    )
+
+    logo.check_updates(disable_indicator_check=True)
+
+    assert "Plugin updates" not in capsys.readouterr().out
+
+
+def test_logo_prints_nothing_when_throttled_without_findings(
+    capsys, no_version_check, throttled_cache
+):
+    throttled_cache([], [FakeDistribution("example-plugin")])
+
+    logo.check_updates(disable_indicator_check=True)
+
+    assert "Plugin updates" not in capsys.readouterr().out
+
+
+def test_logo_survives_a_corrupt_plugin_cache(
+    monkeypatch, capsys, data_folder, no_version_check
+):
+    data_folder.mkdir(parents=True, exist_ok=True)
+    (data_folder / "plugin_updates.json").write_text(
+        json.dumps([{"oops": 1}]), encoding="utf-8"
+    )
+    monkeypatch.setattr(
+        logo,
+        "installed_plugin_distributions",
+        lambda: [FakeDistribution("example-plugin")],
+    )
+    monkeypatch.setattr(PluginUpdates, "should_check", lambda self: (False, 8))
+
+    logo.check_updates(disable_indicator_check=True)
+
+    assert "Plugin updates" not in capsys.readouterr().out
+
+
+def test_logo_skips_the_plugin_check_without_plugins(
+    monkeypatch, capsys, no_version_check
+):
+    monkeypatch.setattr(logo, "installed_plugin_distributions", list)
+    monkeypatch.setattr(
+        PluginUpdates,
+        "should_check",
+        lambda self: pytest.fail("plugins must not be checked without plugins"),
+    )
+
+    logo.check_updates(disable_indicator_check=True)
+
+    assert "Plugin updates" not in capsys.readouterr().out
+
+
+def test_logo_skips_the_plugin_check_without_network_access(
+    monkeypatch, capsys, no_version_check
+):
+    monkeypatch.setattr("mvt.common.logo.settings.NETWORK_ACCESS_ALLOWED", False)
+    monkeypatch.setattr(
+        logo,
+        "installed_plugin_distributions",
+        lambda: pytest.fail("plugins must not be listed without network access"),
+    )
+
+    logo.check_updates(disable_indicator_check=True)
+
+    assert "Plugin updates" not in capsys.readouterr().out
+
+
+def test_logo_skips_the_plugin_check_when_update_checks_are_disabled(
+    monkeypatch, capsys
+):
+    monkeypatch.setattr(
+        logo,
+        "installed_plugin_distributions",
+        lambda: pytest.fail("plugins must not be checked with --disable-update-check"),
+    )
+
+    logo.check_updates(disable_version_check=True, disable_indicator_check=True)
+
+    assert capsys.readouterr().out == ""

tests/common/test_url.py

@@ -22,3 +22,23 @@ def test_google_maps_url_is_not_shortened(url):
 
 def test_other_google_short_url_is_shortened():
     assert URL("https://goo.gl/example").check_if_shortened() is True
+
+
+@pytest.mark.parametrize(
+    "url, domain",
+    [
+        ("https://www.example.com/path", "example.com"),
+        # Only the whole "www." prefix comes off, not any leading "w" or "." character.
+        ("https://web.example.com", "web.example.com"),
+        ("https://wow.com", "wow.com"),
+        ("https://wired.com", "wired.com"),
+        ("https://www.wow.com", "wow.com"),
+    ],
+)
+def test_get_domain_strips_only_a_whole_www_prefix(url, domain):
+    assert URL(url).domain == domain
+
+
+def test_shortener_starting_with_w_is_detected():
+    assert URL("https://w3t.org/example").check_if_shortened() is True
+    assert URL("https://www.w3t.org/example").check_if_shortened() is True

tests/common/test_utils.py

@@ -18,6 +18,7 @@ from mvt.common.utils import (
     generate_hashes_from_path,
     get_sha256_from_file_path,
     init_logging,
+    set_verbose_logging,
 )
 
 from ..utils import get_artifact_folder
@@ -122,3 +123,29 @@ class TestInitLogging:
             sum(isinstance(handler, MVTLogHandler) for handler in log.handlers)
             == handler_count
         )
+
+    def test_verbose_logging_finds_the_console_handler_among_others(self):
+        # Something else may have attached a handler to the "mvt" logger
+        # before MVT did, so the console handler is not always the first.
+        log = logging.getLogger("mvt")
+        init_logging()
+        foreign_handler = logging.NullHandler()
+        foreign_handler.setLevel(logging.CRITICAL)
+        log.handlers.insert(0, foreign_handler)
+
+        try:
+            set_verbose_logging(True)
+            console_handlers = [
+                handler
+                for handler in log.handlers
+                if isinstance(handler, MVTLogHandler)
+            ]
+            assert console_handlers
+            assert all(handler.level == logging.DEBUG for handler in console_handlers)
+            assert foreign_handler.level == logging.CRITICAL
+
+            set_verbose_logging(False)
+            assert all(handler.level == logging.INFO for handler in console_handlers)
+            assert foreign_handler.level == logging.CRITICAL
+        finally:
+            log.handlers.remove(foreign_handler)