mvt/tests/common/test_command.py

# Mobile Verification Toolkit (MVT)
# Copyright (c) 2021-2023 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 threading
import time
from io import StringIO

import pytest
from rich.console import Console

from mvt.common.command import Command
from mvt.common.config import settings
from mvt.common.log import MVTLogHandler
from mvt.common.module import EncryptedBackupError, MVTModule
from unittest.mock import patch

from click.testing import CliRunner

from mvt.common.indicators import Indicators


class RecordingModule(MVTModule):
    run_order: list[str] = []

    def run(self):
        self.run_order.append(self.__class__.__name__)

    def check_indicators(self):
        pass


class FirstModule(RecordingModule):
    def run(self):
        super().run()
        self.results = ["first"]


class SecondModule(RecordingModule):
    dependencies = (FirstModule,)

    def run(self):
        super().run()
        self.results = self.get_dependency_results(FirstModule) + ["second"]


class ThirdModule(RecordingModule):
    dependencies = (SecondModule,)


class IndependentModule(RecordingModule):
    pass


class URLRecordingModule(RecordingModule):
    def collect_url_results(self):
        self.add_url_result(
            "https://example.org/message",
            "2026-07-29 12:00:00.000000",
            "test-chat",
        )


class CustomIOSBackupModule(RecordingModule):
    supported_commands = (("ios", "check-backup"),)


class CustomIOSFSModule(RecordingModule):
    supported_commands = (("ios", "check-fs"),)


class UnscopedCustomModule(RecordingModule):
    pass


