41a6cc8692`TestVisionCacheOnException::test_exception_result_cached` currently
patches `load_model_config` with `side_effect=OSError("network down")`
and asserts `assert_called_once()`. That assertion is impossible by
design: `_is_vision_model_uncached` in
`studio/backend/utils/models/model_config.py` intentionally returns
`None` for `OSError` so `is_vision_model` does not cache the fallback
and retries on the next call. The module docstring on
`_vision_detection_cache` itself spells this out:
Only definitive results (True/False from successful detection) are
cached; transient failures (network errors, timeouts) are NOT
cached so they can be retried.
The test has been failing identically on every downstream review run
against `unslothai/unsloth` main (e.g. `unsloth#5115`, `unsloth#5080`),
but the failure is not introduced by any of those PRs and does not
gate correctness.
Fix the collision by splitting the class into the two contracts the
code actually implements:
1. `test_permanent_exception_result_cached` keeps the original
intent ("exception falls back to False and that False is cached")
but uses `ValueError`, which is one of the exception types
`_is_vision_model_uncached` treats as permanent and caches. No
`huggingface_hub` import needed.
2. `test_transient_exception_not_cached` pins the opposite contract
with the original `OSError("network down")`: the call returns
False but the second invocation re-runs detection
(`call_count == 2`). This guards against a future regression
where somebody caches transient failures and then users with a
flaky network permanently see wrong detection for a model.
Both tests use `assert ... is False` on the public API and mock-count
assertions on `load_model_config`; no private helpers are touched.
265 lines
12 KiB
Python
265 lines
12 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""Tests for is_vision_model() caching behaviour.
|
|
|
|
The vision detection cache (``_vision_detection_cache``) mirrors the existing
|
|
``_audio_detection_cache`` pattern used by ``detect_audio_type()``. These
|
|
tests verify that:
|
|
|
|
* Repeated calls for the same model hit the cache (no redundant work).
|
|
* Different models each trigger their own detection.
|
|
* Both True and False results are cached.
|
|
* The subprocess path (transformers 5.x models) is also cached.
|
|
* Exceptions that fall back to False are cached.
|
|
"""
|
|
|
|
import sys
|
|
import types as _types
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# sys.path + logger stub — same pattern as the rest of the test suite
|
|
# ---------------------------------------------------------------------------
|
|
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
|
|
if _BACKEND_DIR not in sys.path:
|
|
sys.path.insert(0, _BACKEND_DIR)
|
|
|
|
_loggers_stub = _types.ModuleType("loggers")
|
|
_loggers_stub.get_logger = lambda name: __import__("logging").getLogger(name)
|
|
sys.modules.setdefault("loggers", _loggers_stub)
|
|
|
|
from utils.models.model_config import (
|
|
is_vision_model,
|
|
_is_vision_model_uncached,
|
|
_vision_detection_cache,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(autouse = True)
|
|
def _clear_vision_cache():
|
|
"""Ensure every test starts with a fresh cache."""
|
|
_vision_detection_cache.clear()
|
|
yield
|
|
_vision_detection_cache.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cache hit / miss tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVisionCacheHitMiss:
|
|
"""Verify the cache prevents redundant detection calls."""
|
|
|
|
@patch("utils.models.model_config._is_vision_model_uncached", return_value = True)
|
|
def test_second_call_uses_cache(self, mock_uncached):
|
|
"""Calling is_vision_model() twice for the same model should invoke
|
|
the uncached function only once."""
|
|
assert is_vision_model("org/my-vlm") is True
|
|
assert is_vision_model("org/my-vlm") is True
|
|
mock_uncached.assert_called_once_with("org/my-vlm", None)
|
|
|
|
@patch("utils.models.model_config._is_vision_model_uncached", return_value = False)
|
|
def test_different_models_each_detected(self, mock_uncached):
|
|
"""Different model names should each trigger detection."""
|
|
is_vision_model("model-a")
|
|
is_vision_model("model-b")
|
|
assert mock_uncached.call_count == 2
|
|
|
|
@patch("utils.models.model_config._is_vision_model_uncached", return_value = True)
|
|
def test_cache_returns_correct_value(self, mock_uncached):
|
|
"""The cached value must match what _is_vision_model_uncached returned."""
|
|
first = is_vision_model("org/vlm")
|
|
second = is_vision_model("org/vlm")
|
|
assert first is True
|
|
assert second is True
|
|
|
|
|
|
class TestVisionCacheStoresFalse:
|
|
"""Non-VLM results (False) must also be cached to avoid re-detection."""
|
|
|
|
@patch("utils.models.model_config._is_vision_model_uncached", return_value = False)
|
|
def test_false_result_cached(self, mock_uncached):
|
|
assert is_vision_model("org/text-only") is False
|
|
assert is_vision_model("org/text-only") is False
|
|
mock_uncached.assert_called_once()
|
|
assert _vision_detection_cache[("org/text-only", None)] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subprocess path (transformers 5.x) caching
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVisionCacheSubprocessPath:
|
|
"""Models needing transformers 5.x go through _is_vision_model_subprocess.
|
|
The cache should prevent the subprocess from being spawned more than once
|
|
per model per process."""
|
|
|
|
@patch("utils.models.model_config._is_vision_model_subprocess", return_value = True)
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = True)
|
|
def test_subprocess_called_once_with_cache(self, mock_needs_t5, mock_subprocess):
|
|
"""Subprocess should only fire on the first call; second is cached."""
|
|
# First call: goes through uncached → subprocess
|
|
assert is_vision_model("unsloth/Qwen3.5-2B") is True
|
|
# Second call: cache hit, no subprocess
|
|
assert is_vision_model("unsloth/Qwen3.5-2B") is True
|
|
|
|
mock_subprocess.assert_called_once()
|
|
assert _vision_detection_cache[("unsloth/Qwen3.5-2B", None)] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Exception handling — cache the False fallback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVisionCacheOnException:
|
|
"""When detection raises an exception, _is_vision_model_uncached
|
|
distinguishes permanent failures (cached as False) from transient
|
|
failures (returned as None, not cached so the next call can retry).
|
|
Verify both contracts."""
|
|
|
|
@patch(
|
|
"utils.models.model_config.load_model_config",
|
|
side_effect = ValueError("bad config"),
|
|
)
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = False)
|
|
def test_permanent_exception_result_cached(self, mock_needs_t5, mock_load_config):
|
|
"""A permanent failure (ValueError / RepositoryNotFoundError /
|
|
GatedRepoError / JSONDecodeError) should be caught, return False,
|
|
and that False should be cached so subsequent calls don't retry.
|
|
|
|
ValueError is used here because it's the simplest of the
|
|
code-path's cacheable exception types and does not require an
|
|
import of huggingface_hub errors (whose module path varies
|
|
across versions)."""
|
|
# First call: load_model_config raises -> except branch -> False.
|
|
assert is_vision_model("broken/model") is False
|
|
# Second call: cache hit, load_model_config not called again.
|
|
assert is_vision_model("broken/model") is False
|
|
mock_load_config.assert_called_once()
|
|
|
|
@patch(
|
|
"utils.models.model_config.load_model_config",
|
|
side_effect = OSError("network down"),
|
|
)
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = False)
|
|
def test_transient_exception_not_cached(self, mock_needs_t5, mock_load_config):
|
|
"""A transient failure (OSError, timeouts) should return None from
|
|
_is_vision_model_uncached, surface as False to the caller, and
|
|
NOT be cached, so the next call retries detection. This matches
|
|
the documented behaviour on _vision_detection_cache:
|
|
'transient failures (network errors, timeouts) are NOT cached so
|
|
they can be retried.'"""
|
|
# First call: load_model_config raises OSError -> uncached None
|
|
# -> caller returns False without caching.
|
|
assert is_vision_model("broken/model") is False
|
|
# Second call: cache miss again, load_model_config called a
|
|
# second time.
|
|
assert is_vision_model("broken/model") is False
|
|
assert mock_load_config.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Direct detection path (non-transformers-5 models) caching
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVisionCacheDirectPath:
|
|
"""For models that do NOT need transformers 5.x, the detection goes through
|
|
load_model_config directly. The cache must work the same way."""
|
|
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = False)
|
|
@patch("utils.models.model_config.load_model_config")
|
|
def test_direct_vlm_detection_cached(self, mock_load_config, mock_needs_t5):
|
|
"""A standard VLM detected via architecture suffix should be cached."""
|
|
cfg = MagicMock(spec = []) # strict: only explicitly set attrs exist
|
|
cfg.model_type = "gemma3"
|
|
cfg.architectures = ["Gemma3ForConditionalGeneration"]
|
|
mock_load_config.return_value = cfg
|
|
|
|
assert is_vision_model("google/gemma-3-4b-it") is True
|
|
assert is_vision_model("google/gemma-3-4b-it") is True
|
|
# load_model_config should only be called once
|
|
mock_load_config.assert_called_once()
|
|
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = False)
|
|
@patch("utils.models.model_config.load_model_config")
|
|
def test_direct_non_vlm_detection_cached(self, mock_load_config, mock_needs_t5):
|
|
"""A standard text model (no VLM indicators) should cache False."""
|
|
cfg = MagicMock(spec = []) # spec=[] means no attributes at all
|
|
cfg.model_type = "llama"
|
|
cfg.architectures = ["LlamaForCausalLM"]
|
|
mock_load_config.return_value = cfg
|
|
|
|
# LlamaForCausalLM doesn't end with VLM suffixes, no vision_config, etc.
|
|
assert is_vision_model("meta-llama/Llama-3-8B") is False
|
|
assert is_vision_model("meta-llama/Llama-3-8B") is False
|
|
mock_load_config.assert_called_once()
|
|
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = False)
|
|
@patch("utils.models.model_config.load_model_config")
|
|
def test_vision_config_attr_detected_and_cached(
|
|
self, mock_load_config, mock_needs_t5
|
|
):
|
|
"""Models with vision_config (LLaVA, Qwen2-VL, etc.) should be cached as True."""
|
|
cfg = MagicMock(spec = []) # strict: only explicitly set attrs exist
|
|
cfg.model_type = "qwen2_vl"
|
|
cfg.architectures = ["Qwen2VLForCausalLM"] # Doesn't match VLM suffixes
|
|
cfg.vision_config = {"hidden_size": 1024}
|
|
mock_load_config.return_value = cfg
|
|
|
|
assert is_vision_model("Qwen/Qwen2-VL-7B") is True
|
|
assert is_vision_model("Qwen/Qwen2-VL-7B") is True
|
|
mock_load_config.assert_called_once()
|
|
|
|
@patch("utils.transformers_version.needs_transformers_5", return_value = False)
|
|
@patch("utils.models.model_config.load_model_config")
|
|
def test_audio_model_excluded_and_cached(self, mock_load_config, mock_needs_t5):
|
|
"""Audio-only models (csm, whisper) with ForConditionalGeneration
|
|
should be excluded from VLM detection and cached as False."""
|
|
cfg = MagicMock(spec = []) # strict: only explicitly set attrs exist
|
|
cfg.model_type = "whisper"
|
|
cfg.architectures = ["WhisperForConditionalGeneration"]
|
|
mock_load_config.return_value = cfg
|
|
|
|
assert is_vision_model("openai/whisper-large-v3") is False
|
|
assert is_vision_model("openai/whisper-large-v3") is False
|
|
mock_load_config.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# hf_token handling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVisionCacheTokenHandling:
|
|
"""The cache is keyed on (model_name, hf_token).
|
|
Different tokens for the same model should trigger separate detections
|
|
to handle gated models correctly."""
|
|
|
|
@patch("utils.models.model_config._is_vision_model_uncached", return_value = True)
|
|
def test_different_tokens_trigger_new_detection(self, mock_uncached):
|
|
"""Calls with different tokens should trigger separate detections to
|
|
handle gated models correctly (e.g. unauthenticated probe → False,
|
|
then authenticated call should re-check)."""
|
|
assert is_vision_model("gated/model", hf_token = "token-a") is True
|
|
assert is_vision_model("gated/model", hf_token = "token-b") is True
|
|
assert mock_uncached.call_count == 2
|
|
|
|
@patch("utils.models.model_config._is_vision_model_uncached", return_value = True)
|
|
def test_same_token_uses_cache(self, mock_uncached):
|
|
"""Repeated calls with identical model + token should hit cache."""
|
|
assert is_vision_model("gated/model", hf_token = "token-a") is True
|
|
assert is_vision_model("gated/model", hf_token = "token-a") is True
|
|
mock_uncached.assert_called_once()
|