transformers/tests/utils/test_import_utils.py

import sys
from contextlib import contextmanager
from types import ModuleType
from unittest.mock import DEFAULT, MagicMock, patch

import pytest
from packaging.version import parse as parse_version
from parameterized import parameterized

from transformers import logging
from transformers.testing_utils import (
    CaptureLogger,
    LoggingLevel,
    require_torch,
    require_torchvision,
    run_test_using_subprocess,
)
from transformers.utils.import_utils import (
    _candidate_distribution_names,
    _is_package_available,
    _LazyModule,
    clear_import_cache,
    is_flash_attn_2_available,
    is_flash_attn_3_available,
)


@run_test_using_subprocess
def test_clear_import_cache():
    """Test the clear_import_cache function."""

    # Save initial state
    initial_modules = {name: mod for name, mod in sys.modules.items() if name.startswith("transformers.")}
    assert len(initial_modules) > 0, "No transformers modules loaded before test"

    # Execute clear_import_cache() function
    clear_import_cache()

    # Verify modules were removed
    remaining_modules = {name: mod for name, mod in sys.modules.items() if name.startswith("transformers.")}
    assert len(remaining_modules) < len(initial_modules), "No modules were removed"

    # Import and verify module exists
    from transformers.models.auto import modeling_auto

    assert "transformers.models.auto.modeling_auto" in sys.modules
    assert modeling_auto.__name__ == "transformers.models.auto.modeling_auto"


def test_is_package_available_edge_cases():
    pkg_name = "definitely_not_a_real_pkg_xyz"

    namespace_shadow = ModuleType(pkg_name)
    versionless_install = ModuleType(pkg_name)
    versionless_install.__file__ = f"/path/to/site-packages/{pkg_name}/__init__.py"
    with_version = ModuleType(pkg_name)
    with_version.__version__ = "1.2.3"

    cases = [
        (namespace_shadow, (False, "N/A")),
        (versionless_install, (True, "N/A")),
        (with_version, (True, "1.2.3")),
    ]
    for fake_module, expected in cases:
        with (
            patch("transformers.utils.import_utils.importlib.util.find_spec", return_value=object()),
            patch("transformers.utils.import_utils.importlib.import_module", return_value=fake_module),
        ):
            assert _is_package_available(pkg_name, return_version=True) == expected


def test_lazy_module_error_points_to_debug_log():
    logger = logging.get_logger("transformers.utils.import_utils")
    lazy_module = _LazyModule(
        "transformers.test_lazy_module",
        __file__,
        {"broken_module": ["BrokenObject"]},
    )

    original_error = RuntimeError("simulated broken dependency")

    with CaptureLogger(logger) as captured_logs:
        with LoggingLevel(logging.DEBUG), patch.object(lazy_module, "_get_module", side_effect=original_error):
            try:
                lazy_module.BrokenObject
            except ModuleNotFoundError as error:
                assert "Could not import module 'BrokenObject'" in str(error)
                assert "Set the logging verbosity to DEBUG for the original import error." in str(error)
                assert "simulated broken dependency" not in str(error)
                assert (
                    "Original import error for 'BrokenObject': simulated broken dependency"
                    in captured_logs.io.getvalue()
                )
            else:
                raise AssertionError("Expected ModuleNotFoundError")


def test_is_package_available_unmapped_distribution_does_not_import():
    """A package missing from `packages_distributions()` must be versioned from metadata, not by importing it.

    `torch` >= 2.14 ships no `top_level.txt`, so on Python < 3.12 it is absent from the mapping. Falling back
    to `importlib.import_module` there pulls all of torch into every `import transformers`.
    """
    pkg_name = "definitely_not_a_real_pkg_xyz"

    with (
        patch("transformers.utils.import_utils.importlib.util.find_spec", return_value=object()),
        patch("transformers.utils.import_utils.PACKAGE_DISTRIBUTION_MAPPING", {}),
        patch("transformers.utils.import_utils.importlib.metadata.version", return_value="1.2.3") as version,
        patch("transformers.utils.import_utils.importlib.import_module") as import_module,
    ):
        assert _is_package_available(pkg_name, return_version=True) == (True, "1.2.3")
        version.assert_called_once_with(pkg_name.replace("_", "-"))
        import_module.assert_not_called()