class CustomDependsOnBuiltin(RecordingModule):
    supported_commands = (("ios", "check-backup"),)
    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

    def module_init(self, module):
        pass

    def finish(self):
        pass


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 = []

    def test_store_alerts_handles_bytes(self, tmp_path):
        cmd = Command(results_path=str(tmp_path))
        cmd.alertstore.medium(
            "bytes event",
            "",
            {"payload": b"\xa8\xa9"},
        )

        cmd._store_alerts()

        alerts = json.loads((tmp_path / "alerts.json").read_text())
        assert alerts[0]["event"]["payload"] == "\\xa8\\xa9"

    def test_timeline_deduplication_preserves_first_seen_order(self):
        timeline = [
            {"timestamp": "same", "event": "first"},
            {"timestamp": "same", "event": "second"},
            {"timestamp": "same", "event": "first"},
        ]

        assert MVTModule._deduplicate_timeline(timeline) == timeline[:2]

    def test_stores_collected_urls(self, tmp_path):
        cmd = RecordingCommand(results_path=str(tmp_path))
        cmd.modules = [URLRecordingModule]

        cmd.run()

        assert json.loads((tmp_path / "urls.json").read_text()) == [
            {
                "url": "https://example.org/message",
                "expanded_url": None,
                "timestamp": "2026-07-29 12:00:00.000000",
                "source": "test-chat",
            }
        ]

    def test_modules_run_in_stable_topological_order(self):
        cmd = RecordingCommand(jobs=1)
        cmd.modules = [ThirdModule, IndependentModule, SecondModule, FirstModule]

        cmd.run()

        assert RecordingModule.run_order == [
            "IndependentModule",
            "FirstModule",
            "SecondModule",
            "ThirdModule",
        ]
        second = next(module for module in cmd.executed if isinstance(module, SecondModule))
        assert second.results == ["first", "second"]

    def test_selected_module_runs_transitive_dependencies(self):
        cmd = RecordingCommand(module_name="ThirdModule", jobs=1)
        cmd.modules = [ThirdModule, SecondModule, FirstModule, IndependentModule]

        cmd.run()

        assert RecordingModule.run_order == [
            "FirstModule",
            "SecondModule",
            "ThirdModule",
        ]

    @pytest.mark.parametrize("jobs", [1, 2])
    @pytest.mark.parametrize("parallel_safe", [False, True])
    def test_initialization_hook_sees_completed_dependencies(self, jobs, parallel_safe):
        class Prerequisite(FirstModule):
            pass

        Prerequisite.parallel_safe = parallel_safe

        class Dependent(SecondModule):
            dependencies = (Prerequisite,)

            def run(self):
                self.results = self.get_dependency_results(Prerequisite) + ["second"]

        class HookCommand(RecordingCommand):
            def module_init(self, module):
                if isinstance(module, Dependent):
                    self.seen = [
                        executed.results
                        for executed in self.executed
                        if isinstance(executed, Prerequisite)
                    ]

        cmd = HookCommand(jobs=jobs)
        cmd.modules = [Dependent, Prerequisite]

        cmd.run()

        assert cmd.seen == [["first"]]
        assert [type(module) for module in cmd.executed] == [Prerequisite, Dependent]
        assert cmd.executed[1].results == ["first", "second"]

    def test_circular_dependency_warns_and_stops(self, caplog):
        class CircularOne(RecordingModule):
            pass

        class CircularTwo(RecordingModule):
            dependencies = (CircularOne,)

        CircularOne.dependencies = (CircularTwo,)

        cmd = RecordingCommand()
        cmd.modules = [CircularOne, CircularTwo]

        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_unavailable_dependency_only_skips_the_dependent_module(self, caplog):
        class UnavailableModule(RecordingModule):
            pass

        class DependentModule(RecordingModule):
            dependencies = (UnavailableModule,)

        cmd = RecordingCommand()
        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 "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()
        cmd.platform = "ios"
        cmd.name = "check-backup"
        cmd.modules = [FirstModule]
        cmd.custom_modules = [
            CustomIOSBackupModule,
            CustomIOSFSModule,
            UnscopedCustomModule,
        ]

        assert [module.__name__ for module in cmd._ordered_modules()] == [
            "FirstModule",
            "CustomIOSBackupModule",
        ]

    def test_selected_custom_module_runs(self):
        cmd = RecordingCommand(module_name="CustomIOSBackupModule", jobs=1)
        cmd.platform = "ios"
        cmd.name = "check-backup"
        cmd.custom_modules = [CustomIOSBackupModule]

        cmd.run()

        assert RecordingModule.run_order == ["CustomIOSBackupModule"]

    def test_selected_unsupported_custom_module_does_not_run(self):
        cmd = RecordingCommand(module_name="CustomIOSFSModule")
        cmd.platform = "ios"
        cmd.name = "check-backup"
        cmd.custom_modules = [CustomIOSFSModule]

        cmd.run()

        assert RecordingModule.run_order == []

    def test_custom_module_dependencies_use_topological_order(self):
        cmd = RecordingCommand(module_name="CustomDependsOnBuiltin", jobs=1)
        cmd.platform = "ios"
        cmd.name = "check-backup"
        cmd.modules = [SecondModule, FirstModule]
        cmd.custom_modules = [CustomDependsOnBuiltin]

        cmd.run()

        assert RecordingModule.run_order == ["FirstModule", "CustomDependsOnBuiltin"]

    def test_parallel_modules_overlap_and_respect_worker_limit(self):
        lock = threading.Lock()
        barrier = threading.Barrier(2)
        active = 0
        maximum = 0
        overlapped = []

        class ParallelModule(RecordingModule):
            def run(self):
                nonlocal active, maximum
                with lock:
                    active += 1
                    maximum = max(maximum, active)
                try:
                    barrier.wait(timeout=2)
                    overlapped.append(self.__class__.__name__)
                    time.sleep(0.02)
                finally:
                    with lock:
                        active -= 1

        class ParallelOne(ParallelModule):
            pass

        class ParallelTwo(ParallelModule):
            pass

        class ParallelThree(ParallelModule):
            pass

        cmd = RecordingCommand(jobs=2)
        cmd.modules = [ParallelOne, ParallelTwo, ParallelThree]
        cmd.run()

        assert maximum == 2
        assert {"ParallelOne", "ParallelTwo"}.issubset(overlapped)

    def test_dependencies_wait_and_transitive_dependencies_run_once(self):
        prerequisite_finished = threading.Event()
        runs = []

        class Prerequisite(RecordingModule):
            def run(self):
                runs.append("prerequisite")
                self.results = ["ready"]
                prerequisite_finished.set()

        class Dependent(RecordingModule):
            dependencies = (Prerequisite,)

            def run(self):
                assert prerequisite_finished.is_set()
                assert self.get_dependency_results(Prerequisite) == ["ready"]
                runs.append(self.__class__.__name__)

        class DependentOne(Dependent):
            pass

        class DependentTwo(Dependent):
            pass

        cmd = RecordingCommand(jobs=3)
        cmd.modules = [DependentOne, DependentTwo, Prerequisite]
        cmd.run()

        assert runs.count("prerequisite") == 1
        assert set(runs[1:]) == {"DependentOne", "DependentTwo"}

    def test_parallel_unsafe_module_runs_exclusively(self):
        lock = threading.Lock()
        active = 0
        unsafe_ran_exclusively = []

        class SafeOne(RecordingModule):
            def run(self):
                nonlocal active
                with lock:
                    active += 1
                time.sleep(0.03)
                with lock:
                    active -= 1

        class Unsafe(RecordingModule):
            parallel_safe = False

            def run(self):
                nonlocal active
                with lock:
                    unsafe_ran_exclusively.append(active == 0)
                    active += 1
                time.sleep(0.01)
                with lock:
                    active -= 1

        class SafeTwo(SafeOne):
            pass

        cmd = RecordingCommand(jobs=3)
        cmd.modules = [SafeOne, Unsafe, SafeTwo]
        cmd.run()

        assert unsafe_ran_exclusively == [True]

    def test_results_are_aggregated_in_topological_order(self):
        release_first = threading.Event()

        class SlowFirst(RecordingModule):
            def run(self):
                release_first.wait(timeout=2)
                self.results = ["first"]
                self.timeline = [{"module": "first"}]
                self.alertstore.info("first", "", {"order": 1})

        class FastSecond(RecordingModule):
            def run(self):
                self.results = ["second"]
                self.timeline = [{"module": "second"}]
                self.alertstore.info("second", "", {"order": 2})
                release_first.set()

        cmd = RecordingCommand(jobs=2)
        cmd.modules = [SlowFirst, FastSecond]
        cmd.run()

        assert [type(module) for module in cmd.executed] == [SlowFirst, FastSecond]
        assert [entry["module"] for entry in cmd.timeline] == ["first", "second"]
        assert [alert.message for alert in cmd.alertstore.alerts] == ["first", "second"]

    def test_parallel_console_logs_are_grouped_and_file_log_is_complete(
        self, tmp_path
    ):
        output = StringIO()
        handler = MVTLogHandler(
            console=Console(file=output, force_terminal=False, color_system=None)
        )
        handler.setFormatter(logging.Formatter("%(message)s"))
        logger = logging.getLogger("mvt")
        logger.setLevel(logging.DEBUG)
        logger.addHandler(handler)
        release_slow = threading.Event()

        class SlowLog(RecordingModule):
            def run(self):
                self.log.info("slow-a")
                release_slow.wait(timeout=2)
                self.log.info("slow-b")

        class FastLog(RecordingModule):
            def run(self):
                self.log.info("fast-a")
                self.log.info("fast-b")
                release_slow.set()

        SlowLog.__module__ = "mvt.tests.parallel"
        FastLog.__module__ = "mvt.tests.parallel"
        try:
            cmd = RecordingCommand(results_path=str(tmp_path), jobs=2)
            cmd.modules = [SlowLog, FastLog]
            cmd.run()
        finally:
            logger.removeHandler(handler)

        console_output = output.getvalue()
        fast_block = (
            console_output.index("--- FastLog ---"),
            console_output.index("fast-a"),
            console_output.index("fast-b"),
        )
        slow_block = (
            console_output.index("--- SlowLog ---"),
            console_output.index("slow-a"),
            console_output.index("slow-b"),
        )
        assert fast_block == tuple(sorted(fast_block))
        assert slow_block == tuple(sorted(slow_block))
        assert fast_block[-1] < slow_block[0] or slow_block[-1] < fast_block[0]
        file_output = (tmp_path / "command.log").read_text()
        for message in ("slow-a", "slow-b", "fast-a", "fast-b"):
            assert message in file_output

    def test_encrypted_backup_stops_scheduling_and_cleans_up(self, caplog):
        finish_called = []

        class Encrypted(RecordingModule):
            def run(self):
                raise EncryptedBackupError

        class NeverScheduled(RecordingModule):
            dependencies = (Encrypted,)

        class CleanupCommand(RecordingCommand):
            def finish(self):
                finish_called.append(True)

        cmd = CleanupCommand(jobs=2)
        cmd.modules = [Encrypted, NeverScheduled]
        with caplog.at_level(logging.CRITICAL):
            cmd.run()

        assert finish_called == [True]
        assert not any(isinstance(module, NeverScheduled) for module in cmd.executed)
        assert "backup appears to be encrypted" in caplog.text

    def test_profiling_forces_sequential_execution(self, monkeypatch, caplog):
        run_order = []

        class ProfileOne(RecordingModule):
            def run(self):
                run_order.append("one")

        class ProfileTwo(RecordingModule):
            def run(self):
                run_order.append("two")

        monkeypatch.setattr(settings, "PROFILE", True)
        cmd = RecordingCommand(jobs=4)
        cmd.modules = [ProfileOne, ProfileTwo]
        with caplog.at_level(logging.WARNING):
            cmd.run()

        assert run_order == ["one", "two"]
        assert "forcing sequential module execution" in caplog.text
    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