def test_candidate_distribution_names():
    with patch("transformers.utils.import_utils.PACKAGE_DISTRIBUTION_MAPPING", {"PIL": ["pillow"]}):
        # A mapped import name prefers its distribution, but keeps the import name as a fallback.
        assert _candidate_distribution_names("PIL") == ["pillow", "PIL"]
        # An unmapped import name falls back on itself, normalized first, without duplicates.
        assert _candidate_distribution_names("torch") == ["torch"]
        assert _candidate_distribution_names("torch_xla") == ["torch-xla", "torch_xla"]


@contextmanager
def mock_flash_attn_env(
    installed_packages: dict[str, str] | None = None,
    cuda_available: bool = False,
    kernels_available: bool = False,
    kernel_download_fails: bool = False,
):
    """Mock the environment probed by `is_flash_attn_{2,3}_available`. Args:
    - `installed_packages`: maps import names to versions, e.g. `{"flash_attn": "2.6.0"}`. The distribution name is
      assumed to match the import name (with underscores replaced by hyphens), except for `flash_attn_interface`
      which is distributed as `flash-attn-3`.
    - `cuda_available`: whether CUDA is available or not.
    - `kernels_available`: whether the kernels library is available.
    - `kernel_download_fails`: if this flag is set to True, the get_kernel method of the fake kernels module will raise
        a RuntimeError to simulate a kernel download failure.
    """
    installed_packages = {} if installed_packages is None else installed_packages
    distribution_names = {"flash_attn_interface": "flash-attn-3"}

    def fake_is_package_available(pkg_name: str, return_version: bool = False) -> tuple[bool, str]:
        is_available = pkg_name in installed_packages
        version = installed_packages.get(pkg_name, "N/A") if return_version else None
        return is_available, version

    fake_distribution_mapping = {
        pkg: [distribution_names.get(pkg, pkg.replace("_", "-"))] for pkg in installed_packages
    }
    fake_kernels_module = ModuleType("kernels")
    fake_kernels_module.get_kernel = MagicMock(
        side_effect=RuntimeError("kernel unavailable") if kernel_download_fails else None
    )

    is_flash_attn_2_available.cache_clear()
    is_flash_attn_3_available.cache_clear()
    try:
        with (
            patch("transformers.utils.import_utils._is_package_available", side_effect=fake_is_package_available),
            patch("transformers.utils.import_utils.PACKAGE_DISTRIBUTION_MAPPING", fake_distribution_mapping),
            patch("transformers.utils.import_utils.is_torch_cuda_available", return_value=cuda_available),
            patch("transformers.utils.import_utils.is_torch_mlu_available", return_value=False),
            patch("transformers.utils.import_utils.is_torch_musa_available", return_value=False),
            patch("transformers.utils.import_utils.is_kernels_available", return_value=kernels_available),
            patch.dict(sys.modules, {"kernels": fake_kernels_module}),
        ):
            yield fake_kernels_module.get_kernel
    finally:
        is_flash_attn_2_available.cache_clear()
        is_flash_attn_3_available.cache_clear()


@parameterized.expand([("2.0.0",), ("2.3.3",), ("2.6.0",)])
def test_flash_attn_2_available_with_package(version: str):
    # If the package version is below 2.3.3, the package is too old, and FA should be unavailable
    expected = parse_version(version) >= parse_version("2.3.3")

    with mock_flash_attn_env(installed_packages={"flash_attn": version}, cuda_available=True) as get_kernel:
        # Check the result is the expected one
        is_available = is_flash_attn_2_available()
        assert is_available == expected, (
            f"Expected is_flash_attn_2_available() to be {expected} but got {is_available}"
        )
        # Check the kernels fallback was not probed (kernels_fallback_ok default value is False)
        get_kernel.assert_not_called()
        # Ensure the kernels fallback is not probed (should not happen when the package is present and cuda available)
        assert is_flash_attn_2_available(kernels_fallback_ok=True) == expected
        get_kernel.assert_not_called()


def test_flash_attn_3_available_with_package():
    with mock_flash_attn_env(installed_packages={"flash_attn_interface": "3.0.0"}, cuda_available=True) as get_kernel:
        assert is_flash_attn_3_available()
        assert is_flash_attn_3_available(kernels_fallback_ok=True)
        get_kernel.assert_not_called()


@parameterized.expand(
    [(2, False, False), (2, True, False), (2, True, True), (3, False, False), (3, True, False), (3, True, True)]
)
def test_flash_attn_cuda_kernels_fallback(fa_version: int, kernels_available: bool, download_fails: bool):
    from transformers.integrations.hub_kernels import get_attn_kernel_version
    from transformers.modeling_flash_attention_utils import FLASH_ATTN_KERNEL_FALLBACK

    # Test is expected to pass only if the kernels library is available and the kernel download does not fail
    expected = kernels_available and not download_fails

    # Mock an env where the package is not available and kernels availability depends on the parameters
    with mock_flash_attn_env(kernels_available=kernels_available, kernel_download_fails=download_fails) as get_kernel:
        # Ensure the FA is not available without kernels fallback
        if fa_version == 2:
            assert not is_flash_attn_2_available()
        elif fa_version == 3:
            assert not is_flash_attn_3_available()
        else:
            raise ValueError(f"Invalid FA version: {fa_version}")

        # Check expected value
        if fa_version == 2:
            is_available = is_flash_attn_2_available(kernels_fallback_ok=True)
        elif fa_version == 3:
            is_available = is_flash_attn_3_available(kernels_fallback_ok=True)
        else:
            raise ValueError(f"Invalid FA version: {fa_version}")

        if is_available != expected:
            raise RuntimeError(
                f"Expected is_flash_attn_{fa_version}_available() to be {expected} but got {is_available}"
            )

        # Check the number of calls to get_kernel
        if kernels_available:
            repo_id = FLASH_ATTN_KERNEL_FALLBACK[f"flash_attention_{fa_version}"]
            get_kernel.assert_called_once_with(repo_id, version=get_attn_kernel_version(repo_id))
        else:
            get_kernel.assert_not_called()


def test_flash_attn_2_fallback_rescues_non_cuda_platform():
    # Package installed but no CUDA/MLU device (e.g. XPU): the kernels fallback should still kick in
    with mock_flash_attn_env(installed_packages={"flash_attn": "2.6.0"}, cuda_available=False, kernels_available=True):
        assert not is_flash_attn_2_available()
        assert is_flash_attn_2_available(kernels_fallback_ok=True)


def test_require_flash_attn_decorators_accept_kernels_fallback():
    # Smoke test: these decorators call is_flash_attn_2_available(kernels_fallback_ok=True) and must not raise
    from transformers.testing_utils import require_all_flash_attn, require_flash_attn

    class DummyTest:
        pass

    with mock_flash_attn_env(kernels_available=True):
        assert require_flash_attn(DummyTest) is not None
        assert require_all_flash_attn(DummyTest) is not None


@run_test_using_subprocess
def test_broken_torchaudio_does_not_break_import():
    """
    ``loss/loss_rnnt.py`` is imported eagerly from ``modeling_utils``, so it must NOT import torchaudio at
    module scope: a torchaudio whose compiled extension was built against a different torch ABI raises
    ``OSError`` on import, which would otherwise break ``import transformers`` -- and pytest collection for
    the whole suite (the daily quantization CI collapse, Jul 2026). torchaudio is imported lazily inside
    ``rnnt_loss`` instead, so:
      * importing the module (hence ``import transformers``) never touches torchaudio;
      * a broken install surfaces its own ``OSError`` at the call site -- we don't mask it;
      * a genuinely missing torchaudio yields a clean ``ImportError``.
    """
    import builtins

    import torch

    # Importing loss_rnnt (and thus transformers) must succeed regardless of torchaudio's state, and must
    # not have imported torchaudio at module scope.
    from transformers.loss import loss_rnnt
    from transformers.utils import import_utils

    assert not hasattr(loss_rnnt, "torchaudio"), "torchaudio must be imported lazily, not at module scope"

    # ``rnnt_loss`` is guarded by ``@requires(backends=("torchaudio",))``, which resolves availability
    # through ``BACKENDS_MAPPING`` at call time, so that is what has to be patched here.
    def patch_torchaudio_available(available: bool):
        error_message = import_utils.BACKENDS_MAPPING["torchaudio"][1]
        return patch.dict(import_utils.BACKENDS_MAPPING, {"torchaudio": (lambda: available, error_message)})

    def _call_rnnt_loss():
        loss_rnnt.rnnt_loss(
            logits=torch.zeros(1, 2, 3, 4),
            targets=torch.zeros(1, 3),
            logit_lengths=torch.ones(1),
            target_lengths=torch.ones(1),
            blank_token_id=0,
        )

    # torchaudio is installed (is_torchaudio_available() is True) but its C extension won't load: the raw
    # OSError must surface at the call site, not be swallowed.
    real_import = builtins.__import__

    def failing_import(name, *args, **kwargs):
        if name == "torchaudio" or name.startswith("torchaudio."):
            raise OSError("_torchaudio.abi3.so: undefined symbol: simulated_abi_mismatch")
        return real_import(name, *args, **kwargs)

    for name in list(sys.modules):
        if name == "torchaudio" or name.startswith("torchaudio."):
            del sys.modules[name]

    with (
        patch_torchaudio_available(True),
        patch.object(builtins, "__import__", failing_import),
    ):
        try:
            _call_rnnt_loss()
        except OSError:
            pass
        else:
            raise AssertionError("rnnt_loss must surface the torchaudio OSError at call time")

    # torchaudio genuinely absent: rnnt_loss raises a clean ImportError.
    with patch_torchaudio_available(False):
        try:
            _call_rnnt_loss()
        except ImportError:
            pass
        else:
            raise AssertionError("rnnt_loss must raise ImportError when torchaudio is unavailable")


@require_torch
@run_test_using_subprocess
def test_import_without_torch_distributed():
    """
    Checks that Transformers can still be imported and used when PyTorch was built with USE_DISTRIBUTED=0
    (e.g. AMD's Windows ROCm 7.2.1 wheels). This make sure that distributed guarding works correctly.
    """

    import torch

    # Forget transformers, so that importing it below actually re-runs its module-scope imports.
    for name in list(sys.modules):
        if name.startswith("transformers"):
            del sys.modules[name]

    # Emulate USE_DISTRIBUTED=0 by temporarily faking torch.distributed availability to False.
    dist_modules_to_remove = [
        name
        for name in list(sys.modules)
        if name.startswith(
            (
                "torch.distributed.tensor",
                "torch.distributed.checkpoint",
                "torch.distributed.fsdp",
                "torch.distributed._composable",
            )
        )
    ]

    with (
        patch.object(torch.distributed, "is_available", return_value=False),
        patch.dict(sys.modules, {"torch._C._distributed_c10d": None}),
        patch.dict(sys.modules, dict.fromkeys(dist_modules_to_remove, DEFAULT)),
    ):
        # If transformers import errors out, it means that the distributed guarding is not working correctly.
        from transformers import AutoImageProcessor  # noqa: F401


def _compile_constant_helpers():
    """Every helper carrying `@_make_compile_constant`, as (name, args) for the test below.

    Derived from the marker rather than hand-listed: marking a helper opts it into verification, so the
    two can never drift. Helpers needing arguments get them here; the rest are called with none.
    """
    import inspect

    import transformers.utils.import_utils as import_utils

    with_args = {"is_torch_greater_or_equal": ("2.5",), "is_torch_less_or_equal": ("99.0",)}
    cases = []
    for name in sorted(dir(import_utils)):
        fn = getattr(import_utils, name)
        if not getattr(fn, "_dynamo_marked_constant", False):
            continue
        if name in with_args:
            cases.append((name, with_args[name]))
            continue
        try:
            inspect.signature(fn).bind()  # skip anything needing args we have not supplied
        except (TypeError, ValueError):
            continue
        cases.append((name, ()))
    return cases


@require_torch
@parameterized.expand(_compile_constant_helpers())
def test_availability_helpers_are_compile_safe(helper_name: str, args: tuple):
    """
    These helpers get called from inside `torch.compile`d regions — e.g. `is_dtensor`, which every MoE
    kernel integration reaches through `to_local`. Each carries `@_make_compile_constant`, so dynamo evaluates
    it once at trace time and never enters the body; this checks the marker actually takes effect.

    Folding rather than keeping the bodies traceable is deliberate. Most bottom out in
    `_is_package_available`, whose `importlib.metadata` lookup dynamo cannot follow — and follows
    differently per Python version, so a body that traces on one interpreter breaks on another. An
    untraced body cannot break on any of them. `@lru_cache` is no protection either: dynamo steps past
    cache wrappers and traces the wrapped function, which is why the marker sits underneath the cache —
    above it, the marker is a silent no-op.

    Add a helper here when compiled code starts calling it. Two are deliberately excluded and must never
    be marked: `is_cuda_stream_capturing` and `is_torch_deterministic` genuinely change answer during a
    process, so folding a transient into the graph would be worse than the graph break.
    """
    import torch

    import transformers.utils.import_utils as import_utils

    helper = getattr(import_utils, helper_name)
    torch.compiler.reset()

    @torch.compile(fullgraph=True)
    def run(x):
        return x + 1 if helper(*args) else x - 1

    run(torch.zeros(3))  # a graph break inside the helper would raise here


@run_test_using_subprocess
def test_legacy_module_aliases_are_served_on_demand():
    """
    Deprecated module paths (`transformers.tokenization_utils_fast`, `transformers.models.x.image_processing_x_fast`,
    ...) are served by a `sys.meta_path` finder and must not be pre-registered in `sys.modules`: tools that scan
    `sys.modules` and probe attributes (unittest's `assertWarns`, `pickle`, `inspect`) would otherwise trigger imports
    of optional heavy dependencies. Regression test for https://github.com/huggingface/transformers/issues/48966.
    """
    import unittest
    import warnings

    import transformers
    from transformers.utils import import_utils

    finders = [f for f in sys.meta_path if isinstance(f, import_utils._LegacyModuleAliasFinder)]
    assert len(finders) == 1
    finder = finders[0]
    pre_registered = [name for name in sys.modules if finder.resolve(name) is not None]
    assert pre_registered == [], f"legacy module paths pre-registered in sys.modules: {pre_registered}"
    assert not any(name.startswith("image_processing_") for name in vars(transformers))

    # Scanning `sys.modules` (what `assertWarns` does on entry) must not import anything.
    modules_before = len(sys.modules)
    with unittest.TestCase().assertWarns(UserWarning):
        warnings.warn("probe", UserWarning)
    assert len(sys.modules) == modules_before

    # Legacy paths still work, as thin shims over the replacement module.
    from transformers.tokenization_utils_fast import PreTrainedTokenizerFast

    assert PreTrainedTokenizerFast is transformers.PreTrainedTokenizerFast
    shim = sys.modules["transformers.tokenization_utils_fast"]
    assert shim is transformers.tokenization_utils_fast
    assert shim.__spec__.loader is finder
    assert "PreTrainedTokenizerFast" in vars(shim)
    # Module metadata is the shim's own, so attribute probes behave like on any other module.
    assert getattr(shim, "__warningregistry__", None) is None
    assert getattr(shim, "does_not_exist", None) is None

    with pytest.raises(ModuleNotFoundError):
        import transformers.models.aria.image_processing_nope_fast  # noqa: F401


@require_torchvision
def test_legacy_fast_image_processor_module_alias():
    """`models/x/image_processing_x_fast.py` became `models/x/image_processing_x.py`; the old path is an alias."""
    import transformers.models.aria.image_processing_aria_fast as legacy_module
    from transformers.models.aria.image_processing_aria import AriaImageProcessor
    from transformers.models.aria.image_processing_aria_fast import AriaImageProcessorFast  # legacy class name

    assert AriaImageProcessorFast is AriaImageProcessor
    assert legacy_module.AriaImageProcessor is AriaImageProcessor
    assert legacy_module.__all__ == ["AriaImageProcessor"]

    # Only paths that actually existed are aliased.
    with pytest.raises(ModuleNotFoundError):
        import transformers.models.aria.image_processing_pil_aria_fast  # noqa: F401