vrr/unsloth
2
0
Fork 0
mirror of https://github.com/unslothai/unsloth.git synced 2026-08-25 08:42:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 29
unsloth/studio/backend/tests/test_transformers_version.py
Daniel Han 2fd40d3a3f
Stop the sidecar tests writing into the real Studio install (#9131) 
* Stop the sidecar tests writing into the real Studio install

Three tests in TestDamagedLatestSidecarRepairHandoff fail on a clean checkout of
main, in complete isolation, on any machine that has run this file before:

    assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
    AssertionError: assert '530' == 'latest'

The cause is this file. _patch redirected only _VENV_T5_LATEST_DIR, while
_fake_install writes a sidecar at whatever target it is handed and _probe_tier
walks the tiers provisioning each one it tries. So the 530, 550 and 510 targets
were the real ones, and running the suite wrote a fake transformers 5.99.0 whose
CONFIG_MAPPING_NAMES is {"brandnew": "C"} into ~/.unsloth/studio. On the next
run _lowest_tier_for("brandnew") finds it in tier 530 and returns "530" where the
test asserts "latest".

Confirmed by content and mtime on my own machine, then reproduced from scratch in
a throwaway HOME: a pristine home stays clean and passes, and a home carrying the
markers a real install leaves gets the fake sidecar written into it, after which
the same three tests fail. That is also what CI hit, since a runner accumulates
the same state inside a single session.

Every tier constant now points under tmp_path. Derived from the module rather
than listed, because listing them is the mistake one level down: the list has to
be updated by whoever adds a tier and the consequence of forgetting is invisible
until some later, unrelated test reads the tier mapping. Deriving it immediately
turned up _VENV_T5_DIR, a fifth constant aliasing the 550 sidecar that a
hand-written list of the three obvious ones missed.

The new guard asserts the redirection over whatever constants the module defines,
so a fifth tier fails here rather than somewhere else on someone's laptop. The
file now runs twice in a row with the same result, which it did not before.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2026-08-17 18:47:28 -07:00

4271 lines
186 KiB
Python
Raw Permalink Blame History

# 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 transformers version detection with local checkpoint fallbacks."""
import json
import logging
import os
import pytest
from pathlib import Path
from unittest.mock import patch
# ---------------------------------------------------------------------------
# The studio backend uses relative-style imports (``from utils.…``), so
# add the backend directory to *sys.path* if not already present.
# ---------------------------------------------------------------------------
import sys
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
if _BACKEND_DIR not in sys.path:
sys.path.insert(0, _BACKEND_DIR)
# Stub the custom logger before import so ``from loggers import
# get_logger`` doesn't fail.
import types as _types
_loggers_stub = _types.ModuleType("loggers")
_loggers_stub.get_logger = lambda name: __import__("logging").getLogger(name)
sys.modules.setdefault("loggers", _loggers_stub)
from utils.transformers_version import (
_resolve_base_model,
_is_lora_adapter_dir,
_has_adapter_weights,
_remote_lora_base,
_check_tokenizer_config_needs_v5,
_check_config_needs_510,
_check_config_needs_530,
_check_config_needs_550,
_config_needs_510,
_config_needs_530,
_norm_separators,
_tier_from_name,
_looks_like_hf_id,
_nemotron_h_needs_mlp_support,
_config_json_from_hf_cache,
_load_config_json,
_higher_tier,
_config_json_cache,
_tokenizer_class_cache,
_config_mapping_cache,
_config_needs_510_cache,
_config_needs_530_cache,
_config_needs_550_cache,
_probe_tier_cache,
_probe_tier,
_stderr_is_transient,
needs_transformers_5,
get_transformers_tier,
activate_transformers_for_subprocess,
_venv_dir_is_valid,
_venv_dir_is_valid_and_undamaged,
_CURRENT_EXT_TAG,
_ensure_venv_dir,
hf_endpoint_unreachable,
)
@pytest.fixture(autouse = True)
def _no_ambient_proxy(monkeypatch):
"""Module-wide: these tests patch urlopen, which a selected proxy opener bypasses.
Every remote read here goes through ``_hf_urlopen``, which calls ``opener.open`` when
a proxy applies. A runner with HTTPS_PROXY / ALL_PROXY set would then make a real
request instead of hitting the patch, so results would track ambient CI connectivity.
"""
for key in (
"HTTP_PROXY",
"HTTPS_PROXY",
"ALL_PROXY",
"NO_PROXY",
"http_proxy",
"https_proxy",
"all_proxy",
"no_proxy",
):
monkeypatch.delenv(key, raising = False)
@pytest.fixture(autouse = True)
def _capturable_logger(monkeypatch):
"""Make the ``caplog`` assertions independent of test collection order.
The ``sys.modules.setdefault("loggers", ...)`` stub above only installs the
stdlib-logger stub when ``loggers`` has not been imported yet. In a full
backend pytest run another module (e.g. ``test_log_filter_no_truncation``,
collected earlier) imports the real ``loggers`` first, so the stub is a
no-op and ``transformers_version.logger`` ends up a structlog/stdout logger
that ``caplog`` cannot see -- the tier/activation/install log assertions
would then fail even though the line was emitted. Bind a real stdlib logger
for the duration of each test so the module logs through ``logging`` and
``caplog`` captures them regardless of import order.
"""
monkeypatch.setattr(
"utils.transformers_version.logger",
logging.getLogger("utils.transformers_version"),
)
# ---------------------------------------------------------------------------
# _resolve_base_model — config.json fallback
# ---------------------------------------------------------------------------
class TestResolveBaseModel:
"""Tests for _resolve_base_model() local config fallbacks."""
def test_adapter_config_takes_priority(self, tmp_path: Path):
"""adapter_config.json should be preferred over config.json."""
adapter_cfg = {"base_model_name_or_path": "meta-llama/Llama-3-8B"}
config_cfg = {"_name_or_path": "different/model"}
(tmp_path / "adapter_config.json").write_text(json.dumps(adapter_cfg))
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
result = _resolve_base_model(str(tmp_path))
assert result == "meta-llama/Llama-3-8B"
def test_config_json_fallback_model_name(self, tmp_path: Path):
"""config.json model_name should resolve when no adapter_config."""
config_cfg = {"model_name": "Qwen/Qwen3.5-9B"}
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
result = _resolve_base_model(str(tmp_path))
assert result == "Qwen/Qwen3.5-9B"
def test_config_json_fallback_name_or_path(self, tmp_path: Path):
"""config.json _name_or_path should resolve as secondary fallback."""
config_cfg = {"_name_or_path": "Qwen/Qwen3.5-9B"}
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
result = _resolve_base_model(str(tmp_path))
assert result == "Qwen/Qwen3.5-9B"
def test_non_string_base_does_not_crash(self, tmp_path: Path):
"""A malformed config (list/dict for model_name) must not raise."""
config_cfg = {"model_name": ["x"], "_name_or_path": "Qwen/Qwen3.5-9B"}
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
# Skips the non-string model_name and falls through to _name_or_path.
assert _resolve_base_model(str(tmp_path)) == "Qwen/Qwen3.5-9B"
def test_model_name_takes_priority_over_name_or_path(self, tmp_path: Path):
"""model_name should be preferred over _name_or_path."""
config_cfg = {
"model_name": "Qwen/Qwen3.5-9B",
"_name_or_path": "some/other-model",
}
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
result = _resolve_base_model(str(tmp_path))
assert result == "Qwen/Qwen3.5-9B"
def test_config_json_skips_self_referencing(self, tmp_path: Path):
"""config.json should be ignored if model_name == the checkpoint path."""
config_cfg = {"model_name": str(tmp_path)}
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
result = _resolve_base_model(str(tmp_path))
# Falls through; does not return the self-referencing path.
assert result == str(tmp_path)
def test_no_config_files(self, tmp_path: Path):
"""Returns original name when no config files are present."""
result = _resolve_base_model(str(tmp_path))
assert result == str(tmp_path)
def test_plain_hf_id_passthrough(self):
"""Plain HuggingFace model IDs pass through unchanged."""
result = _resolve_base_model("meta-llama/Llama-3-8B")
assert result == "meta-llama/Llama-3-8B"
class TestRemoteLoraBase:
"""_remote_lora_base reads a remote adapter's base from its Hub adapter_config.json."""
@pytest.fixture(autouse = True)
def _selected_cache_follows_env(self, monkeypatch):
# The cache helpers now read the selected cache (get_hf_cache_paths),
# which snapshots env at import; make it follow the HF_HUB_CACHE these
# tests set so they keep driving the lookup via env.
monkeypatch.setattr(
"utils.transformers_version.get_hf_cache_paths",
lambda: _types.SimpleNamespace(
hub_cache = Path(
os.environ.get("HF_HUB_CACHE")
or os.environ.get("HUGGINGFACE_HUB_CACHE")
or os.path.join(
os.environ.get("HF_HOME") or os.path.expanduser("~/.cache/huggingface"),
"hub",
)
)
),
)
@staticmethod
def _resp(cfg: dict):
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps(cfg).encode()
return _Resp()
def test_fetches_base_from_remote_adapter_config(self, monkeypatch):
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
cfg = {"base_model_name_or_path": "nvidia/NVIDIA-Nemotron-3-Nano-4B"}
with patch("urllib.request.urlopen", return_value = self._resp(cfg)):
assert (
_remote_lora_base("someuser/my-nemotron-lora") == "nvidia/NVIDIA-Nemotron-3-Nano-4B"
)
def test_local_or_noncanonical_returns_none(self):
assert _remote_lora_base("/local/dir/adapter") is None
assert _remote_lora_base("plainname") is None
def test_respects_hf_endpoint(self, monkeypatch):
# Enterprise mirror: the fetch must target HF_ENDPOINT, not hardcoded huggingface.co.
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
monkeypatch.setenv("HF_ENDPOINT", "https://hf.mirror.internal")
seen = {}
def fake_urlopen(req, timeout = 10):
seen["url"] = req.full_url
return self._resp({"base_model_name_or_path": "org/base"})
with patch("urllib.request.urlopen", side_effect = fake_urlopen):
assert _remote_lora_base("user/adapter") == "org/base"
assert seen["url"].startswith("https://hf.mirror.internal/user/adapter/resolve/main/")
@staticmethod
def _seed_adapter_cache(
hub: Path,
repo_id: str,
base: str,
commit: str = "deadbeef",
):
repo = hub / ("models--" + repo_id.replace("/", "--"))
snap = repo / "snapshots" / commit
snap.mkdir(parents = True)
(snap / "adapter_config.json").write_text(json.dumps({"base_model_name_or_path": base}))
(repo / "refs").mkdir(parents = True)
(repo / "refs" / "main").write_text(commit)
def test_offline_reads_base_from_hf_cache(self, tmp_path: Path, monkeypatch):
self._seed_adapter_cache(tmp_path, "user/cached-lora", "nvidia/Nemotron-H-8B")
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.setenv("HF_HUB_OFFLINE", "1")
with patch("urllib.request.urlopen") as mock_url:
assert _remote_lora_base("user/cached-lora") == "nvidia/Nemotron-H-8B"
mock_url.assert_not_called() # offline: cache only, no network
def test_fetch_failure_falls_back_to_cache(self, tmp_path: Path, monkeypatch):
self._seed_adapter_cache(tmp_path, "user/cached-lora", "nvidia/Nemotron-H-8B")
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
with patch("urllib.request.urlopen", side_effect = OSError("boom")):
assert _remote_lora_base("user/cached-lora") == "nvidia/Nemotron-H-8B"
def test_offline_uncached_makes_no_request(self, tmp_path: Path, monkeypatch):
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.setenv("HF_HUB_OFFLINE", "1")
with patch("urllib.request.urlopen") as mock_url:
assert _remote_lora_base("org/adapter") is None
mock_url.assert_not_called()
def test_non_adapter_repo_returns_none(self, tmp_path: Path, monkeypatch):
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
with patch("urllib.request.urlopen", side_effect = OSError("boom")):
assert _remote_lora_base("org/not-an-adapter") is None
def test_existing_relative_path_not_treated_as_repo(self, monkeypatch):
# An existing one-slash relative path (e.g. outputs/run1) is a local checkpoint, not
# a Hub repo: no request, no risk of matching an unrelated remote/cached adapter.
import utils.paths as paths
monkeypatch.setattr(paths, "is_local_path", lambda p: True)
with patch("urllib.request.urlopen") as mock_url:
assert _remote_lora_base("outputs/run1") is None
mock_url.assert_not_called()
def test_404_returns_none_not_stale_cache(self, tmp_path: Path, monkeypatch):
import urllib.error
# The repo is now a full model (adapter_config.json 404s) but a stale LoRA snapshot is
# cached: a definitive 404 must return None, not the stale base.
self._seed_adapter_cache(tmp_path, "user/was-a-lora", "old/base")
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
err = urllib.error.HTTPError("url", 404, "Not Found", {}, None)
with patch("urllib.request.urlopen", side_effect = err):
assert _remote_lora_base("user/was-a-lora") is None
def test_transient_http_error_falls_back_to_cache(self, tmp_path: Path, monkeypatch):
import urllib.error
self._seed_adapter_cache(tmp_path, "user/cached-lora", "nvidia/Nemotron-H-8B")
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
err = urllib.error.HTTPError("url", 503, "Service Unavailable", {}, None)
with patch("urllib.request.urlopen", side_effect = err):
assert _remote_lora_base("user/cached-lora") == "nvidia/Nemotron-H-8B"
# ---------------------------------------------------------------------------
# _check_tokenizer_config_needs_v5 — local file check
# ---------------------------------------------------------------------------
class TestCheckTokenizerConfigNeedsV5:
"""Tests for local tokenizer_config.json fallback."""
def setup_method(self):
_tokenizer_class_cache.clear()
def test_local_tokenizer_config_v5(self, tmp_path: Path):
"""Local tokenizer_config.json with v5 tokenizer should return True."""
tc = {"tokenizer_class": "TokenizersBackend"}
(tmp_path / "tokenizer_config.json").write_text(json.dumps(tc))
result = _check_tokenizer_config_needs_v5(str(tmp_path))
assert result is True
def test_local_tokenizer_config_v4(self, tmp_path: Path):
"""Local tokenizer_config.json with standard tokenizer should return False."""
tc = {"tokenizer_class": "LlamaTokenizerFast"}
(tmp_path / "tokenizer_config.json").write_text(json.dumps(tc))
result = _check_tokenizer_config_needs_v5(str(tmp_path))
assert result is False
def test_local_file_skips_network(self, tmp_path: Path):
"""When local file exists, no network request should be made."""
tc = {"tokenizer_class": "LlamaTokenizerFast"}
(tmp_path / "tokenizer_config.json").write_text(json.dumps(tc))
with patch("urllib.request.urlopen") as mock_urlopen:
result = _check_tokenizer_config_needs_v5(str(tmp_path))
mock_urlopen.assert_not_called()
assert result is False
def test_result_is_cached(self, tmp_path: Path):
"""Subsequent calls should use the cache."""
tc = {"tokenizer_class": "TokenizersBackend"}
(tmp_path / "tokenizer_config.json").write_text(json.dumps(tc))
key = (str(tmp_path), None)
_check_tokenizer_config_needs_v5(str(tmp_path))
assert key in _tokenizer_class_cache
assert _tokenizer_class_cache[key] is True
def test_token_cache_isolation_and_auth_fetch(self, monkeypatch):
# A gated repo: the unauthenticated miss (cached under (model, None)) must not block a
# later authed fetch (separate key), and the token rides in the Authorization header.
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_env_offline", lambda: False)
seen_auth = []
class _Resp:
def __init__(self, body):
self._b = body
def read(self):
return self._b.encode()
def __enter__(self):
return self
def __exit__(self, *a):
return False
def fake_urlopen(req, timeout = 10):
auth = req.get_header("Authorization")
seen_auth.append(auth)
if auth:
return _Resp(json.dumps({"tokenizer_class": "TokenizersBackend"}))
raise OSError("HTTP 401")
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
assert _check_tokenizer_config_needs_v5("org/gated") is False # unauth miss
assert _check_tokenizer_config_needs_v5("org/gated", "tok") is True # authed hit
assert seen_auth == [None, "Bearer tok"]
assert _tokenizer_class_cache[("org/gated", None)] is False # miss not poisoning
def test_remote_fetch_respects_hf_endpoint(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_env_offline", lambda: False)
monkeypatch.setenv("HF_ENDPOINT", "https://hf.mirror.internal/")
seen = {}
class _Resp:
def read(self):
return json.dumps({"tokenizer_class": "TokenizersBackend"}).encode()
def __enter__(self):
return self
def __exit__(self, *a):
return False
def fake_urlopen(req, timeout = 10):
seen["url"] = req.full_url
return _Resp()
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
assert _check_tokenizer_config_needs_v5("org/model") is True
assert seen["url"] == (
"https://hf.mirror.internal/org/model/resolve/main/tokenizer_config.json"
)
# ---------------------------------------------------------------------------
# needs_transformers_5 — integration-level
# ---------------------------------------------------------------------------
class TestNeedsTransformers5:
"""Integration tests for the top-level needs_transformers_5() function."""
def setup_method(self):
_tokenizer_class_cache.clear()
def test_qwen35_substring(self):
assert needs_transformers_5("Qwen/Qwen3.5-9B") is True
def test_qwen3_30b_a3b_substring(self):
assert needs_transformers_5("Qwen/Qwen3-30B-A3B-Instruct-2507") is True
def test_ministral_substring(self):
assert needs_transformers_5("mistralai/Ministral-3-8B-Instruct-2512") is True
def test_llama_does_not_need_v5(self):
"""Standard models should not trigger v5."""
# Patch network call to avoid a real fetch.
with patch(
"utils.transformers_version._check_tokenizer_config_needs_v5",
return_value = False,
):
assert needs_transformers_5("meta-llama/Llama-3-8B") is False
def test_local_checkpoint_resolved_via_config(self, tmp_path: Path):
"""Local checkpoint with config.json pointing to Qwen3.5 needs v5."""
config_cfg = {"model_name": "Qwen/Qwen3.5-9B"}
(tmp_path / "config.json").write_text(json.dumps(config_cfg))
# needs_transformers_5 only does substring matching, so test the
# full resolution chain via _resolve_base_model here.
resolved = _resolve_base_model(str(tmp_path))
assert needs_transformers_5(resolved) is True
# ---------------------------------------------------------------------------
# _check_config_needs_550 — config.json architecture/model_type check
# ---------------------------------------------------------------------------
class TestCheckConfigNeeds550:
"""Tests for _check_config_needs_550() local config.json checks."""
def setup_method(self):
_config_json_cache.clear()
_config_needs_550_cache.clear()
def test_gemma4_architecture(self, tmp_path: Path):
"""config.json with Gemma4ForConditionalGeneration should return True."""
cfg = {
"architectures": ["Gemma4ForConditionalGeneration"],
"model_type": "gemma4",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_550(str(tmp_path)) is True
def test_gemma4_model_type_only(self, tmp_path: Path):
"""config.json with model_type=gemma4 (no architectures) should return True."""
cfg = {"model_type": "gemma4"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_550(str(tmp_path)) is True
@pytest.mark.parametrize(
"cfg",
(
{"architectures": ["KimiK3ForConditionalGeneration"]},
{"model_type": "kimi_k3"},
{"architectures": ["LocateAnythingForConditionalGeneration"]},
{"model_type": "locateanything"},
{"architectures": ["DiffusionGemmaForBlockDiffusion"]},
{"model_type": "diffusion_gemma"},
),
ids = (
"kimi-k3-architecture",
"kimi-k3-model-type",
"locate-anything-architecture",
"locate-anything-model-type",
"diffusion-gemma-architecture",
"diffusion-gemma-model-type",
),
)
def test_mlx_vlm_v5_processor_config(self, tmp_path: Path, cfg: dict):
"""mlx-vlm processors that require Transformers 5 should select 5.5."""
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_550(str(tmp_path)) is True
def test_llama_architecture(self, tmp_path: Path):
"""config.json with LlamaForCausalLM should return False."""
cfg = {"architectures": ["LlamaForCausalLM"], "model_type": "llama"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_550(str(tmp_path)) is False
def test_no_config_json(self, tmp_path: Path):
"""Missing config.json should return False (fail-open)."""
# Patch network call to avoid a real fetch.
with patch("urllib.request.urlopen") as mock_urlopen:
mock_urlopen.side_effect = Exception("no network")
assert _check_config_needs_550(str(tmp_path)) is False
def test_result_is_cached(self, tmp_path: Path):
"""Subsequent calls should use the cache."""
cfg = {"architectures": ["Gemma4ForConditionalGeneration"]}
(tmp_path / "config.json").write_text(json.dumps(cfg))
key = (str(tmp_path), None)
_check_config_needs_550(str(tmp_path))
assert key in _config_needs_550_cache
assert _config_needs_550_cache[key] is True
def test_local_file_skips_network(self, tmp_path: Path):
"""When local config.json exists, no network request should be made."""
cfg = {"architectures": ["LlamaForCausalLM"]}
(tmp_path / "config.json").write_text(json.dumps(cfg))
with patch("urllib.request.urlopen") as mock_urlopen:
_check_config_needs_550(str(tmp_path))
mock_urlopen.assert_not_called()
# ---------------------------------------------------------------------------
# _check_config_needs_510 — config.json architecture/model_type check
# ---------------------------------------------------------------------------
class TestCheckConfigNeeds510:
"""Tests for _check_config_needs_510() local config.json checks."""
def setup_method(self):
_config_json_cache.clear()
_config_needs_510_cache.clear()
def test_gemma4_unified_architecture(self, tmp_path: Path):
"""config.json with Gemma4UnifiedForConditionalGeneration should return True."""
cfg = {
"architectures": ["Gemma4UnifiedForConditionalGeneration"],
"model_type": "gemma4_unified",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is True
def test_gemma4_unified_model_type_only(self, tmp_path: Path):
"""config.json with model_type=gemma4_unified should return True."""
cfg = {"model_type": "gemma4_unified"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is True
def test_gemma4_unified_assistant_architecture(self, tmp_path: Path):
"""Assistant Gemma 4 Unified configs should return True."""
cfg = {
"architectures": ["Gemma4UnifiedAssistantForCausalLM"],
"model_type": "gemma4_unified_assistant",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is True
def test_gemma4_unified_assistant_model_type_only(self, tmp_path: Path):
"""Assistant Gemma 4 Unified model_type should return True."""
cfg = {"model_type": "gemma4_unified_assistant"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is True
def test_gemma4_assistant_architecture(self, tmp_path: Path):
"""Assistant Gemma 4 configs should return True."""
cfg = {
"architectures": ["Gemma4AssistantForCausalLM"],
"model_type": "gemma4_assistant",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is True
def test_gemma4_assistant_model_type_only(self, tmp_path: Path):
"""Assistant Gemma 4 model_type should return True."""
cfg = {"model_type": "gemma4_assistant"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is True
def test_gemma4_non_unified_returns_false(self, tmp_path: Path):
"""Older Gemma 4 config should stay on the 550 tier."""
cfg = {
"architectures": ["Gemma4ForConditionalGeneration"],
"model_type": "gemma4",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert _check_config_needs_510(str(tmp_path)) is False
def test_no_config_json(self, tmp_path: Path):
"""Missing config.json should return False (fail-open)."""
# Patch network call to avoid real fetch
with patch("urllib.request.urlopen") as mock_urlopen:
mock_urlopen.side_effect = Exception("no network")
assert _check_config_needs_510(str(tmp_path)) is False
def test_result_is_cached(self, tmp_path: Path):
"""Subsequent calls should use the cache."""
cfg = {"architectures": ["Gemma4UnifiedForConditionalGeneration"]}
(tmp_path / "config.json").write_text(json.dumps(cfg))
key = (str(tmp_path), None)
_check_config_needs_510(str(tmp_path))
assert key in _config_needs_510_cache
assert _config_needs_510_cache[key] is True
def test_local_file_skips_network(self, tmp_path: Path):
"""When local config.json exists, no network request should be made."""
cfg = {"architectures": ["LlamaForCausalLM"]}
(tmp_path / "config.json").write_text(json.dumps(cfg))
with patch("urllib.request.urlopen") as mock_urlopen:
_check_config_needs_510(str(tmp_path))
mock_urlopen.assert_not_called()
# ---------------------------------------------------------------------------
# NemotronH dense (MLP) models need the 5.10 tier
# ---------------------------------------------------------------------------
class TestNemotronHNeedsMlpSupport:
"""Dense NemotronH configs (MLP layers) require transformers >= 5.10."""
def test_hybrid_override_pattern_with_dash(self):
cfg = {
"model_type": "nemotron_h",
"hybrid_override_pattern": "M-M-M*-M-",
}
assert _nemotron_h_needs_mlp_support(cfg) is True
def test_layers_block_type_with_mlp(self):
cfg = {
"model_type": "nemotron_h",
"layers_block_type": ["mamba", "mlp", "attention", "mamba"],
}
assert _nemotron_h_needs_mlp_support(cfg) is True
def test_nemotron_h_moe_only_returns_false(self):
"""A pure MoE NemotronH (no MLP) does not need the 5.10 tier."""
cfg = {
"model_type": "nemotron_h",
"hybrid_override_pattern": "MEME*MEM",
}
assert _nemotron_h_needs_mlp_support(cfg) is False
def test_non_nemotron_with_dash_returns_false(self):
"""The dash heuristic only applies to nemotron_h configs."""
cfg = {"model_type": "llama", "hybrid_override_pattern": "M-M-"}
assert _nemotron_h_needs_mlp_support(cfg) is False
def test_config_needs_510_includes_dense_nemotron_h(self):
cfg = {
"model_type": "nemotron_h",
"hybrid_override_pattern": "M-M-M*-",
}
assert _config_needs_510(cfg) is True
def test_nested_llm_config_with_dash(self):
# VL wrapper (e.g. NemotronH_Nano_VL_V2): dense LM is under llm_config.
cfg = {
"model_type": "NemotronH_Nano_VL_V2",
"llm_config": {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"},
}
assert _nemotron_h_needs_mlp_support(cfg) is True
assert _config_needs_510(cfg) is True
def test_nested_text_config_with_mlp(self):
cfg = {
"model_type": "wrapper",
"text_config": {"model_type": "nemotron_h", "layers_block_type": ["mamba", "mlp"]},
}
assert _nemotron_h_needs_mlp_support(cfg) is True
def test_nested_non_nemotron_returns_false(self):
cfg = {"model_type": "wrapper", "llm_config": {"model_type": "llama"}}
assert _nemotron_h_needs_mlp_support(cfg) is False
def test_non_dict_and_missing_nested_do_not_raise(self):
assert _nemotron_h_needs_mlp_support(None) is False
assert _nemotron_h_needs_mlp_support({"model_type": "wrapper", "llm_config": None}) is False
def _hf_response(cfg: dict):
"""A urlopen() context-manager stand-in returning *cfg* as JSON bytes."""
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps(cfg).encode()
return _Resp()
class TestConfigJsonHfCacheFallback:
"""HF hub cache is consulted only offline or after a failed fetch (never stale online)."""
@pytest.fixture(autouse = True)
def _selected_cache_follows_env(self, monkeypatch):
# As above: route the selected-cache lookup through the HF_HUB_CACHE env
# these tests set, since get_hf_cache_paths snapshots env at import.
monkeypatch.setattr(
"utils.transformers_version.get_hf_cache_paths",
lambda: _types.SimpleNamespace(
hub_cache = Path(
os.environ.get("HF_HUB_CACHE")
or os.environ.get("HUGGINGFACE_HUB_CACHE")
or os.path.join(
os.environ.get("HF_HOME") or os.path.expanduser("~/.cache/huggingface"),
"hub",
)
)
),
)
def setup_method(self):
_config_json_cache.clear()
@staticmethod
def _seed_cache(
hub: Path,
repo_id: str,
cfg: dict,
commit: str = "deadbeef",
):
repo = hub / ("models--" + repo_id.replace("/", "--"))
snap = repo / "snapshots" / commit
snap.mkdir(parents = True)
(snap / "config.json").write_text(json.dumps(cfg))
(repo / "refs").mkdir(parents = True)
(repo / "refs" / "main").write_text(commit)
def test_offline_reads_from_cache(self, tmp_path: Path, monkeypatch):
cfg = {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"}
self._seed_cache(tmp_path, "unsloth/NVIDIA-Nemotron-3-Nano-4B", cfg)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.setenv("HF_HUB_OFFLINE", "1")
with patch("urllib.request.urlopen") as mock_url:
assert _load_config_json("unsloth/NVIDIA-Nemotron-3-Nano-4B") == cfg
mock_url.assert_not_called()
def test_online_prefers_network_over_cache(self, tmp_path: Path, monkeypatch):
stale = {"model_type": "nemotron_h", "hybrid_override_pattern": "MMMM"}
fresh = {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"}
self._seed_cache(tmp_path, "org/model", stale)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
monkeypatch.delenv("TRANSFORMERS_OFFLINE", raising = False)
with patch("urllib.request.urlopen", return_value = _hf_response(fresh)):
assert _load_config_json("org/model") == fresh # network wins, not stale cache
def test_remote_fetch_respects_hf_endpoint(self, monkeypatch):
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
monkeypatch.delenv("TRANSFORMERS_OFFLINE", raising = False)
monkeypatch.setenv("HF_ENDPOINT", "https://hf.mirror.internal/")
seen = {}
def fake_urlopen(req, timeout = 10):
seen["url"] = req.full_url
return _hf_response({"model_type": "llama"})
with patch("urllib.request.urlopen", side_effect = fake_urlopen):
assert _load_config_json("org/model") == {"model_type": "llama"}
assert seen["url"] == "https://hf.mirror.internal/org/model/resolve/main/config.json"
def test_network_failure_falls_back_to_cache(self, tmp_path: Path, monkeypatch):
cfg = {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"}
self._seed_cache(tmp_path, "org/model", cfg)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
with patch("urllib.request.urlopen", side_effect = OSError("boom")):
assert _load_config_json("org/model") == cfg
def test_offline_uncached_returns_none(self, tmp_path: Path, monkeypatch):
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.setenv("HF_HUB_OFFLINE", "1")
with patch("urllib.request.urlopen") as mock_url:
assert _load_config_json("private/unknown") is None
mock_url.assert_not_called()
def test_helper_ignores_local_paths(self, tmp_path: Path):
# A filesystem path is not a repo id; never treat it as one.
assert _config_json_from_hf_cache(str(tmp_path)) is None
assert _config_json_from_hf_cache("plainname") is None
def test_no_refs_main_picks_newest_snapshot(self, tmp_path: Path, monkeypatch):
# No refs/main (commit-pinned downloads): lexicographic order would pick the older
# SHA; selection must follow mtime so the newest snapshot wins.
repo = tmp_path / "models--org--model"
old = repo / "snapshots" / "0000old"
new = repo / "snapshots" / "ffffnew"
old.mkdir(parents = True)
new.mkdir(parents = True)
(old / "config.json").write_text(json.dumps({"model_type": "stale"}))
(new / "config.json").write_text(json.dumps({"model_type": "fresh"}))
os.utime(old / "config.json", (1000, 1000))
os.utime(new / "config.json", (2000, 2000))
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
assert _config_json_from_hf_cache("org/model") == {"model_type": "fresh"}
def test_transient_failure_does_not_cache_fallback(self, tmp_path: Path, monkeypatch):
stale = {"model_type": "nemotron_h", "hybrid_override_pattern": "MMMM"}
fresh = {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"}
self._seed_cache(tmp_path, "org/model", stale)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
# Network fails -> serve the cached snapshot, but it must not be memoized.
with patch("urllib.request.urlopen", side_effect = OSError("boom")):
assert _load_config_json("org/model") == stale
# Connectivity returns: the next call must hit the network for the fresh config.
with patch("urllib.request.urlopen", return_value = _hf_response(fresh)):
assert _load_config_json("org/model") == fresh
def test_auth_failure_does_not_serve_cache(self, tmp_path: Path, monkeypatch):
import urllib.error
# config.json cached from an earlier authorized session; an unauthenticated 4xx
# must not be handed that private metadata.
cfg = {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"}
self._seed_cache(tmp_path, "private/model", cfg)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
for code in (401, 403, 404):
_config_json_cache.clear()
err = urllib.error.HTTPError("url", code, "denied", {}, None)
with patch("urllib.request.urlopen", side_effect = err):
assert _load_config_json("private/model") is None
def test_server_error_still_falls_back_to_cache(self, tmp_path: Path, monkeypatch):
import urllib.error
cfg = {"model_type": "nemotron_h", "hybrid_override_pattern": "M-M*-"}
self._seed_cache(tmp_path, "org/model", cfg)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
# A 5xx is transient, not an access decision: keep serving the cache.
err = urllib.error.HTTPError("url", 503, "busy", {}, None)
with patch("urllib.request.urlopen", side_effect = err):
assert _load_config_json("org/model") == cfg
class TestTierCheckTransientRetry:
"""tier-needs checks must not memoize a transient fetch fallback."""
def setup_method(self):
_config_json_cache.clear()
_config_needs_510_cache.clear()
_config_needs_550_cache.clear()
@staticmethod
def _seed_cache(
hub: Path,
repo_id: str,
cfg: dict,
commit: str = "deadbeef",
):
repo = hub / ("models--" + repo_id.replace("/", "--"))
snap = repo / "snapshots" / commit
snap.mkdir(parents = True)
(snap / "config.json").write_text(json.dumps(cfg))
(repo / "refs").mkdir(parents = True)
(repo / "refs" / "main").write_text(commit)
def test_transient_fallback_not_memoized_then_retries(self, tmp_path: Path, monkeypatch):
stale = {"model_type": "llama"} # does not need 510
fresh = {"architectures": ["Gemma4UnifiedForConditionalGeneration"]} # needs 510
self._seed_cache(tmp_path, "org/model", stale)
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
# Network blip -> serve the cache, but do NOT pin the tier result.
with patch("urllib.request.urlopen", side_effect = OSError("boom")):
assert _check_config_needs_510("org/model") is False
assert ("org/model", None) not in _config_needs_510_cache
# Connectivity returns: the next call re-fetches and sees the higher tier.
with patch("urllib.request.urlopen", return_value = _hf_response(fresh)):
assert _check_config_needs_510("org/model") is True
assert _config_needs_510_cache[("org/model", None)] is True # definitive read memoized
def test_definitive_network_read_is_memoized(self, tmp_path: Path, monkeypatch):
fresh = {"architectures": ["Gemma4ForConditionalGeneration"]} # needs 550
monkeypatch.setenv("HF_HUB_CACHE", str(tmp_path))
monkeypatch.delenv("HF_HUB_OFFLINE", raising = False)
with patch("urllib.request.urlopen", return_value = _hf_response(fresh)) as mock_url:
assert _check_config_needs_550("org/model") is True
assert _check_config_needs_550("org/model") is True
assert mock_url.call_count == 1 # second call served from the tier cache
class TestHigherTier:
def test_picks_stronger_tier(self):
assert _higher_tier("default", "510") == "510"
assert _higher_tier("530", "550") == "550"
assert _higher_tier("510", "default") == "510"
assert _higher_tier("default", "default") == "default"
# ---------------------------------------------------------------------------
# get_transformers_tier — tier detection
# ---------------------------------------------------------------------------
class TestGetTransformersTier:
"""Tests for get_transformers_tier() tiered version detection."""
def setup_method(self):
_tokenizer_class_cache.clear()
_config_json_cache.clear()
_config_needs_510_cache.clear()
_config_needs_550_cache.clear()
def test_gemma4_substring_returns_550(self):
assert get_transformers_tier("google/gemma-4-E2B-it") == "550"
def test_gemma4_12b_substring_returns_510(self):
assert get_transformers_tier("unsloth/gemma-4-12b-it") == "510"
def test_gemma4_assistant_substring_returns_510(self):
assert get_transformers_tier("google/gemma-4-E2B-it-assistant") == "510"
def test_gemma4_alt_substring_returns_550(self):
assert get_transformers_tier("unsloth/gemma4-E4B-it") == "550"
@pytest.mark.parametrize(
"model_id",
(
"moonshotai/Kimi-K3",
"mlx-community/KimiK3-4bit",
"nvidia/LocateAnything-3B",
"mlx-community/locate-anything-3b-4bit",
"mlx-community/diffusiongemma-26B-A4B-it-4bit",
"mlx-community/diffusion-gemma-26B-A4B-it-4bit",
),
)
def test_mlx_vlm_v5_processor_name_returns_550(self, model_id: str):
assert get_transformers_tier(model_id) == "550"
def test_gemma4_config_json_returns_550(self, tmp_path: Path):
"""Local checkpoint with Gemma4 architecture → 550."""
cfg = {
"architectures": ["Gemma4ForConditionalGeneration"],
"model_type": "gemma4",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert get_transformers_tier(str(tmp_path)) == "550"
def test_gemma4_unified_config_json_returns_510(self, tmp_path: Path):
"""Local checkpoint with Gemma4 Unified architecture → 510."""
cfg = {
"architectures": ["Gemma4UnifiedForConditionalGeneration"],
"model_type": "gemma4_unified",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert get_transformers_tier(str(tmp_path)) == "510"
def test_gemma4_assistant_config_json_returns_510(self, tmp_path: Path):
"""Local checkpoint with Gemma4 Assistant architecture → 510."""
cfg = {
"architectures": ["Gemma4AssistantForCausalLM"],
"model_type": "gemma4_assistant",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
assert get_transformers_tier(str(tmp_path)) == "510"
def test_dense_nemotron_h_config_json_returns_510(self, tmp_path: Path):
"""Local dense NemotronH checkpoint → 510 (MLP layers need >= 5.10)."""
cfg = {
"model_type": "nemotron_h",
"hybrid_override_pattern": "M-M-M*-M-",
}
(tmp_path / "config.json").write_text(json.dumps(cfg))
# A v5 tokenizer would otherwise route this to 530; 510 must win.
(tmp_path / "tokenizer_config.json").write_text(
json.dumps({"tokenizer_class": "TokenizersBackend"})
)
with patch("urllib.request.urlopen") as mock_urlopen:
assert get_transformers_tier(str(tmp_path)) == "510"
mock_urlopen.assert_not_called()
def test_dense_nemotron_h_remote_config_returns_510(self):
"""Remote dense NemotronH (HF id) → 510 via config.json fetch, not 530."""
class _Response:
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def read(self):
return json.dumps(
{
"model_type": "nemotron_h",
"hybrid_override_pattern": "M-M-M*-M-",
}
).encode()
with patch("urllib.request.urlopen", return_value = _Response()):
assert get_transformers_tier("unsloth/NVIDIA-Nemotron-3-Nano-4B") == "510"
def test_local_config_json_short_circuits_path_substrings(self, tmp_path: Path):
"""Local config.json should prevent false matches from parent directory names."""
model_dir = tmp_path / "gemma-4-12b-experiment" / "llama-checkpoint"
model_dir.mkdir(parents = True)
(model_dir / "config.json").write_text(
json.dumps(
{
"architectures": ["LlamaForCausalLM"],
"model_type": "llama",
}
)
)
(model_dir / "tokenizer_config.json").write_text(
json.dumps({"tokenizer_class": "LlamaTokenizerFast"})
)
with patch("urllib.request.urlopen") as mock_urlopen:
assert get_transformers_tier(str(model_dir)) == "default"
mock_urlopen.assert_not_called()
def test_remote_config_json_is_fetched_once_for_config_tiers(self):
"""510 and 550 slow-path checks should share one config.json fetch."""
class _Response:
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def read(self):
return json.dumps(
{
"architectures": ["Gemma4ForConditionalGeneration"],
"model_type": "gemma4",
}
).encode()
with patch("urllib.request.urlopen", return_value = _Response()) as mock_urlopen:
assert get_transformers_tier("org/no-fast-substring-model") == "550"
assert mock_urlopen.call_count == 1
def test_qwen35_returns_530(self):
with (
patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
),
patch(
"utils.transformers_version._check_config_needs_510",
return_value = False,
),
):
assert get_transformers_tier("Qwen/Qwen3.5-9B") == "530"
def test_ministral_returns_530(self):
with (
patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
),
patch(
"utils.transformers_version._check_config_needs_510",
return_value = False,
),
):
assert get_transformers_tier("mistralai/Ministral-3-8B-Instruct-2512") == "530"
def test_llama_returns_default(self):
with (
patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
),
patch(
"utils.transformers_version._check_config_needs_510",
return_value = False,
),
patch(
"utils.transformers_version._check_tokenizer_config_needs_v5",
return_value = False,
),
):
assert get_transformers_tier("meta-llama/Llama-3-8B") == "default"
def test_550_checked_before_530(self):
"""5.5.0 is checked before 5.3.0 - a model matching both gets 550."""
assert get_transformers_tier("gemma-4-model") == "550"
# ---- issue #6103: the tier decision must be traceable in the logs ----
def test_tier_550_selection_is_logged(self, caplog):
caplog.set_level(logging.INFO)
assert get_transformers_tier("google/gemma-4-E2B-it") == "550"
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "550" in text, f"tier selection not logged: {text!r}"
assert "gemma-4-e2b-it" in text, f"tier log omits the model: {text!r}"
def test_tier_530_selection_is_logged(self, caplog):
caplog.set_level(logging.INFO)
with patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
):
assert get_transformers_tier("Qwen/Qwen3.5-9B") == "530"
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "530" in text, f"tier selection not logged: {text!r}"
assert "qwen3.5-9b" in text, f"tier log omits the model: {text!r}"
def test_tier_default_selection_is_logged(self, caplog):
caplog.set_level(logging.INFO)
with (
patch(
"utils.transformers_version._check_config_needs_510",
return_value = False,
),
patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
),
patch(
"utils.transformers_version._check_tokenizer_config_needs_v5",
return_value = False,
),
):
assert get_transformers_tier("meta-llama/Llama-3-8B") == "default"
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "default" in text, f"tier selection not logged: {text!r}"
def test_local_config_json_selection_is_logged(self, tmp_path: Path, caplog):
cfg = {"architectures": ["Gemma4ForConditionalGeneration"], "model_type": "gemma4"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
caplog.set_level(logging.INFO)
assert get_transformers_tier(str(tmp_path)) == "550"
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "550" in text and "local config.json" in text, f"local tier not logged: {text!r}"
def test_needs_transformers_5_compat(self):
"""needs_transformers_5 should return True for 510, 530, and 550 models."""
assert needs_transformers_5("unsloth/gemma-4-12b-it") is True
assert needs_transformers_5("google/gemma-4-E2B-it") is True
with (
patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
),
patch(
"utils.transformers_version._check_config_needs_510",
return_value = False,
),
):
assert needs_transformers_5("Qwen/Qwen3.5-9B") is True
with (
patch(
"utils.transformers_version._check_config_needs_550",
return_value = False,
),
patch(
"utils.transformers_version._check_config_needs_510",
return_value = False,
),
patch(
"utils.transformers_version._check_tokenizer_config_needs_v5",
return_value = False,
),
):
assert needs_transformers_5("meta-llama/Llama-3-8B") is False
def _proc(returncode, stderr = ""):
from types import SimpleNamespace
return SimpleNamespace(returncode = returncode, stdout = "", stderr = stderr)
class TestProbeTier:
"""_probe_tier resolves the tier by parsing config in each sidecar and escalating."""
def setup_method(self):
_probe_tier_cache.clear()
def _patch_common(self, monkeypatch):
for fn in (
"_ensure_venv_t5_530_exists",
"_ensure_venv_t5_550_exists",
"_ensure_venv_t5_510_exists",
):
monkeypatch.setattr(f"utils.transformers_version.{fn}", lambda: True)
monkeypatch.delenv("UNSLOTH_DISABLE_TIER_PROBE", raising = False)
def _venv_dirs(self):
import utils.transformers_version as tv
return [tv._VENV_T5_530_DIR, tv._VENV_T5_550_DIR, tv._VENV_T5_510_DIR]
def test_escalates_to_first_parsing_tier(self, monkeypatch):
self._patch_common(monkeypatch)
seen = []
results = iter([_proc(1, "KeyError: '-'"), _proc(1, "KeyError: '-'"), _proc(0)])
def fake_run(cmd, **k):
seen.append(cmd[3]) # target_dir
return next(results)
monkeypatch.setattr("utils.transformers_version.subprocess.run", fake_run)
assert _probe_tier("org/dense-nemotron", None, "x") == "510"
assert seen == self._venv_dirs() # escalated 530 -> 550 -> 510
def test_first_success_stops_escalation(self, monkeypatch):
self._patch_common(monkeypatch)
calls = []
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: calls.append(cmd[3]) or _proc(0),
)
assert _probe_tier("org/m", None, "x") == "530"
assert len(calls) == 1
def test_middle_tier_parses(self, monkeypatch):
self._patch_common(monkeypatch)
results = iter([_proc(1, "ValueError: bad"), _proc(0)])
monkeypatch.setattr(
"utils.transformers_version.subprocess.run", lambda cmd, **k: next(results)
)
assert _probe_tier("org/m", None, "x") == "550"
def test_nothing_parses_stays_530_and_caches(self, monkeypatch):
# All tiers probed, none parse -> a remote-code model that loads via its own code;
# keep 530 (never jump to 510). Conclusive, so cached by model_name.
self._patch_common(monkeypatch)
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: _proc(1, "KeyError: '-'"),
)
assert _probe_tier("org/m", None, "x") == "530"
assert _probe_tier_cache["org/m"] == "530"
def test_partial_sidecars_no_parse_is_530_uncached(self, monkeypatch):
# 510 sidecar missing and 530/550 fail to parse -> environment is incomplete, so we
# cannot conclude; return 530 uncached so it is retried once 510 is available.
monkeypatch.delenv("UNSLOTH_DISABLE_TIER_PROBE", raising = False)
for fn in ("_ensure_venv_t5_530_exists", "_ensure_venv_t5_550_exists"):
monkeypatch.setattr(f"utils.transformers_version.{fn}", lambda: True)
monkeypatch.setattr("utils.transformers_version._ensure_venv_t5_510_exists", lambda: False)
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: _proc(1, "KeyError: '-'"),
)
assert _probe_tier("org/m", None, "x") == "530"
assert "org/m" not in _probe_tier_cache
def test_success_not_cached_when_lower_tier_skipped(self, monkeypatch):
# 530 sidecar unavailable but 550 parses: return 550 (best effort now) but do NOT
# cache it, since once 530 is installed it may be the lowest valid tier.
monkeypatch.delenv("UNSLOTH_DISABLE_TIER_PROBE", raising = False)
monkeypatch.setattr("utils.transformers_version._ensure_venv_t5_530_exists", lambda: False)
for fn in ("_ensure_venv_t5_550_exists", "_ensure_venv_t5_510_exists"):
monkeypatch.setattr(f"utils.transformers_version.{fn}", lambda: True)
monkeypatch.setattr("utils.transformers_version.subprocess.run", lambda cmd, **k: _proc(0))
assert _probe_tier("org/m", None, "x") == "550"
assert "org/m" not in _probe_tier_cache # skipped a lower tier -> not pinned
def test_cache_hit_skips_subprocess(self, monkeypatch):
self._patch_common(monkeypatch)
monkeypatch.setattr("utils.transformers_version.subprocess.run", lambda cmd, **k: _proc(0))
assert _probe_tier("org/m", None, "x") == "530"
def boom(cmd, **k):
raise AssertionError("should not re-probe a cached model_name")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
assert _probe_tier("org/m", None, "x") == "530"
def test_transient_failure_is_530_and_uncached(self, monkeypatch):
self._patch_common(monkeypatch)
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: _proc(1, "ConnectionError: Max retries exceeded"),
)
assert _probe_tier("org/m", None, "x") == "530"
assert "org/m" not in _probe_tier_cache # retried next load
def test_timeout_is_530_and_uncached(self, monkeypatch):
import subprocess as _sp
self._patch_common(monkeypatch)
def timeout(cmd, **k):
raise _sp.TimeoutExpired(cmd, 60)
monkeypatch.setattr("utils.transformers_version.subprocess.run", timeout)
assert _probe_tier("org/m", None, "x") == "530"
assert "org/m" not in _probe_tier_cache
def test_all_venvs_missing_is_530_no_spawn(self, monkeypatch):
for fn in (
"_ensure_venv_t5_530_exists",
"_ensure_venv_t5_550_exists",
"_ensure_venv_t5_510_exists",
):
monkeypatch.setattr(f"utils.transformers_version.{fn}", lambda: False)
def boom(cmd, **k):
raise AssertionError("no sidecar available; must not spawn")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
assert _probe_tier("org/m", None, "x") == "530"
assert "org/m" not in _probe_tier_cache # nothing probed -> uncached
def test_probe_does_not_import_hub(self, monkeypatch):
# The probe must not import huggingface_hub: that would land before the sidecar is on
# sys.path (activation never purges), pinning the default-env hub. So no in-process sha.
self._patch_common(monkeypatch)
monkeypatch.setattr("utils.transformers_version.subprocess.run", lambda cmd, **k: _proc(0))
sys.modules.pop("huggingface_hub", None)
_probe_tier("org/m", None, "x")
assert "huggingface_hub" not in sys.modules
def test_disable_flag_skips_probe(self, monkeypatch):
monkeypatch.setenv("UNSLOTH_DISABLE_TIER_PROBE", "1")
def boom(cmd, **k):
raise AssertionError("probe disabled; must not spawn")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
assert _probe_tier("org/m", None, "x") == "530"
def test_get_tier_uses_probe_for_remote_tokenizer_signal(self, monkeypatch):
# tokenizer says 5.x but no architecture/substring match -> probe (not a 530 guess).
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_510", lambda m, t = None: False
)
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_550", lambda m, t = None: False
)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: True
)
monkeypatch.setattr("utils.transformers_version._probe_tier", lambda m, t, reason: "510")
assert get_transformers_tier("org/unknown-5x-arch") == "510"
def test_stderr_is_transient(self):
assert _stderr_is_transient("ConnectionError: x") is True
assert _stderr_is_transient("GatedRepoError: need token") is True
assert _stderr_is_transient("KeyError: '-'") is False
assert _stderr_is_transient("ValueError: bad pattern") is False
def test_get_tier_threads_token_to_checks(self, monkeypatch):
# A gated/private model: the token must reach the config/tokenizer checks (and the
# probe), otherwise the authed-only signal is missed and it falls to default 4.x.
seen = {}
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_510",
lambda m, t = None: seen.update({"510": t}) or False,
)
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_550",
lambda m, t = None: seen.update({"550": t}) or False,
)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5",
lambda m, t = None: seen.update({"tok": t}) or True,
)
monkeypatch.setattr(
"utils.transformers_version._probe_tier",
lambda m, t, reason: seen.update({"probe": t}) or "510",
)
assert get_transformers_tier("org/gated-5x", "hf_abc") == "510"
assert seen == {"510": "hf_abc", "550": "hf_abc", "tok": "hf_abc", "probe": "hf_abc"}
def test_activate_threads_token_to_tier(self, monkeypatch):
# activate_transformers_for_subprocess must forward hf_token to tier detection, or
# the gated-model checks above run unauthenticated and the fix is unreachable.
seen = {}
monkeypatch.setattr("utils.transformers_version._resolve_base_model", lambda m: m)
monkeypatch.setattr(
"utils.transformers_version.get_transformers_tier",
lambda m, t = None: seen.update({"model": m, "token": t}) or "default",
)
activate_transformers_for_subprocess("org/gated", "hf_xyz")
assert seen == {"model": "org/gated", "token": "hf_xyz"}
def test_local_checkpoint_reprobes_after_config_change(self, monkeypatch, tmp_path):
# A local checkpoint overwritten in place must re-probe: the cache key folds in the
# config.json signature, so a different config does not serve the stale tier.
self._patch_common(monkeypatch)
cfg = tmp_path / "config.json"
cfg.write_text(json.dumps({"model_type": "a"}))
local = str(tmp_path)
calls = []
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: calls.append(1) or _proc(0),
)
assert _probe_tier(local, None, "x") == "530"
assert _probe_tier(local, None, "x") == "530" # cache hit, no re-spawn
assert len(calls) == 1
cfg.write_text(json.dumps({"model_type": "a_longer_value_changing_the_size"}))
assert _probe_tier(local, None, "x") == "530"
assert len(calls) == 2 # signature changed -> re-probed
def test_probe_child_enables_implicit_token(self, monkeypatch):
# With a token, the probe child must clear an inherited HF_HUB_DISABLE_IMPLICIT_TOKEN=1
# so HF_TOKEN authenticates the gated config fetch instead of 401ing to 530.
self._patch_common(monkeypatch)
monkeypatch.setenv("HF_HUB_DISABLE_IMPLICIT_TOKEN", "1")
captured = {}
def fake_run(cmd, **k):
captured.update(k.get("env") or {})
return _proc(0)
monkeypatch.setattr("utils.transformers_version.subprocess.run", fake_run)
assert _probe_tier("org/gated", "secret-token", "x") == "530"
assert captured.get("HF_TOKEN") == "secret-token"
assert captured.get("HF_HUB_DISABLE_IMPLICIT_TOKEN") == "0"
class TestProbeGating:
"""probe=False suppresses sidecar probes (the log-only needs_transformers_5 path); a
config saved by transformers 5.x is probed default-first so a new 5.x-only arch is
caught without mis-routing a 4.57.x-loadable model onto a sidecar."""
def setup_method(self):
_probe_tier_cache.clear()
_config_json_cache.clear()
_config_needs_510_cache.clear()
_config_needs_550_cache.clear()
_tokenizer_class_cache.clear()
# Sixth cache, and the one this class used to miss. get_transformers_tier consults
# CONFIG_MAPPING_NAMES per tier and upgrades a model_type the ambient default does
# not ship, so a mapping parsed by an earlier test decides the answer here.
_config_mapping_cache.clear()
def _patch_venvs(self, monkeypatch):
for fn in (
"_ensure_venv_t5_530_exists",
"_ensure_venv_t5_550_exists",
"_ensure_venv_t5_510_exists",
):
monkeypatch.setattr(f"utils.transformers_version.{fn}", lambda: True)
monkeypatch.delenv("UNSLOTH_DISABLE_TIER_PROBE", raising = False)
def _patch_checks_to_tokenizer(self, monkeypatch):
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_510", lambda m, t = None: False
)
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_550", lambda m, t = None: False
)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: True
)
# ---- needs_transformers_5 / probe=False must not spawn probes --------------
def test_needs_transformers_5_does_not_spawn_probe(self, monkeypatch):
self._patch_checks_to_tokenizer(monkeypatch)
def boom(cmd, **k):
raise AssertionError("needs_transformers_5 must not spawn a probe")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
# Still correctly reports 5.x from the tokenizer signal, just without probing.
assert needs_transformers_5("org/unknown-5x") is True
def test_probe_false_returns_530_for_tokenizer_signal(self, monkeypatch):
self._patch_checks_to_tokenizer(monkeypatch)
def boom(cmd, **k):
raise AssertionError("probe=False must not spawn a probe")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
assert get_transformers_tier("org/unknown-5x", probe = False) == "530"
# ---- version-field probe is default-first (no mis-routing of 4.x models) ----
def test_version_field_probe_stays_default_when_default_parses(self, monkeypatch):
self._patch_venvs(monkeypatch)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: False
)
# A model_type the ambient default DOES ship. "brandnew" is in none of the
# mappings, so the static config-mapping tier upgraded it to 530 before the
# version-field probe under test ever ran, and the assert below only passed
# when an earlier test had left _config_mapping_cache in a state that skipped
# that path. The probe, not the mapping upgrade, is the subject here.
_config_json_cache[("org/new", None)] = {
"model_type": "llama",
"transformers_version": "5.0.0",
}
seen = []
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: seen.append(cmd[3]) or _proc(0),
)
assert get_transformers_tier("org/new") == "default"
assert seen == [""] # probed the ambient default tier first, it parsed -> stayed default
def test_version_field_probe_escalates_when_default_fails(self, monkeypatch):
import utils.transformers_version as tv
self._patch_venvs(monkeypatch)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: False
)
# Shipped by the ambient default, for the same reason as the test above.
_config_json_cache[("org/new", None)] = {
"model_type": "llama",
"transformers_version": "5.6.0",
}
results = iter([_proc(1, "KeyError: 'x'"), _proc(1, "KeyError: 'x'"), _proc(0)])
seen = []
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: seen.append(cmd[3]) or next(results),
)
assert get_transformers_tier("org/new") == "550"
assert seen == ["", tv._VENV_T5_530_DIR, tv._VENV_T5_550_DIR]
def test_a_model_type_the_default_lacks_never_reaches_the_version_field_probe(
self, monkeypatch
):
"""The mapping check answers first, and short-circuits the probe below it.
This ordering is what quietly invalidated the two tests above. They pinned
model_type "brandnew", which no mapping ships, so once #7043 landed the mapping
check answered before the version-field probe they were written for ever ran, and
they began asserting the wrong branch's answer. Nothing pinned the ordering, so
the only symptom was those two failing in a way that read like a probe bug.
Pinning it here means a future reshuffle that puts the probe first fails as
itself, and the two tests above keep testing the probe rather than this.
"""
import utils.transformers_version as tv
self._patch_venvs(monkeypatch)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: False
)
monkeypatch.setattr(
tv,
"_config_model_types",
lambda tier: frozenset({"llama"} if tier == "default" else {"llama", "brandnew"}),
)
# Declares 5.x too, so the version-field probe below WOULD fire if it were reached.
_config_json_cache[("org/brandnew", None)] = {
"model_type": "brandnew",
"transformers_version": "5.0.0",
}
probed = []
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: probed.append(cmd[3]) or _proc(0),
)
tier = get_transformers_tier("org/brandnew")
assert tier != "default", f"a model_type the default lacks must be raised, got {tier!r}"
assert probed == [], "the mapping check must answer before any sidecar probe runs"
def test_ordinary_4x_config_does_not_probe(self, monkeypatch):
self._patch_venvs(monkeypatch)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: False
)
_config_json_cache[("org/llama", None)] = {
"model_type": "llama",
"transformers_version": "4.57.0",
}
def boom(cmd, **k):
raise AssertionError("a 4.x-saved config must not trigger a probe")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
assert get_transformers_tier("org/llama") == "default"
def test_needs_transformers_5_true_for_version_field_only(self, monkeypatch):
# A 5.x-saved standard-tokenizer model must report as 5.x (for vision routing)
# without spawning a probe.
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_510", lambda m, t = None: False
)
monkeypatch.setattr(
"utils.transformers_version._check_config_needs_550", lambda m, t = None: False
)
monkeypatch.setattr(
"utils.transformers_version._check_tokenizer_config_needs_v5", lambda m, t = None: False
)
_config_json_cache[("org/new", None)] = {
"model_type": "brandnew",
"transformers_version": "5.2.0",
}
def boom(cmd, **k):
raise AssertionError("needs_transformers_5 must not spawn a probe")
monkeypatch.setattr("utils.transformers_version.subprocess.run", boom)
assert needs_transformers_5("org/new") is True
def test_default_first_result_not_reused_for_tokenizer_path(self, monkeypatch, tmp_path):
# A default-first probe can cache "default"; a later tokenizer/known-5.x call
# (floor=530) must re-probe, not reuse that "default".
self._patch_venvs(monkeypatch)
(tmp_path / "config.json").write_text(
json.dumps({"model_type": "brandnew", "transformers_version": "5.0.0"})
)
local = str(tmp_path)
monkeypatch.setattr("utils.transformers_version.subprocess.run", lambda cmd, **k: _proc(0))
assert (
_probe_tier(local, None, "version", include_default = True, floor = "default") == "default"
)
seen = []
monkeypatch.setattr(
"utils.transformers_version.subprocess.run",
lambda cmd, **k: seen.append(cmd[3]) or _proc(0),
)
# Tokenizer/known-5.x mode (floor=530): must re-probe and never reuse "default".
assert _probe_tier(local, None, "tokenizer needs 5.x") == "530"
assert seen, "tokenizer path reused the cached default result instead of re-probing"
class TestLocalCheckpointFilesAppear:
"""A local checkpoint dir inspected before its files exist must not cache the miss or hit
the network, so files written later in the same process are still read (in-progress
checkpoints)."""
def setup_method(self):
_tokenizer_class_cache.clear()
_config_json_cache.clear()
def test_tokenizer_config_appearing_later_is_read(self, tmp_path: Path, monkeypatch):
local = str(tmp_path)
def boom(*a, **k):
raise AssertionError("a local checkpoint must not be fetched from the Hub")
monkeypatch.setattr("urllib.request.urlopen", boom)
# Before the file exists: not 5.x, no network, and the miss must not be pinned.
assert _check_tokenizer_config_needs_v5(local) is False
# The file appears with a 5.x-only tokenizer -> the next call must read it.
(tmp_path / "tokenizer_config.json").write_text(
json.dumps({"tokenizer_class": "TokenizersBackend"})
)
assert _check_tokenizer_config_needs_v5(local) is True
def test_config_json_appearing_later_is_read(self, tmp_path: Path, monkeypatch):
local = str(tmp_path)
def boom(*a, **k):
raise AssertionError("a local checkpoint must not be fetched from the Hub")
monkeypatch.setattr("urllib.request.urlopen", boom)
assert _load_config_json(local) is None
(tmp_path / "config.json").write_text(json.dumps({"model_type": "gemma4"}))
assert _load_config_json(local) == {"model_type": "gemma4"}
# ---------------------------------------------------------------------------
# activate_transformers_for_subprocess — issue #6103
# The early log must make clear it only prepends to sys.path; the real
# confirmation comes later from "Subprocess loaded transformers X.X.X".
# ---------------------------------------------------------------------------
class TestActivateLoggingClarity:
"""issue #6103: 'Activated transformers' was misleading (path-prepend only)."""
def _snapshot_env(self):
return list(sys.path), os.environ.get("PYTHONPATH")
def _restore_env(self, snapshot):
saved_path, saved_pp = snapshot
sys.path[:] = saved_path
if saved_pp is None:
os.environ.pop("PYTHONPATH", None)
else:
os.environ["PYTHONPATH"] = saved_pp
def test_activate_550_log_clarifies_path_prepend_only(self, caplog):
caplog.set_level(logging.INFO)
snap = self._snapshot_env()
try:
with (
patch(
"utils.transformers_version._resolve_base_model",
side_effect = lambda m: m,
),
patch(
"utils.transformers_version.get_transformers_tier",
return_value = "550",
),
patch(
"utils.transformers_version._ensure_venv_t5_550_exists",
return_value = True,
),
):
activate_transformers_for_subprocess("google/gemma-4-E2B-it")
finally:
self._restore_env(snap)
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "5.5.0" in text, f"version not logged: {text!r}"
# Must signal this is only a sys.path manipulation, not a confirmed import.
assert (
"sys.path" in text or "path only" in text
), f"early activation log does not clarify it is path-prepend only: {text!r}"
def test_activate_530_log_clarifies_path_prepend_only(self, caplog):
caplog.set_level(logging.INFO)
snap = self._snapshot_env()
try:
with (
patch(
"utils.transformers_version._resolve_base_model",
side_effect = lambda m: m,
),
patch(
"utils.transformers_version.get_transformers_tier",
return_value = "530",
),
patch(
"utils.transformers_version._ensure_venv_t5_530_exists",
return_value = True,
),
):
activate_transformers_for_subprocess("Qwen/Qwen3.5-9B")
finally:
self._restore_env(snap)
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "5.3.0" in text, f"version not logged: {text!r}"
assert (
"sys.path" in text or "path only" in text
), f"early activation log does not clarify it is path-prepend only: {text!r}"
def test_activate_prefers_local_checkpoint_tier_over_resolved_base(self, caplog, tmp_path):
# Base resolves to an offline/private id (default tier); the local config.json wins.
(tmp_path / "config.json").write_text(json.dumps({"model_type": "llama"}))
local = str(tmp_path)
caplog.set_level(logging.INFO)
snap = self._snapshot_env()
tiers = {local: "510", "private/base": "default"}
try:
with (
patch(
"utils.transformers_version._resolve_base_model",
return_value = "private/base",
),
patch(
"utils.transformers_version.get_transformers_tier",
side_effect = lambda m, t = None: tiers[m],
),
patch(
"utils.transformers_version._ensure_venv_t5_510_exists",
return_value = True,
),
):
activate_transformers_for_subprocess(local)
finally:
self._restore_env(snap)
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "5.10.2" in text, f"local checkpoint tier did not win: {text!r}"
def test_activate_adapter_without_config_skips_path_name_recheck(self, caplog, tmp_path):
# LoRA adapter in a dir named 'gemma-4' (base resolves elsewhere): the resolved
# base drives the tier; the path name must not re-check or upgrade it.
adapter = tmp_path / "gemma-4-experiment" / "llama-lora"
adapter.mkdir(parents = True)
(adapter / "adapter_config.json").write_text(
json.dumps({"base_model_name_or_path": "meta/llama"})
)
local = str(adapter)
caplog.set_level(logging.INFO)
snap = self._snapshot_env()
seen = []
def fake_tier(m, t = None):
seen.append(m)
return "550" if "gemma-4" in m else "default"
try:
with (
patch(
"utils.transformers_version._resolve_base_model",
return_value = "meta/llama",
),
patch(
"utils.transformers_version.get_transformers_tier",
side_effect = fake_tier,
),
):
activate_transformers_for_subprocess(local)
finally:
self._restore_env(snap)
assert seen == ["meta/llama"], f"adapter path was re-checked via substrings: {seen!r}"
text = " ".join(r.getMessage() for r in caplog.records).lower()
assert "default transformers" in text, f"adapter wrongly upgraded: {text!r}"
# ---------------------------------------------------------------------------
# _venv_dir_is_valid — issue #6103
# A version mismatch triggers a full wipe + reinstall, so it must be logged
# at WARNING (not INFO) so the reinstall is visible.
# ---------------------------------------------------------------------------
class TestVenvDirIsValidLogging:
def _make_venv(self, venv_dir: Path, pkg: str, version: str):
"""Create a fake target-dir install of *pkg* at *version*."""
(venv_dir / pkg).mkdir(parents = True)
di = venv_dir / f"{pkg}-{version}.dist-info"
di.mkdir()
(di / "METADATA").write_text(f"Name: {pkg}\nVersion: {version}\n")
def test_version_mismatch_logged_at_warning(self, tmp_path: Path, caplog):
venv_dir = tmp_path / "venv"
self._make_venv(venv_dir, "transformers", "5.0.0") # wrong version
caplog.set_level(logging.INFO)
result = _venv_dir_is_valid(str(venv_dir), ("transformers==5.3.0",))
assert result is False
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
assert warnings, (
"version mismatch must be logged at WARNING; got: "
f"{[(r.levelname, r.getMessage()) for r in caplog.records]!r}"
)
joined = " ".join(r.getMessage() for r in warnings)
assert (
"5.0.0" in joined and "5.3.0" in joined
), f"mismatch log omits the versions: {joined!r}"
def test_correct_version_does_not_warn(self, tmp_path: Path, caplog):
venv_dir = tmp_path / "venv"
self._make_venv(venv_dir, "transformers", "5.3.0") # correct version
caplog.set_level(logging.INFO)
result = _venv_dir_is_valid(str(venv_dir), ("transformers==5.3.0",))
assert result is True
assert not [
r for r in caplog.records if r.levelno >= logging.WARNING
], "no warning expected when the installed version matches"
# ---------------------------------------------------------------------------
# _venv_dir_is_valid_and_undamaged — issue #7715
# A sidecar whose METADATA survived a disk-full or an interrupted pip passes
# every package-level check, so the wipe-and-reinstall in _ensure_venv_dir
# never fires and the worker dies importing transformers instead.
# ---------------------------------------------------------------------------
class TestVenvDirFileIntegrity:
def _make_venv(
self,
venv_dir: Path,
pkg: str = "transformers",
version: str = "5.3.0",
files: dict | None = None,
record_extra: list | None = None,
) -> Path:
"""Fake target-dir install of *pkg* with a RECORD matching what it wrote."""
di = venv_dir / f"{pkg}-{version}.dist-info"
di.mkdir(parents = True)
(di / "METADATA").write_text(f"Name: {pkg}\nVersion: {version}\n")
files = files if files is not None else {f"{pkg}/__init__.py": "x" * 40}
rows = []
for rel, body in files.items():
path = venv_dir / rel
path.parent.mkdir(parents = True, exist_ok = True)
path.write_text(body)
rows.append(f"{rel},sha256=deadbeef,{len(body)}")
rows.extend(record_extra or [])
# pip records its own metadata with a blank size, as real wheels do.
rows.append(f"{pkg}-{version}.dist-info/METADATA,sha256=cafe,")
rows.append(f"{pkg}-{version}.dist-info/RECORD,,")
(di / "RECORD").write_text("\n".join(rows) + "\n")
return venv_dir
def test_intact_sidecar_is_valid(self, tmp_path: Path):
venv_dir = self._make_venv(tmp_path / "venv")
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_truncated_file_is_detected(self, tmp_path: Path, caplog):
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "__init__.py").write_text("x") # disk-full shape
caplog.set_level(logging.INFO)
result = _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
assert result is False, "a truncated file must force the wipe-and-reinstall"
joined = " ".join(r.getMessage() for r in caplog.records if r.levelno >= logging.WARNING)
assert "transformers/__init__.py" in joined, f"damage not named in the log: {joined!r}"
def test_deleted_file_is_detected(self, tmp_path: Path):
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "__init__.py").unlink()
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
@pytest.mark.parametrize(
"ext_name",
[
"_regex.cpython-{tag}-darwin.so",
"_regex.cp{tag}-win_amd64.pyd",
"_regex.cpython-{tag}t-darwin.so",
],
)
def test_extension_built_for_another_interpreter_is_detected(
self, tmp_path: Path, ext_name: str
):
# Pick a stale tag by VERSION, not by whole tag: one template appends "t", so
# deriving it from _CURRENT_EXT_TAG directly builds "313t" -- the current tag --
# when the suite runs on a free-threaded 3.13, and the case asserts damage.
stale = "313" if _CURRENT_EXT_TAG.rstrip("t") != "313" else "312"
venv_dir = self._make_venv(
tmp_path / "venv",
files = {
"transformers/__init__.py": "x" * 40,
f"regex/{ext_name.format(tag = stale)}": "y" * 40,
},
)
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_extension_for_the_running_interpreter_passes(self, tmp_path: Path):
venv_dir = self._make_venv(
tmp_path / "venv",
files = {
"transformers/__init__.py": "x" * 40,
f"regex/_regex.cpython-{_CURRENT_EXT_TAG}-darwin.so": "y" * 40,
"yaml/_yaml.so": "w" * 40,
# Spellings the tag regex deliberately does not recognise. Each has to fail
# OPEN: guessing wrong here costs a several-hundred-MB re-download.
"regex/_regex.pypy311-pp73-x86_64-linux-gnu.so": "p" * 40,
"regex/_regex.graalpy311-310-native-x86_64-linux.so": "g" * 40,
"regex/_regex.cpython-313d-x86_64-linux-gnu.so": "d" * 40,
},
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
@pytest.mark.parametrize(
"rel",
[
# The tag sits in a DIRECTORY component; the binary itself is untagged and fine.
"vendor/build.cp312/libhelper.so",
"pkg.cp312.libs/libfoo.so",
"data/v.cp39/native.pyd",
],
)
def test_a_version_tag_in_a_directory_name_is_not_damage(self, tmp_path: Path, rel: str):
"""Matching the whole RECORD path wipes the sidecar over a directory name."""
venv_dir = self._make_venv(
tmp_path / "venv",
files = {"transformers/__init__.py": "x" * 40, rel: "y" * 40},
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
@pytest.mark.parametrize(
("current_tag", "expect_damage"),
[("313", False), ("314", False), ("313t", True), ("314t", True)],
)
def test_stable_abi_is_damage_only_on_a_free_threaded_build(
self, tmp_path: Path, monkeypatch, current_tag: str, expect_damage: bool
):
"""A GIL build loads any older .abi3.so. A free-threaded one segfaults on it, and
the file survives an in-place interpreter swap because nothing else changes."""
monkeypatch.setattr("utils.transformers_version._CURRENT_EXT_TAG", current_tag)
venv_dir = self._make_venv(
tmp_path / "venv",
files = {
"transformers/__init__.py": "x" * 40,
"hf_xet/hf_xet.abi3.so": "z" * 40,
"win/_thing.abi3.pyd": "z" * 40,
},
)
if expect_damage:
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
else:
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
@pytest.mark.parametrize(
("current_tag", "ext_tag"),
[
# Both directions of the GIL/free-threaded swap, on a fixed pair of versions so
# the case cannot collapse into "current tag" on whichever build runs the suite.
("313t", "313"),
("313", "313t"),
("314t", "314"),
("314", "314t"),
],
)
def test_free_threaded_mismatch_is_detected_in_both_directions(
self, tmp_path: Path, monkeypatch, current_tag: str, ext_tag: str
):
monkeypatch.setattr("utils.transformers_version._CURRENT_EXT_TAG", current_tag)
venv_dir = self._make_venv(
tmp_path / "venv",
files = {
"transformers/__init__.py": "x" * 40,
f"regex/_regex.cpython-{ext_tag}-x86_64-linux-gnu.so": "y" * 40,
},
)
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_damage_in_an_unpinned_dependency_is_detected(self, tmp_path: Path):
"""The whole dir is prepended to sys.path, so a shadowing dep breaks the
import just as surely as transformers itself."""
venv_dir = self._make_venv(tmp_path / "venv")
self._make_venv(venv_dir, pkg = "regex", version = "2026.7.19")
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
(venv_dir / "regex" / "__init__.py").write_text("")
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_console_script_rows_are_ignored(self, tmp_path: Path):
"""pip --target records scripts at a path that cannot resolve.
pip installs through a temporary normal-scheme prefix and writes RECORD
before flattening, so a healthy sidecar records ../../bin/hf while the
file lands in <target>/bin. Believing those rows fails CLOSED on a
healthy install, and since the repair reinstalls with the same pip it
would wipe and re-download several hundred MB on every activation,
forever.
"""
venv_dir = self._make_venv(
tmp_path / "venv",
record_extra = [
"../../bin/hf,sha256=deadbeef,343",
"../../Scripts/hf.exe,sha256=deadbeef,343",
],
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_shared_non_runtime_rows_are_ignored(self, tmp_path: Path):
"""A top-level test/ tree is shared, so one wheel's uninstall deletes
another's files. Nothing imports it: the stdlib shadows `test`, and
`tests`/`scripts` ship no __init__.py."""
venv_dir = self._make_venv(
tmp_path / "venv",
record_extra = ["test/conftest.py,sha256=deadbeef,20650"],
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_shared_non_runtime_rows_are_ignored_for_our_own_wheels_too(self, tmp_path: Path):
"""Unreliable ownership is a property of the path, not of the claimant.
unsloth_zoo <= 2026.8.5 shipped a top-level tests/ into the same squatted
namespace, so exempting only third parties left it reported forever."""
venv_dir = self._make_venv(
tmp_path / "venv",
pkg = "unsloth_zoo",
record_extra = ["tests/conftest.py,sha256=deadbeef,11429"],
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("unsloth_zoo==5.3.0",))
def test_our_own_runtime_tree_is_still_checked(self, tmp_path: Path):
"""Only the shared roots are exempt. A tests/ tree inside our package is
ours alone, so damage there still forces the wipe-and-reinstall."""
venv_dir = self._make_venv(
tmp_path / "venv",
pkg = "unsloth_zoo",
files = {"unsloth_zoo/__init__.py": "x" * 40, "unsloth_zoo/tests/h.py": "y" * 10},
)
(venv_dir / "unsloth_zoo" / "tests" / "h.py").unlink()
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("unsloth_zoo==5.3.0",))
def test_an_installer_rewritten_file_keeps_its_existence_check(self, tmp_path: Path):
"""setup.ps1 runs npm install in the installed tree, so the lockfile's
recorded size drifts. npm never deletes it, so only the size is dropped."""
rel = "studio/backend/core/data_recipe/oxc-validator/package-lock.json"
venv_dir = self._make_venv(
tmp_path / "venv",
record_extra = [f"{rel},sha256=deadbeef,28473"],
)
lock = venv_dir / rel
lock.parent.mkdir(parents = True, exist_ok = True)
lock.write_text("x" * 27225) # shrunk by npm, not damage
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
lock.unlink() # gone entirely, which npm never does
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_in_target_script_rows_are_ignored(self, tmp_path: Path):
"""uv records bin/hf, which does resolve -- but pip --upgrade rmtree's a
colliding directory in the target, so a later install into the same
sidecar deletes an earlier package's scripts. Nothing imports from
there, so the row carries no signal either way."""
venv_dir = self._make_venv(
tmp_path / "venv",
record_extra = [
"bin/hf,sha256=deadbeef,343",
"Scripts/hf.exe,sha256=deadbeef,343",
],
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_skipping_scripts_does_not_mask_real_damage(self, tmp_path: Path):
"""The skip must be scoped to script rows, not a blanket exemption."""
venv_dir = self._make_venv(
tmp_path / "venv", record_extra = ["../../bin/hf,sha256=deadbeef,343"]
)
(venv_dir / "transformers" / "__init__.py").unlink()
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_larger_than_recorded_is_not_damage(self, tmp_path: Path):
"""A packaging collision leaves a bigger file, not a broken one."""
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "__init__.py").write_text("y" * 400)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_multiply_owned_path_skips_the_size_check_but_not_existence(self, tmp_path: Path):
venv_dir = self._make_venv(
tmp_path / "venv",
files = {"transformers/__init__.py": "x" * 40, "shared/mod.py": "z" * 90},
)
self._make_venv(
venv_dir,
pkg = "other",
version = "1.0.0",
files = {"other/__init__.py": "o" * 10},
record_extra = ["shared/mod.py,sha256=beef,999999"],
)
# Two RECORDs disagree on the size; whichever copy landed says nothing
# about either, so this is not damage.
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
# Ambiguous sizes cannot explain the file being gone.
(venv_dir / "shared" / "mod.py").unlink()
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_directory_standing_in_for_a_recorded_file_is_damage(self, tmp_path: Path):
venv_dir = self._make_venv(tmp_path / "venv")
target = venv_dir / "transformers" / "__init__.py"
target.unlink()
target.mkdir() # st_size 4096 on POSIX would sail past a shrinkage test
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_blank_size_row_still_reports_a_deletion(self, tmp_path: Path):
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "data.bin").write_text("q")
record = venv_dir / "transformers-5.3.0.dist-info" / "RECORD"
record.write_text(record.read_text() + "transformers/data.bin,,\n")
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
(venv_dir / "transformers" / "data.bin").unlink()
assert not _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_missing_record_is_not_damage(self, tmp_path: Path):
"""Fails open: the answer to a finding is to delete hundreds of MB."""
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers-5.3.0.dist-info" / "RECORD").unlink()
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_pyc_and_dist_info_rows_are_ignored(self, tmp_path: Path):
venv_dir = self._make_venv(tmp_path / "venv")
record = venv_dir / "transformers-5.3.0.dist-info" / "RECORD"
record.write_text(
record.read_text()
+ "transformers/__pycache__/__init__.cpython-311.pyc,sha256=aa,500\n"
+ "transformers-5.3.0.dist-info/direct_url.json,sha256=bb,500\n"
)
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_directory_rows_are_ignored(self, tmp_path: Path):
venv_dir = self._make_venv(tmp_path / "venv")
record = venv_dir / "transformers-5.3.0.dist-info" / "RECORD"
record.write_text(record.read_text() + "transformers/subdir/,,\n")
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_default_is_package_level_only(self, tmp_path: Path):
"""The routing predicate keeps the cheap semantics: it runs several times
per /validate request with no cache in front of it."""
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "__init__.py").write_text("x")
assert _venv_dir_is_valid(str(venv_dir), ("transformers==5.3.0",)) is True
def test_env_kill_switch_disables_the_scan(self, tmp_path: Path, monkeypatch):
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "__init__.py").write_text("x")
monkeypatch.setenv("UNSLOTH_SKIP_SIDECAR_FILE_CHECK", "1")
assert _venv_dir_is_valid_and_undamaged(str(venv_dir), ("transformers==5.3.0",))
def test_ensure_venv_dir_wipes_and_reinstalls_a_damaged_sidecar(
self, tmp_path: Path, monkeypatch
):
"""End to end: the damage reaches the repair that already exists."""
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / "transformers" / "__init__.py").write_text("x")
installed = []
def _fake_install(pkg, target):
installed.append(pkg)
return True
monkeypatch.setattr("utils.transformers_version._install_to_dir", _fake_install)
ok = _ensure_venv_dir(str(venv_dir), ("transformers==5.3.0",), "transformers 5.3.0")
assert ok is True
assert installed == ["transformers==5.3.0"], "damaged sidecar was not reinstalled"
assert not (venv_dir / "transformers").exists(), "damaged tree was not wiped first"
def test_ensure_venv_dir_restores_the_studio_owned_marker(self, tmp_path: Path, monkeypatch):
"""The wipe takes setup.sh's ownership marker with the old directory. Without a
new one, the next `unsloth studio update` under a custom UNSLOTH_STUDIO_HOME
aborts on a sidecar we just repaired, and adoption cannot rescue it because that
needs a prebuilt-info file a venv never has."""
venv_dir = self._make_venv(tmp_path / "venv")
(venv_dir / ".unsloth-studio-owned").touch()
(venv_dir / "transformers" / "__init__.py").write_text("x")
monkeypatch.setattr("utils.transformers_version._install_to_dir", lambda pkg, target: True)
assert _ensure_venv_dir(str(venv_dir), ("transformers==5.3.0",), "transformers 5.3.0")
assert (venv_dir / ".unsloth-studio-owned").is_file()
# ---------------------------------------------------------------------------
# _ensure_venv_dir — issue #6103
# A slow runtime install must log each package as it starts, otherwise it
# looks like a hang.
# ---------------------------------------------------------------------------
class TestEnsureVenvDirProgressLogging:
def test_logs_each_package_with_progress(self, tmp_path: Path, caplog):
installed = []
caplog.set_level(logging.INFO)
with (
patch(
"utils.transformers_version._venv_dir_is_valid",
return_value = False,
),
patch(
"utils.transformers_version._install_to_dir",
side_effect = lambda pkg, d: (installed.append(pkg), True)[1],
),
):
ok = _ensure_venv_dir(
str(tmp_path / "venv"),
("transformers==5.3.0", "tokenizers==0.21.0"),
"transformers 5.3.0",
)
assert ok is True
assert installed == ["transformers==5.3.0", "tokenizers==0.21.0"]
msgs = " ".join(r.getMessage() for r in caplog.records)
assert "transformers==5.3.0" in msgs, f"first package not logged: {msgs!r}"
assert "tokenizers==0.21.0" in msgs, f"second package not logged: {msgs!r}"
# progress counter present so a slow install is not mistaken for a hang
assert "1/2" in msgs and "2/2" in msgs, f"progress count missing: {msgs!r}"
def test_no_install_logging_when_venv_already_valid(self, tmp_path: Path, caplog):
caplog.set_level(logging.INFO)
with (
patch(
"utils.transformers_version._venv_dir_is_valid",
return_value = True,
),
patch(
"utils.transformers_version._install_to_dir",
) as mock_install,
):
ok = _ensure_venv_dir(
str(tmp_path / "venv"),
("transformers==5.3.0",),
"transformers 5.3.0",
)
assert ok is True
mock_install.assert_not_called()
assert "Installing" not in " ".join(r.getMessage() for r in caplog.records)
# ---------------------------------------------------------------------------
# _tier_from_name — shared name-based detection helper
# ---------------------------------------------------------------------------
class TestTierFromName:
"""Unit tests for _tier_from_name(), which backs both the fast substring
path and the config _name_or_path fallback."""
def test_returns_none_for_unknown(self):
assert _tier_from_name("meta-llama/Llama-3-8B") is None
def test_gemma4_returns_550(self):
tier, _ = _tier_from_name("google/gemma-4-E2B-it")
assert tier == "550"
def test_gemma4_assistant_returns_510(self):
tier, match = _tier_from_name("google/gemma-4-E2B-it-assistant")
assert tier == "510"
assert "assistant" in match
def test_gemma4_12b_returns_510(self):
tier, _ = _tier_from_name("unsloth/gemma-4-12b-it")
assert tier == "510"
def test_qwen35_returns_530(self):
tier, match = _tier_from_name("Qwen/Qwen3.5-7B")
assert tier == "530"
assert "qwen3.5" in match
def test_ministral3_returns_530(self):
# The existing substring "ministral-3-" matches the 2512 naming style.
tier, _ = _tier_from_name("mistralai/Ministral-3-8B-Instruct-2512")
assert tier == "530"
def test_qwen3_moe_substring_returns_530(self):
tier, _ = _tier_from_name("Qwen/Qwen3-30B-A3B-Instruct-2507")
assert tier == "530"
def test_510_beats_550(self):
"""gemma-4-12b matches 510 (checked first), not 550."""
tier, _ = _tier_from_name("google/gemma-4-12b-it")
assert tier == "510"
def test_550_beats_530(self):
"""gemma-4 matches 550, not 530."""
tier, _ = _tier_from_name("gemma-4-model")
assert tier == "550"
# ---------------------------------------------------------------------------
# Local-folder tier detection via config.json
#
# When a local checkpoint's config.json architecture/model_type matches a known
# sidecar set, that's the authoritative answer. When it doesn't match (unknown
# or future family), the HF model ID from _name_or_path / model_name in the
# config is run through the same name-based rules so renamed folders are still
# routed correctly without introducing path false-positives.
# ---------------------------------------------------------------------------
class TestLocalConfig530Tier:
def setup_method(self):
_config_json_cache.clear()
_tokenizer_class_cache.clear()
_config_needs_530_cache.clear()
# --- config-set matches -------------------------------------------------
def test_config_needs_530_qwen3_5_model_type(self):
assert _config_needs_530({"model_type": "qwen3_5"}) is True
def test_config_needs_530_qwen3_5_conditional_generation(self):
assert _config_needs_530({"architectures": ["Qwen3_5ForConditionalGeneration"]}) is True
def test_config_needs_530_qwen3_moe(self):
assert _config_needs_530({"model_type": "qwen3_moe"}) is True
def test_config_needs_530_glm4_moe_lite(self):
assert _config_needs_530({"model_type": "glm4_moe_lite"}) is True
def test_config_needs_530_lfm2_vl(self):
assert _config_needs_530({"model_type": "lfm2_vl"}) is True
def test_config_needs_530_qwen3_5_moe(self):
"""Qwen3.5 MoE (Qwen3.5-35B-A3B / 122B-A10B) uses qwen3_5_moe ids."""
assert (
_config_needs_530(
{
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
}
)
is True
)
def test_config_needs_530_qwen3_next(self):
assert (
_config_needs_530(
{"model_type": "qwen3_next", "architectures": ["Qwen3NextForCausalLM"]}
)
is True
)
def test_config_needs_530_qwen3_5_text_towers(self):
"""Text-tower configs (architectures may be stripped) still need 5.3.0."""
assert _config_needs_530({"model_type": "qwen3_5_text"}) is True
assert _config_needs_530({"model_type": "qwen3_5_moe_text"}) is True
def test_config_needs_530_plain_qwen3_is_false(self):
"""Regular Qwen3 (non-MoE, non-3.5) must not be promoted to 5.3.0."""
assert _config_needs_530({"model_type": "qwen3"}) is False
def test_tier_local_qwen35_config_selects_530(self, tmp_path: Path):
"""Reported case: a local Qwen3.5 folder routes to 530 via config.json."""
d = tmp_path / "Qwen3.5-2B"
d.mkdir()
(d / "config.json").write_text(json.dumps({"model_type": "qwen3_5"}))
assert get_transformers_tier(str(d)) == "530"
def test_tier_local_qwen3_moe_config_selects_530(self, tmp_path: Path):
"""Local Qwen3 MoE checkpoint routes to 530 via config.json."""
d = tmp_path / "my-qwen3-moe"
d.mkdir()
(d / "config.json").write_text(
json.dumps({"model_type": "qwen3_moe", "architectures": ["Qwen3MoeForCausalLM"]})
)
assert get_transformers_tier(str(d)) == "530"
def test_tier_local_glm4_moe_lite_config_selects_530(self, tmp_path: Path):
"""Local GLM-4.7-Flash checkpoint routes to 530 via config.json."""
d = tmp_path / "my-glm-model"
d.mkdir()
(d / "config.json").write_text(
json.dumps({"model_type": "glm4_moe_lite", "architectures": ["Glm4MoeLiteForCausalLM"]})
)
assert get_transformers_tier(str(d)) == "530"
def test_tier_local_lfm2_vl_config_selects_530(self, tmp_path: Path):
"""Local LFM2.5-VL checkpoint routes to 530 via config.json."""
d = tmp_path / "my-liquid-model"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{"model_type": "lfm2_vl", "architectures": ["Lfm2VlForConditionalGeneration"]}
)
)
assert get_transformers_tier(str(d)) == "530"
def test_tier_local_qwen35_moe_config_selects_530(self, tmp_path: Path):
"""A renamed Qwen3.5 MoE folder (no name hint) routes to 530 via config."""
d = tmp_path / "my-custom-moe"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
}
)
)
assert get_transformers_tier(str(d)) == "530"
# --- Qwen3.6 reuses Qwen3.5 config ids but routes to 550 by name ---------
def test_local_qwen36_config_keeps_550_name_tier(self, tmp_path: Path):
"""Qwen3.6 config carries qwen3_5 ids; a higher-tier name match wins."""
d = tmp_path / "Qwen3.6-27B"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{"model_type": "qwen3_5", "architectures": ["Qwen3_5ForConditionalGeneration"]}
)
)
assert get_transformers_tier(str(d)) == "550"
def test_local_qwen36_moe_via_name_or_path_keeps_550(self, tmp_path: Path):
"""Renamed Qwen3.6 MoE folder: _name_or_path carries the 5.5 name signal."""
d = tmp_path / "renamed-q36-moe"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
"_name_or_path": "Qwen/Qwen3.6-35B-A3B",
}
)
)
assert get_transformers_tier(str(d)) == "550"
def test_stale_absolute_name_or_path_not_promoted(self, tmp_path: Path):
"""A non-5.x checkpoint with a stale absolute _name_or_path isn't name-matched."""
d = tmp_path / "my-llama-ckpt"
d.mkdir()
(d / "config.json").write_text(
json.dumps({"model_type": "llama", "_name_or_path": "/old/run/qwen3.5-source"})
)
with patch(
"utils.transformers_version._check_tokenizer_config_needs_v5", return_value = False
):
assert get_transformers_tier(str(d)) == "default"
# --- _name_or_path fallback ---------------------------------------------
def test_renamed_folder_falls_back_to_hf_id_in_config(self, tmp_path: Path):
"""A renamed local folder with an unrecognised model_type but a known
HF ID in _name_or_path still routes to the correct tier."""
d = tmp_path / "my-custom-name"
d.mkdir()
# Simulate a future/unknown model_type; the HF ID carries the tier signal.
(d / "config.json").write_text(
json.dumps(
{
"model_type": "future_unknown_type",
"_name_or_path": "Qwen/Qwen3.5-7B",
}
)
)
assert get_transformers_tier(str(d)) == "530"
def test_hf_id_fallback_respects_550_tier(self, tmp_path: Path):
"""_name_or_path pointing to a Gemma-4 HF ID routes to 550."""
d = tmp_path / "renamed-gemma"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "future_unknown_type",
"_name_or_path": "google/gemma-4-E2B-it",
}
)
)
assert get_transformers_tier(str(d)) == "550"
def test_hf_id_fallback_skipped_when_same_as_path(self, tmp_path: Path):
"""If _name_or_path equals the model path, skip the name fallback to
avoid false positives from self-referencing configs."""
d = tmp_path / "qwen3.5-experiment"
d.mkdir()
# _name_or_path is the local path itself (e.g. saved via save_pretrained)
(d / "config.json").write_text(
json.dumps(
{
"model_type": "llama",
"_name_or_path": str(d),
}
)
)
with patch(
"utils.transformers_version._check_tokenizer_config_needs_v5", return_value = False
):
# "qwen3.5" is in the path but config says llama and _name_or_path
# is self-referencing — must not be promoted to 530.
assert get_transformers_tier(str(d)) == "default"
def test_hf_id_fallback_not_triggered_when_name_or_path_is_absolute_self(self, tmp_path: Path):
"""_name_or_path == absolute path of the same checkpoint while model_name
is a relative path: the two strings differ, but both point to the same
directory. The absolute path must not be scanned for tier substrings."""
d = tmp_path / "qwen3.5-experiment"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "llama",
# absolute path — textually different from a relative model_name
"_name_or_path": str(d),
}
)
)
with patch(
"utils.transformers_version._check_tokenizer_config_needs_v5", return_value = False
):
# Even though str(d) contains "qwen3.5", the local-dir branch recurses
# into config checks on the resolved path, which returns default.
assert get_transformers_tier(str(d)) == "default"
# --- false-positive guard -----------------------------------------------
def test_tier_local_plain_model_still_default(self, tmp_path: Path):
"""A local non-5.x checkpoint returns default; the directory-name
false-positive guard is preserved."""
d = tmp_path / "checkpoint-1000"
d.mkdir()
(d / "config.json").write_text(
json.dumps({"architectures": ["LlamaForCausalLM"], "model_type": "llama"})
)
with patch(
"utils.transformers_version._check_tokenizer_config_needs_v5",
return_value = False,
):
assert get_transformers_tier(str(d)) == "default"
# ---------------------------------------------------------------------------
# _check_config_needs_530 — slow HF-ID path (network stub)
# ---------------------------------------------------------------------------
class TestCheckConfigNeeds530:
"""_check_config_needs_530 is used in the slow HF-ID fallback path for
private or renamed repos whose names don't contain a 5.3 substring."""
def setup_method(self):
_config_json_cache.clear()
_config_needs_530_cache.clear()
def test_returns_true_for_qwen3_5_model_type(self):
with patch(
"utils.transformers_version._load_config_json",
return_value = {"model_type": "qwen3_5"},
):
assert _check_config_needs_530("some-private/qwen3.5-variant") is True
def test_returns_true_for_qwen3_moe_architecture(self):
with patch(
"utils.transformers_version._load_config_json",
return_value = {"architectures": ["Qwen3MoeForCausalLM"]},
):
assert _check_config_needs_530("org/private-moe-model") is True
def test_returns_false_for_llama(self):
with patch(
"utils.transformers_version._load_config_json",
return_value = {"model_type": "llama", "architectures": ["LlamaForCausalLM"]},
):
assert _check_config_needs_530("meta-llama/Llama-3-8B") is False
def test_returns_false_when_config_unavailable(self):
with patch("utils.transformers_version._load_config_json", return_value = None):
assert _check_config_needs_530("org/unreachable-model") is False
def test_result_is_cached(self, tmp_path: Path):
"""A definitive (local) read is cached by (model, token), mirroring 510/550."""
cfg = {"model_type": "qwen3_5"}
(tmp_path / "config.json").write_text(json.dumps(cfg))
key = (str(tmp_path), None)
_check_config_needs_530(str(tmp_path))
assert key in _config_needs_530_cache
assert _config_needs_530_cache[key] is True
# ---------------------------------------------------------------------------
# _norm_separators
# ---------------------------------------------------------------------------
class TestNormSeparators:
def test_underscore_to_hyphen(self):
assert _norm_separators("qwen3_5") == "qwen3-5"
def test_dot_preserved(self):
assert _norm_separators("qwen3.5") == "qwen3.5"
def test_hyphen_unchanged(self):
assert _norm_separators("gemma-4") == "gemma-4"
def test_mixed(self):
assert _norm_separators("Qwen3_5.MoE") == "Qwen3-5.MoE"
def test_whitespace_to_hyphen(self):
assert _norm_separators("some model") == "some-model"
def test_empty(self):
assert _norm_separators("") == ""
# ---------------------------------------------------------------------------
# _tier_from_name — separator-insensitive matching
# ---------------------------------------------------------------------------
class TestTierFromNameSeparatorNorm:
"""Verify that underscore/dot aliases in model IDs resolve to the same
tier as their canonical hyphen/dot counterparts."""
def test_qwen3_underscore_5_returns_530(self):
tier, _ = _tier_from_name("Qwen/Qwen3_5-7B")
assert tier == "530"
def test_qwen3_next_underscore_returns_530(self):
tier, _ = _tier_from_name("org/Qwen3_Next-14B")
assert tier == "530"
def test_gemma_4_underscore_returns_550(self):
tier, _ = _tier_from_name("google/gemma_4_E2B_it")
assert tier == "550"
def test_gemma_4_12b_underscore_returns_510(self):
tier, _ = _tier_from_name("unsloth/gemma_4_12b_it")
assert tier == "510"
def test_canonical_dot_still_works(self):
tier, _ = _tier_from_name("Qwen/Qwen3.5-7B")
assert tier == "530"
def test_unrelated_underscores_not_promoted(self):
assert _tier_from_name("meta_llama/Llama_3_8B") is None
def test_qwen3_hyphen_6_size_not_promoted(self):
"""Qwen3-6B is a size name, not the qwen3.6 release line."""
assert _tier_from_name("Qwen/Qwen3-6B-Instruct") is None
def test_qwen3_hyphen_5_size_not_promoted(self):
assert _tier_from_name("Qwen/Qwen3-5B") is None
# ---------------------------------------------------------------------------
# _resolve_base_model — model_name-then-_name_or_path fallback
# ---------------------------------------------------------------------------
class TestResolveBaseModelNameOrPathFallback:
"""When config.json has both 'model_name' (self-referential local path) and
'_name_or_path' (original HF ID), _resolve_base_model must use _name_or_path."""
def test_name_or_path_used_when_model_name_is_self_ref(self, tmp_path: Path):
d = tmp_path / "my-qwen35-finetune"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_name": str(d),
"_name_or_path": "Qwen/Qwen3.5-7B",
}
)
)
assert _resolve_base_model(str(d)) == "Qwen/Qwen3.5-7B"
def test_model_name_used_when_not_self_ref(self, tmp_path: Path):
d = tmp_path / "adapter"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_name": "unsloth/Qwen3.5-7B-bnb-4bit",
"_name_or_path": "Qwen/Qwen3.5-7B",
}
)
)
# model_name is not the local path, so it wins
assert _resolve_base_model(str(d)) == "unsloth/Qwen3.5-7B-bnb-4bit"
def test_tier_resolved_via_name_or_path_when_model_name_self_refs(self, tmp_path: Path):
"""End-to-end: get_transformers_tier picks up the sidecar tier from
_name_or_path even when model_name is set to the checkpoint's own path."""
d = tmp_path / "my-custom-finetune"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "future_unknown_type",
"model_name": str(d),
"_name_or_path": "Qwen/Qwen3.5-7B",
}
)
)
assert get_transformers_tier(str(d)) == "530"
def test_local_config_tier_not_bypassed_by_private_name_or_path(self, tmp_path: Path):
"""Full checkpoint with model_type: qwen3_5 must still route to 530 even
when _name_or_path is a private HF ID with no recognisable tier substring.
Regression guard: before the adapter-only pre-resolve fix,
activate_transformers_for_subprocess would resolve to the private HF ID
and then fail to probe it offline, returning default instead of 530.
"""
d = tmp_path / "my-finetuned-model"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "qwen3_5",
"model_name": str(d),
"_name_or_path": "my-org/private-custom-id",
}
)
)
# get_transformers_tier reads config.json directly and returns 530
# without needing to probe the private HF ID.
assert get_transformers_tier(str(d)) == "530"
# ---------------------------------------------------------------------------
# adapter_model-only LoRA resolution (no adapter_config.json)
# ---------------------------------------------------------------------------
class TestAdapterModelOnlyLoRA:
"""A LoRA dir with adapter_model*.safetensors but no adapter_config.json must
still be detected as an adapter and resolved to its base model so the worker
activates the base model's sidecar instead of tiering off the adapter folder."""
def test_has_adapter_weights_detects_safetensors_and_bin(self, tmp_path: Path):
d = tmp_path / "adapter"
d.mkdir()
assert _has_adapter_weights(d) is False
(d / "adapter_model.safetensors").write_text("")
assert _has_adapter_weights(d) is True
d2 = tmp_path / "adapter_bin"
d2.mkdir()
(d2 / "adapter_model.bin").write_text("")
assert _has_adapter_weights(d2) is True
def test_is_lora_adapter_dir_for_config_and_weights_only(self, tmp_path: Path):
# adapter_config.json present
a = tmp_path / "cfg"
a.mkdir()
(a / "adapter_config.json").write_text("{}")
assert _is_lora_adapter_dir(a) is True
# adapter_model weights only, no config
b = tmp_path / "weights_only"
b.mkdir()
(b / "adapter_model.safetensors").write_text("")
assert _is_lora_adapter_dir(b) is True
# plain checkpoint dir (neither)
c = tmp_path / "plain"
c.mkdir()
(c / "config.json").write_text("{}")
assert _is_lora_adapter_dir(c) is False
# not a directory
assert _is_lora_adapter_dir(tmp_path / "missing") is False
def test_resolve_adapter_only_lora_via_unsloth_dir_name(self, tmp_path: Path):
"""adapter_model-only LoRA with the unsloth_<model>_<ts> naming resolves to
unsloth/<model> through the import-light directory-name parse."""
d = tmp_path / "unsloth_Qwen3.5-7B_20260620"
d.mkdir()
(d / "adapter_model.safetensors").write_text("")
assert _resolve_base_model(str(d)) == "unsloth/Qwen3.5-7B"
def test_activation_pre_resolves_adapter_only_lora(self, tmp_path: Path):
"""Regression: activate_transformers_for_subprocess must pre-resolve an
adapter_model-only LoRA dir (weights present, adapter_config.json absent).
Before the gate used _is_lora_adapter_dir, the adapter_config-only check
skipped resolution and the worker tiered off the adapter folder itself."""
d = tmp_path / "my-custom-lora"
d.mkdir()
(d / "adapter_model.safetensors").write_text("")
snap = (list(sys.path), os.environ.get("PYTHONPATH"))
try:
with (
patch(
"utils.transformers_version._resolve_base_model",
side_effect = lambda m: m,
) as mock_resolve,
patch(
"utils.transformers_version.get_transformers_tier",
return_value = "default",
),
):
activate_transformers_for_subprocess(str(d))
finally:
sys.path[:] = snap[0]
if snap[1] is None:
os.environ.pop("PYTHONPATH", None)
else:
os.environ["PYTHONPATH"] = snap[1]
mock_resolve.assert_called_once_with(str(d))
# ---------------------------------------------------------------------------
# 530-config override must not be flipped by stale local path hints
# ---------------------------------------------------------------------------
class TestConfig530OverrideGuard:
"""A correct 530 config must not be flipped to 550 by a 5.5-looking substring in
a stale/renamed local path; only a real Hub id or the folder basename may override."""
def test_stale_local_path_does_not_flip_530_to_550(self, tmp_path: Path):
d = tmp_path / "my-qwen35-run"
d.mkdir()
(d / "config.json").write_text(
json.dumps(
{
"model_type": "qwen3_5",
"_name_or_path": "/old/run/qwen3.6-source",
}
)
)
# Stale path is not a Hub id, so the 530 config wins over its qwen3.6 substring.
assert get_transformers_tier(str(d)) == "530"
def test_current_basename_can_still_override_to_550(self, tmp_path: Path):
d = tmp_path / "Qwen3.6-27B"
d.mkdir()
# Qwen3.6 reuses the qwen3_5 config id but is a 5.5 model by name.
(d / "config.json").write_text(
json.dumps({"model_type": "qwen3_5", "_name_or_path": str(d)})
)
assert get_transformers_tier(str(d)) == "550"
def test_real_hub_id_can_still_override_to_550(self, tmp_path: Path):
d = tmp_path / "my-finetune"
d.mkdir()
(d / "config.json").write_text(
json.dumps({"model_type": "qwen3_5", "_name_or_path": "Qwen/Qwen3.6-27B"})
)
assert get_transformers_tier(str(d)) == "550"
def test_remote_qwen36_name_or_path_overrides_530(self):
"""Slow path: a fetched qwen3_5 config naming Qwen3.6 in _name_or_path -> 550."""
_config_needs_530_cache.clear()
_config_json_cache.clear()
with patch(
"utils.transformers_version._load_config_json",
return_value = {"model_type": "qwen3_5", "_name_or_path": "Qwen/Qwen3.6-27B"},
):
assert get_transformers_tier("private/renamed-q36") == "550"
class TestLooksLikeHfId:
def test_empty_and_whitespace_are_not_ids(self):
assert _looks_like_hf_id("") is False
assert _looks_like_hf_id(" ") is False
def test_plain_hub_id(self):
assert _looks_like_hf_id("Qwen/Qwen3.5-7B") is True
def test_absolute_and_dot_paths_are_not_ids(self):
assert _looks_like_hf_id("/old/run/qwen3.5-source") is False
assert _looks_like_hf_id("./qwen3.5-source") is False
def test_existing_local_path_is_not_an_id(self, tmp_path: Path):
d = tmp_path / "Qwen3.5-7B"
d.mkdir()
import os as _os
cwd = _os.getcwd()
try:
_os.chdir(tmp_path)
# "Qwen3.5-7B" exists relative to cwd, so it is a path, not a Hub id.
assert _looks_like_hf_id("Qwen3.5-7B") is False
finally:
_os.chdir(cwd)
class TestMalformedInputRobustness:
"""Tier detection fails open to default instead of crashing on bad input."""
def setup_method(self):
_config_json_cache.clear()
_config_needs_530_cache.clear()
def test_non_string_model_type_does_not_crash(self):
assert _config_needs_530({"model_type": ["qwen3_5"]}) is False
def test_non_list_architectures_does_not_crash(self):
assert _config_needs_530({"architectures": "Qwen3_5ForCausalLM"}) is False
def test_local_config_non_string_fields_returns_default(self, tmp_path: Path):
d = tmp_path / "weird"
d.mkdir()
(d / "config.json").write_text(
json.dumps({"model_type": ["qwen3_5"], "_name_or_path": {"x": 1}})
)
with patch(
"utils.transformers_version._check_tokenizer_config_needs_v5", return_value = False
):
assert get_transformers_tier(str(d)) == "default"
def test_pathological_long_name_does_not_crash(self):
# An over-long name makes is_file() raise OSError; must fail open.
assert get_transformers_tier("x" * 5000) == "default"
def test_empty_name_returns_default(self):
assert get_transformers_tier("") == "default"
# ---------------------------------------------------------------------------
# Offline negatives must not poison the version caches (persistent worker)
# ---------------------------------------------------------------------------
class TestOfflineCacheNotPoisoned:
"""An offline first load must not leave a stale negative for a later online read."""
def setup_method(self):
_tokenizer_class_cache.clear()
_config_json_cache.clear()
def test_offline_tokenizer_assumption_not_cached(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_env_offline", lambda: True)
# No local file, not a local dir -> offline branch returns False without caching.
assert _check_tokenizer_config_needs_v5("org/uncached") is False
assert ("org/uncached", None) not in _tokenizer_class_cache
def test_offline_then_online_refetches(self, monkeypatch):
import utils.transformers_version as tv
# 1) Offline: returns False, nothing cached.
monkeypatch.setattr(tv, "_env_offline", lambda: True)
assert _check_tokenizer_config_needs_v5("org/needs5") is False
assert ("org/needs5", None) not in _tokenizer_class_cache
# 2) Back online: the real fetch runs (cache was not poisoned) and is honored.
monkeypatch.setattr(tv, "_env_offline", lambda: False)
class _Resp:
def read(self):
return json.dumps({"tokenizer_class": "TokenizersBackend"}).encode()
def __enter__(self):
return self
def __exit__(self, *a):
return False
monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout = 10: _Resp())
assert _check_tokenizer_config_needs_v5("org/needs5") is True
def test_offline_config_miss_not_cached(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_env_offline", lambda: True)
monkeypatch.setattr(tv, "_config_json_from_hf_cache", lambda name: None)
assert _load_config_json("org/uncached-config", None) is None
assert ("org/uncached-config", None) not in _config_json_cache
# ---------------------------------------------------------------------------
# hf_endpoint_unreachable — bounded, proxy/egress-aware reachability probe
# ---------------------------------------------------------------------------
class TestHfEndpointUnreachable:
# Ambient proxies are cleared module-wide by _no_ambient_proxy.
def test_reachable_returns_false(self, monkeypatch):
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
monkeypatch.setattr("urllib.request.urlopen", lambda *a, **k: _Resp())
assert hf_endpoint_unreachable(timeout = 2) is False
def test_gateway_error_is_unreachable(self, monkeypatch):
import urllib.error
def _gw(*a, **k):
raise urllib.error.HTTPError("http://x", 504, "Gateway Timeout", {}, None)
monkeypatch.setattr("urllib.request.urlopen", _gw)
assert hf_endpoint_unreachable(timeout = 2) is True
def test_other_http_status_is_reachable(self, monkeypatch):
import urllib.error
def _405(*a, **k):
raise urllib.error.HTTPError("http://x", 405, "Method Not Allowed", {}, None)
monkeypatch.setattr("urllib.request.urlopen", _405)
assert hf_endpoint_unreachable(timeout = 2) is False
def test_tls_failure_is_reachable(self, monkeypatch):
import ssl
import urllib.error
def _tls(*a, **k):
raise urllib.error.URLError(ssl.SSLCertVerificationError("self-signed"))
monkeypatch.setattr("urllib.request.urlopen", _tls)
# TLS reached the server: treat as reachable so the load surfaces the cert error.
assert hf_endpoint_unreachable(timeout = 2) is False
def test_connection_refused_is_reachable(self, monkeypatch):
import urllib.error
def _refused(*a, **k):
raise urllib.error.URLError(ConnectionRefusedError("refused"))
monkeypatch.setattr("urllib.request.urlopen", _refused)
assert hf_endpoint_unreachable(timeout = 2) is False
def test_connection_reset_is_reachable(self, monkeypatch):
"""A reset proves the path answered, exactly like the refusal above.
urllib wraps an OSError raised while sending, so this is the URLError form.
"""
import urllib.error
def _reset(*a, **k):
raise urllib.error.URLError(ConnectionResetError(104, "Connection reset by peer"))
monkeypatch.setattr("urllib.request.urlopen", _reset)
assert hf_endpoint_unreachable(timeout = 2) is False
def test_remote_disconnect_is_reachable(self, monkeypatch):
"""getresponse() is not inside urllib's OSError->URLError wrapper, so a server
that accepts the HEAD then closes raises http.client.RemoteDisconnected raw --
and it subclasses ConnectionResetError -> OSError."""
import http.client
def _disconnect(*a, **k):
raise http.client.RemoteDisconnected("Remote end closed connection without response")
monkeypatch.setattr("urllib.request.urlopen", _disconnect)
assert hf_endpoint_unreachable(timeout = 2) is False
def test_real_server_that_accepts_then_closes_is_reachable(self, monkeypatch):
"""End-to-end over a real socket: no mocking of the exception type at all."""
import socket
import threading
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("127.0.0.1", 0))
srv.listen(4)
def _accept_then_close():
try:
conn, _ = srv.accept()
except OSError:
return
try:
conn.recv(4096)
except OSError:
pass
conn.close()
thread = threading.Thread(target = _accept_then_close, daemon = True)
thread.start()
monkeypatch.setenv("HF_ENDPOINT", f"http://127.0.0.1:{srv.getsockname()[1]}")
try:
assert hf_endpoint_unreachable(timeout = 2) is False
finally:
srv.close()
thread.join(5)
def test_dns_failure_is_unreachable(self, monkeypatch):
import socket
import urllib.error
def _dns(*a, **k):
raise urllib.error.URLError(socket.gaierror(-2, "Name or service not known"))
monkeypatch.setattr("urllib.request.urlopen", _dns)
assert hf_endpoint_unreachable(timeout = 2) is True
def test_hung_probe_is_bounded(self, monkeypatch):
import time
def _hang(*a, **k):
time.sleep(30)
monkeypatch.setattr("urllib.request.urlopen", _hang)
t0 = time.time()
result = hf_endpoint_unreachable(timeout = 2)
assert result is True and (time.time() - t0) < 6.0
class TestLatestTierActiveFor:
"""latest_tier_active_for: the 16-bit guard for the consented latest sidecar."""
@staticmethod
def _pin(
monkeypatch,
tv,
version = "5.13.1",
):
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: version)
monkeypatch.setattr(tv, "_remote_lora_base", lambda name, hf_token = None: None)
def test_true_when_tier_latest(self, monkeypatch):
import utils.transformers_version as tv
self._pin(monkeypatch, tv)
monkeypatch.setattr(tv, "get_transformers_tier", lambda *a, **k: "latest")
assert tv.latest_tier_active_for("Zyphra/ZAYA1-8B") is True
def test_false_for_fixed_tiers(self, monkeypatch):
import utils.transformers_version as tv
self._pin(monkeypatch, tv)
for tier in ("default", "530", "550", "510"):
monkeypatch.setattr(tv, "get_transformers_tier", lambda *a, _t = tier, **k: _t)
assert tv.latest_tier_active_for("some/model") is False
def test_false_without_pin_and_no_resolution(self, monkeypatch):
"""No sidecar pin returns False before any tier or network resolution."""
import utils.transformers_version as tv
def _boom(*a, **k):
raise AssertionError("must not resolve without a pin")
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: None)
monkeypatch.setattr(tv, "_remote_lora_base", _boom)
monkeypatch.setattr(tv, "get_transformers_tier", _boom)
assert tv.latest_tier_active_for("Zyphra/ZAYA1-8B") is False
def test_never_raises(self, monkeypatch):
import utils.transformers_version as tv
def _boom(*a, **k):
raise RuntimeError("tier resolution exploded")
self._pin(monkeypatch, tv)
monkeypatch.setattr(tv, "get_transformers_tier", _boom)
assert tv.latest_tier_active_for("some/model") is False
def test_remote_lora_base_is_resolved(self, monkeypatch):
"""A remote adapter is judged by its base model, like worker activation."""
import utils.transformers_version as tv
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: "5.13.1")
monkeypatch.setattr(tv, "_remote_lora_base", lambda name, hf_token = None: "Zyphra/ZAYA1-8B")
tiers = {"Zyphra/ZAYA1-8B": "latest"}
monkeypatch.setattr(
tv, "get_transformers_tier", lambda name, *a, **k: tiers.get(name, "default")
)
assert tv.latest_tier_active_for("someuser/zaya-lora") is True
def test_local_checkpoint_config_upgrades(self, monkeypatch, tmp_path):
"""An adapter dir with its own config.json merges tiers like activation does."""
import utils.transformers_version as tv
adapter = tmp_path / "ckpt"
adapter.mkdir()
(adapter / "adapter_config.json").write_text("{}")
(adapter / "adapter_model.safetensors").write_text("x")
(adapter / "config.json").write_text("{}")
self._pin(monkeypatch, tv)
monkeypatch.setattr(tv, "_resolve_base_model", lambda name: "base/model")
tiers = {"base/model": "default", str(adapter): "latest"}
monkeypatch.setattr(
tv, "get_transformers_tier", lambda name, *a, **k: tiers.get(name, "default")
)
assert tv.latest_tier_active_for(str(adapter)) is True
class TestLatestTierForces16Bit:
"""The inference worker and load route refuse bnb 4-bit on the latest sidecar."""
def _read(self, rel):
backend_dir = Path(__file__).resolve().parent.parent
return (backend_dir / rel).read_text(encoding = "utf-8")
def test_worker_guard_present(self):
src = self._read("core/inference/worker.py")
assert "latest_tier_active_for" in src, (
"core/inference/worker.py must force load_in_4bit=False when "
"latest_tier_active_for(model) is true: transformers' grouped-MoE "
"kernels crash on bnb-quantized expert weights for brand-new "
"architectures."
)
def test_route_guard_present(self):
src = self._read("routes/inference.py")
assert "latest_tier_active_for" in src, (
"routes/inference.py must size the VRAM guard with the same 16-bit "
"flip the worker applies for latest-sidecar models."
)
def test_validate_route_mirrors_16bit_flip(self):
# Without the same flip, /validate sizes 4-bit and /load then 409s.
src = self._read("routes/inference.py")
body = src.split("async def validate_model", 1)[1].split("\nasync def ", 1)[0]
assert "latest_tier_active_for" in body, (
"validate_model must apply the latest-sidecar 16-bit flip before "
"_guard_chat_load_against_training so /validate and /load agree."
)
# First-time loads have no pin yet, so an installable upgrade must also size 16-bit.
assert body.index("check_upgrade_for_model") < body.index(
"_guard_chat_load_against_training"
), "the upgrade check must run before the training guard"
assert (
"supported_in_pypi" in body.split("_guard_chat_load_against_training")[0]
), "an installable upgrade must force 16-bit sizing for the guard"
def test_validate_offered_upgrade_preserves_custom_code_4bit(self):
# A merely-offered (not installed) upgrade must NOT force 16-bit sizing when the
# model has a custom-code (auto_map) fallback: /load loads it 4-bit without the
# install, and the install route refuses during active training, so 16-bit sizing
# here would 409 the only viable 4-bit path.
src = self._read("routes/inference.py")
body = src.split("async def validate_model", 1)[1].split("\nasync def ", 1)[0]
flip = body.split("Mirror /load's latest-sidecar 16-bit flip", 1)[1].split(
"_guard_chat_load_against_training", 1
)[0]
assert "not requires_trust_remote_code" in flip, (
"the offered-upgrade 16-bit flip must be gated on the absence of a custom-code "
"fallback so /validate does not 409 a 4-bit load /load would allow"
)
# requires_trust_remote_code must be resolved before the flip consumes it.
# Anchored on the resolving call, not its expression form: the any() now runs
# inside _offline_guarded on a worker thread.
assert body.index("_requires_trust_remote_code_for_model(_t") < body.index(
"not requires_trust_remote_code"
)
def test_install_route_guards_active_latest_workers(self):
# Stage-and-swap replaces .venv_t5_latest in place, so a live worker on the
# old sidecar would lazy-import files from the new version.
src = self._read("routes/inference.py")
body = src.split("async def install_latest_transformers_route", 1)[1].split(
"\nasync def ", 1
)[0]
assert (
"is_training_active" in body
and "is_export_active" in body
and "inference_lifecycle_gate" in body
), (
"install_latest_transformers_route must refuse while training or export "
"runs, and hold the lifecycle gate while unloading the chat model and "
"swapping the sidecar."
)
# The unload (via before_swap so failed installs keep the model), the export-worker
# teardown, and the install must all sit INSIDE the gate so no /load interleaves.
assert "unload_model(active)" in body
assert "cleanup_memory()" in body
# Export teardown precedes the chat unload so its failure aborts with the model still loaded.
assert body.index("cleanup_memory()") < body.index("unload_model(active)")
assert "install_latest_transformers(" in body and "_unload_before_swap" in body
# The gate must be owned by the shielded task, not the request coroutine: a cancelled
# POST unwinding an async-with would release the only guard /load honors mid-install.
gated_task = body.split("async def _gated_install", 1)[1]
assert "inference_lifecycle_gate():" in gated_task
assert "asyncio.to_thread(_run_install)" in gated_task
# The reservation must be taken BEFORE the (awaitable) gate wait, or a
# training/export start could slip in while this request queues on the gate.
assert body.index("try_begin_sidecar_swap()") < body.index(
"inference_lifecycle_gate():"
), "the swap reservation must be raised before waiting on the lifecycle gate"
# A failed teardown must abort the swap (raise), not fall through to it.
assert body.count("raise RuntimeError") >= 3, (
"export, chat-unload, and idle-worker teardown failures must raise so "
"the staged install never swaps under a live worker"
)
# The installer thread owns (and releases) the reservation, shielded from
# request cancellation, so a cancelled POST cannot unlock a live swap.
assert "asyncio.shield" in body and "end_sidecar_swap()" in body
# In-flight generation streams predate the gate; the route refuses rather than kill them
# via the before_swap unload. The count is rechecked UNDER the gate, since a wait on a
# long /load outlasts the pre-gate fast path and streams take this same gate.
assert "other_inference_request_count" in body
gated_task = body.split("async def _gated_install", 1)[1]
assert "other_inference_request_count" in gated_task
def test_start_routes_refuse_during_install(self):
# A worker spawned mid-swap could activate a half-replaced sidecar.
training = self._read("routes/training.py")
start = training.split("async def start_training", 1)[1].split("\nasync def ", 1)[0]
assert (
"is_install_in_progress" in start
), "training /start must refuse while a transformers install is in progress"
export = self._read("routes/export.py")
helper = export.split("def _ensure_export_supported", 1)[1].split("\ndef ", 1)[0]
assert (
"is_install_in_progress" in helper
), "mutating export routes must refuse while a transformers install is in progress"
def test_spawn_sites_recheck_reservation(self):
# The route-level guards are one-shot; validation between them and the
# actual spawn can outlast an install's start, so the spawn itself rechecks.
training = self._read("core/training/training.py")
assert (
training.count("sidecar_swap_in_progress()") >= 2
), "both training spawn sites must recheck the sidecar swap reservation"
export = self._read("core/export/orchestrator.py")
spawn = export.split("def _spawn_subprocess", 1)[1].split("\n def ", 1)[0]
assert (
"sidecar_swap_kind()" in spawn
), "the export subprocess spawn must recheck the sidecar swap reservation"
# Training marks the spawn active BEFORE its recheck, so either side sees the other:
# is_training_active covers the window between proc.start() and the _proc assignment.
assert training.index("self._spawn_in_progress = True") < training.index(
"if sidecar_swap_in_progress():"
)
active = training.split("def is_training_active", 1)[1].split("\n def ", 1)[0]
assert "_spawn_in_progress" in active
# Export load-checkpoint refuses BEFORE tearing down the old worker, so a
# lost race against an install keeps the loaded checkpoint (no bare 500).
loadck = export.split("def load_checkpoint", 1)[1].split("\n def ", 1)[0]
assert loadck.index("sidecar_swap_in_progress()") < loadck.index("_shutdown_subprocess()")
# The training handshake precedes the VRAM-freeing before_spawn hook, so
# losing the race never tears down chat/export for a run that won't spawn.
assert training.index("self._spawn_in_progress = True") < training.index("before_spawn()")
# The spawn-time export check is op-aware for installs (the install side
# aborts on is_export_active) but always refuses for repairs, which have
# no such abort and can be rebuilding the sidecar right now.
assert (
'_swap_kind == "repair" or (_swap_kind is not None and not self._export_active)'
in spawn
)
class TestSidecarSwapReservation:
"""The lazy repair takes the same reservation the install route and worker starts use."""
def _repair_setup(self, monkeypatch, tmp_path):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(tmp_path / "venv_t5_latest"))
monkeypatch.setattr(
tv,
"_latest_pin_data",
lambda: {
"version": "5.99.0",
"packages": ["transformers==5.99.0"],
},
)
monkeypatch.setattr(tv, "_venv_dir_is_valid", lambda d, p: False)
monkeypatch.setattr(tv, "_env_offline", lambda: False)
return tv
def test_repair_holds_reservation_during_swap(self, monkeypatch, tmp_path):
tv = self._repair_setup(monkeypatch, tmp_path)
seen = {}
def _fake_swap(
version,
packages,
before_swap = None,
):
seen["active_during_swap"] = tv.sidecar_swap_in_progress()
return True
monkeypatch.setattr(tv, "_stage_and_swap_latest_venv", _fake_swap)
assert tv._ensure_venv_t5_latest_exists() is True
assert seen["active_during_swap"] is True
assert tv.sidecar_swap_in_progress() is False
def test_foreign_process_lock_file_visible(self, monkeypatch, tmp_path):
"""A repair in a LIVE worker subprocess is seen (via the lock file) by this
process, and its lock is never broken while the owner is alive."""
import os
import time
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(tmp_path / "venv_t5_latest"))
lock = tv._swap_lock_path()
lock.parent.mkdir(parents = True, exist_ok = True)
# A live owner (this process): visible and never reclaimed, even once aged past
# the cutoff -- a slow but live pip install must keep its lock.
lock.write_text('{"pid": %d}' % os.getpid())
assert tv.sidecar_swap_in_progress() is True
assert tv.try_begin_sidecar_swap() is False
old_ts = time.time() - 3 * 60 * 60
os.utime(lock, (old_ts, old_ts))
assert tv.sidecar_swap_in_progress() is True
assert tv.try_begin_sidecar_swap() is False
def test_dead_owner_lock_reclaimed_promptly(self, monkeypatch, tmp_path):
"""A fresh lock whose recorded owner is dead is reclaimed at once, not after the
long cutoff: a crash mid-install must not wedge loads/training/export for hours."""
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(tmp_path / "venv_t5_latest"))
lock = tv._swap_lock_path()
lock.parent.mkdir(parents = True, exist_ok = True)
# 999999 is not a live PID: a fresh dead-owner lock is immediately stale.
lock.write_text('{"pid": 999999, "kind": "install"}')
assert tv._pid_alive(999999) is False
assert tv.sidecar_swap_in_progress() is False
assert tv.try_begin_sidecar_swap() is True
try:
assert lock.is_file()
finally:
tv.end_sidecar_swap()
assert not lock.exists()
def test_unreadable_pid_lock_uses_age_cutoff(self, monkeypatch, tmp_path):
"""A lock with no readable owner PID (mid create-before-write, or corrupt) is not
reclaimed while fresh -- only after the long cutoff -- so a lock a live owner just
created is not stolen before its PID lands."""
import os
import time
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(tmp_path / "venv_t5_latest"))
lock = tv._swap_lock_path()
lock.parent.mkdir(parents = True, exist_ok = True)
lock.write_text("") # created but metadata not yet written
assert tv.sidecar_swap_in_progress() is True
old_ts = time.time() - (tv._SWAP_LOCK_STALE_SECS + 60)
os.utime(lock, (old_ts, old_ts))
assert tv.sidecar_swap_in_progress() is False
def test_repair_refused_while_install_holds_reservation(self, monkeypatch, tmp_path):
tv = self._repair_setup(monkeypatch, tmp_path)
def _must_not_run(*a, **k):
raise AssertionError("repair must not swap while an install is in progress")
monkeypatch.setattr(tv, "_stage_and_swap_latest_venv", _must_not_run)
assert tv.try_begin_sidecar_swap() is True
try:
assert tv._ensure_venv_t5_latest_exists() is False
finally:
tv.end_sidecar_swap()
class TestRecoverStrandedSidecar:
"""A swap whose activation rename AND rollback both fail strands the previous sidecar
at .old with no live dir (its pin marker went with it). Reading the pin self-heals it,
but never while a swap legitimately holds the reservation."""
def _setup(self, monkeypatch, tmp_path):
import utils.transformers_version as tv
live = str(tmp_path / "venv_t5_latest")
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", live)
# Stranded state: live gone, previous sidecar (with its marker) sits at .old.
retired = Path(live + ".old")
retired.mkdir(parents = True)
(retired / tv._LATEST_PIN_MARKER).write_text(
'{"version": "5.99.0", "packages": ["transformers==5.99.0"]}'
)
return tv, Path(live), retired
def test_stranded_old_recovered_on_pin_read(self, monkeypatch, tmp_path):
tv, live, retired = self._setup(monkeypatch, tmp_path)
data = tv._latest_pin_data()
assert live.is_dir()
assert not retired.exists()
assert data is not None and data["version"] == "5.99.0"
def test_stranded_recovery_skipped_during_swap(self, monkeypatch, tmp_path):
tv, live, retired = self._setup(monkeypatch, tmp_path)
assert tv.try_begin_sidecar_swap() is True
try:
# A swap holds the reservation and may be mid-rename; do not race it.
assert tv._latest_pin_data() is None
assert not live.exists()
assert retired.is_dir()
finally:
tv.end_sidecar_swap()
# Once the swap is done, the next pin read recovers the stranded sidecar.
assert tv._latest_pin_data() is not None
assert live.is_dir()
class TestCachedLatestMappingRevalidated:
"""A cached 'latest' mapping is dropped and re-resolved when the sidecar since broke
in-process, so routing self-heals instead of trusting a mapping parsed from a sidecar
that no longer exists (which would keep routing latest-only models to a broken tier)."""
def test_broken_sidecar_drops_cached_latest_mapping(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_config_mapping_cache", {"latest": frozenset({"brandnew"})})
monkeypatch.setattr(tv, "_latest_sidecar_intact", lambda: False)
seen = {"n": 0}
def _fake_overlay(tier):
seen["n"] += 1
return None # broken/unavailable -> empty, uncached
monkeypatch.setattr(tv, "_overlay_transformers_dir", _fake_overlay)
assert tv._config_model_types("latest") == frozenset()
assert seen["n"] == 1 # re-resolved, not served from the stale cache
assert "latest" not in tv._config_mapping_cache
def test_intact_sidecar_serves_cached_latest_mapping(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_config_mapping_cache", {"latest": frozenset({"brandnew"})})
monkeypatch.setattr(tv, "_latest_sidecar_intact", lambda: True)
monkeypatch.setattr(
tv,
"_overlay_transformers_dir",
lambda tier: pytest.fail("intact sidecar must serve the cache without re-resolving"),
)
assert tv._config_model_types("latest") == frozenset({"brandnew"})
def test_non_latest_cache_not_revalidated(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_config_mapping_cache", {"530": frozenset({"gemma3"})})
monkeypatch.setattr(
tv,
"_latest_sidecar_intact",
lambda: pytest.fail("non-latest tiers must not pay the sidecar-intact check"),
)
assert tv._config_model_types("530") == frozenset({"gemma3"})
def test_deleted_pin_drops_cached_latest_mapping(self, monkeypatch, tmp_path):
# A pin marker deleted after the mapping was cached makes _latest_pin_data None;
# the cache must be dropped (not trusted), so routing re-resolves to no latest tier
# rather than routing to a latest tier that then fails worker activation.
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(tmp_path / "venv_t5_latest"))
monkeypatch.setattr(tv, "_latest_tier_disabled", lambda: False)
monkeypatch.setattr(tv, "_config_mapping_cache", {"latest": frozenset({"brandnew"})})
# No pin marker on disk -> _latest_pin_data() is None -> not intact.
assert tv._latest_sidecar_intact() is False
assert tv._config_model_types("latest") == frozenset()
assert "latest" not in tv._config_mapping_cache
class TestOverlayRepairsIncompleteSidecar:
"""Routing self-heals a pinned latest sidecar that is present but incomplete,
not only one whose transformers/ dir vanished: workers refuse parent-only
repairs, so a sidecar missing a pinned package would fail every load."""
def _setup(self, monkeypatch, tmp_path, valid):
import utils.transformers_version as tv
live = tmp_path / "venv_t5_latest"
(live / "transformers").mkdir(parents = True)
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(live))
# Every other tier dir too, not just latest.
#
# _install_to_dir below is patched but the DESTINATIONS were not, and _probe_tier
# walks 530, 550 and 510 provisioning each one it finds missing. So this class
# wrote a fake transformers 5.99.0 sidecar, whose CONFIG_MAPPING_NAMES is
# {"brandnew": "C"}, into the developer's real ~/.unsloth/studio.
#
# It then poisons the NEXT run rather than this one, which is why it stayed
# hidden: tier resolution finds "brandnew" in 530, and the three tests here that
# assert an unknown model type routes to "latest" get "530" instead. Reproducible
# on plain main, in isolation, on any machine that has run this file before, and
# observed on the CI runner too.
for name in ("_VENV_T5_530_DIR", "_VENV_T5_550_DIR", "_VENV_T5_510_DIR"):
monkeypatch.setattr(tv, name, str(live.parent / name.lower()))
monkeypatch.setattr(tv, "_latest_tier_disabled", lambda: False)
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: "5.99.0")
monkeypatch.setattr(
tv,
"_latest_pin_data",
lambda: {"version": "5.99.0", "packages": ["transformers==5.99.0", "tiktoken"]},
)
monkeypatch.setattr(tv, "_venv_dir_is_valid", lambda d, p: valid)
monkeypatch.setattr(tv, "_latest_repair_failed_at", 0.0)
return tv
def test_incomplete_sidecar_triggers_repair(self, monkeypatch, tmp_path):
tv = self._setup(monkeypatch, tmp_path, valid = False)
called = {"n": 0}
def _fake_repair():
called["n"] += 1
return True
monkeypatch.setattr(tv, "_ensure_venv_t5_latest_exists", _fake_repair)
src = tv._overlay_transformers_dir("latest")
assert called["n"] == 1
assert src == str(tmp_path / "venv_t5_latest" / "transformers")
def test_intact_sidecar_skips_repair(self, monkeypatch, tmp_path):
tv = self._setup(monkeypatch, tmp_path, valid = True)
def _must_not_run():
raise AssertionError("intact sidecar must not trigger a repair")
monkeypatch.setattr(tv, "_ensure_venv_t5_latest_exists", _must_not_run)
assert tv._overlay_transformers_dir("latest") == str(
tmp_path / "venv_t5_latest" / "transformers"
)
def test_failed_repair_backs_off(self, monkeypatch, tmp_path):
tv = self._setup(monkeypatch, tmp_path, valid = False)
called = {"n": 0}
def _fake_repair():
called["n"] += 1
return False
monkeypatch.setattr(tv, "_ensure_venv_t5_latest_exists", _fake_repair)
# A failed repair must not route through the broken sidecar, neither on
# the failing attempt nor while the backoff suppresses the next attempt.
assert tv._overlay_transformers_dir("latest") is None
assert tv._overlay_transformers_dir("latest") is None
assert called["n"] == 1
class TestStageAndSwapBeforeSwap:
"""before_swap fires only when the staged install succeeded and the swap is next."""
def _setup(self, monkeypatch, tmp_path, build_ok):
import utils.transformers_version as tv
live = tmp_path / "venv_latest"
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(live))
def _fake_build(target, packages, label):
if build_ok:
Path(target).mkdir(parents = True, exist_ok = True)
return build_ok
monkeypatch.setattr(tv, "_ensure_venv_dir", _fake_build)
return tv, live
def test_called_after_successful_staging(self, monkeypatch, tmp_path):
tv, live = self._setup(monkeypatch, tmp_path, build_ok = True)
calls = []
assert tv._stage_and_swap_latest_venv(
"5.99.0", ("transformers==5.99.0",), before_swap = lambda: calls.append(1)
)
assert calls == [1] and live.is_dir()
def test_not_called_when_staging_fails(self, monkeypatch, tmp_path):
tv, live = self._setup(monkeypatch, tmp_path, build_ok = False)
calls = []
assert not tv._stage_and_swap_latest_venv(
"5.99.0", ("transformers==5.99.0",), before_swap = lambda: calls.append(1)
)
assert calls == [] and not live.exists()
def test_failure_in_before_swap_keeps_previous_sidecar(self, monkeypatch, tmp_path):
tv, live = self._setup(monkeypatch, tmp_path, build_ok = True)
live.mkdir()
(live / "sentinel").write_text("old")
def _boom():
raise RuntimeError("worker teardown failed")
assert not tv._stage_and_swap_latest_venv(
"5.99.0", ("transformers==5.99.0",), before_swap = _boom
)
assert (live / "sentinel").read_text() == "old"
class TestKillSwitchBeatsMappingCache:
def test_cached_latest_probe_ignored_when_disabled(self, monkeypatch):
import utils.transformers_version as tv
key = tv._probe_cache_key("some/model")
monkeypatch.setitem(tv._probe_tier_cache, key, "latest")
# A cached 'latest' only ever arises while the tier is pinned, since an unpinned
# one is not in the probe order at all, and an unpinned cache entry is now
# re-probed in its own right. Hold the pin so this stays a test of the switch.
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: "5.99.0")
monkeypatch.setenv("UNSLOTH_STUDIO_NO_LATEST_TRANSFORMERS", "1")
# With the switch set, the cached latest entry must not short-circuit;
# the probe re-resolves against the non-latest order (stub it to 530).
monkeypatch.setattr(tv, "_probe_tier_venvs", lambda: {})
monkeypatch.setattr(tv, "_probe_tier_order", lambda: ())
assert tv._probe_tier("some/model", None, "test") != "latest"
# Cached non-latest entries and the unset switch still short-circuit.
monkeypatch.delenv("UNSLOTH_STUDIO_NO_LATEST_TRANSFORMERS")
assert tv._probe_tier("some/model", None, "test") == "latest"
def test_cached_latest_mapping_ignored_when_disabled(self, monkeypatch):
import utils.transformers_version as tv
monkeypatch.setitem(tv._config_mapping_cache, "latest", frozenset({"brandnew"}))
# The cache is trusted only when the sidecar is intact; hold it intact so this
# test isolates the kill switch, not the sidecar-revalidation path.
monkeypatch.setattr(tv, "_latest_sidecar_intact", lambda: True)
monkeypatch.setenv("UNSLOTH_STUDIO_NO_LATEST_TRANSFORMERS", "1")
assert tv._config_model_types("latest") == frozenset()
monkeypatch.delenv("UNSLOTH_STUDIO_NO_LATEST_TRANSFORMERS")
assert tv._config_model_types("latest") == frozenset({"brandnew"})
class TestRaiseTierForNested:
"""_raise_tier_for_nested: a wrapper's nested model_type can raise a fast-path tier."""
def _patch_types(self, monkeypatch, per_tier):
import utils.transformers_version as tv
monkeypatch.setattr(
tv, "_config_model_types", lambda tier: frozenset(per_tier.get(tier, ()))
)
def test_nested_latest_only_type_raises(self, monkeypatch):
import utils.transformers_version as tv
self._patch_types(monkeypatch, {"550": {"gemma4"}, "latest": {"gemma4", "brandnew_arch"}})
cfg = {"model_type": "gemma4", "text_config": {"model_type": "brandnew_arch"}}
assert tv._raise_tier_for_nested(cfg, "550") == "latest"
def test_never_lowers_a_fast_path_tier(self, monkeypatch):
import utils.transformers_version as tv
# Mapping alone would say 530, but the fast path (e.g. a name override) said 550.
self._patch_types(monkeypatch, {"530": {"qwen3_5"}, "550": {"qwen3_5"}})
assert tv._raise_tier_for_nested({"model_type": "qwen3_5"}, "550") == "550"
def test_no_config_keeps_tier(self):
import utils.transformers_version as tv
assert tv._raise_tier_for_nested(None, "550") == "550"
def test_unknown_nested_type_never_vetoes(self, monkeypatch):
import utils.transformers_version as tv
# A nested type unknown everywhere (not even latest) keeps the fast path.
self._patch_types(monkeypatch, {"550": {"gemma4"}, "latest": {"gemma4"}})
cfg = {"model_type": "gemma4", "text_config": {"model_type": "unreleased"}}
assert tv._raise_tier_for_nested(cfg, "550") == "550"
def test_name_fast_path_folds_when_latest_pinned(self, monkeypatch):
"""A fixed-tier name match with a latest-only model_type routes to latest
once the sidecar is pinned; without a pin the name tier stands (no I/O)."""
import utils.transformers_version as tv
self._patch_types(monkeypatch, {"550": {"gemma4"}, "latest": {"brandnew_arch"}})
monkeypatch.setattr(tv, "_tier_from_name", lambda name: ("550", "gemma-4"))
monkeypatch.setattr(
tv, "_load_config_json", lambda name, tok = None: {"model_type": "brandnew_arch"}
)
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: "5.99.0")
assert tv.get_transformers_tier("org/gemma-4-new", probe = False) == "latest"
monkeypatch.setattr(tv, "latest_venv_pinned_version", lambda: None)
monkeypatch.setattr(
tv,
"_load_config_json",
lambda name, tok = None: (_ for _ in ()).throw(AssertionError("no I/O without a pin")),
)
assert tv.get_transformers_tier("org/gemma-4-new", probe = False) == "550"
def test_fast_path_folds_nested_tier(self, monkeypatch, tmp_path):
"""End to end: a local wrapper config on a fixed fast path routes to latest
when its nested type only exists in the installed latest sidecar."""
import utils.transformers_version as tv
ckpt = tmp_path / "wrapper"
ckpt.mkdir()
(ckpt / "config.json").write_text(
json.dumps({"model_type": "gemma4", "text_config": {"model_type": "brandnew_arch"}})
)
self._patch_types(monkeypatch, {"550": {"gemma4"}, "latest": {"gemma4", "brandnew_arch"}})
monkeypatch.setattr(tv, "_config_needs_510", lambda cfg: False)
monkeypatch.setattr(tv, "_config_needs_550", lambda cfg: True)
assert tv.get_transformers_tier(str(ckpt), probe = False) == "latest"
class TestDamagedLatestSidecarRepairHandoff:
"""A worker child refuses to repair the sidecar, so damage it finds must reach the
parent's routing self-heal. The routing predicate is package-level and cannot see a
truncated file, so without a handoff every worker retry fails forever waiting for a
repair nothing ever triggers."""
def _sidecar(
self,
root: Path,
model_types = ("brandnew",),
) -> Path:
"""A pinned latest sidecar whose RECORD matches what it wrote."""
import utils.transformers_version as tv
di = root / "transformers-5.99.0.dist-info"
di.mkdir(parents = True, exist_ok = True)
(di / "METADATA").write_text("Name: transformers\nVersion: 5.99.0\n")
mapping = ", ".join(f'"{t}": "C"' for t in model_types)
files = {
"transformers/__init__.py": "x" * 40,
"transformers/models/auto/configuration_auto.py": "CONFIG_MAPPING_NAMES = {%s}\n"
% mapping,
}
rows = []
for rel, body in files.items():
path = root / rel
path.parent.mkdir(parents = True, exist_ok = True)
path.write_text(body)
rows.append(f"{rel},sha256=deadbeef,{len(body)}")
rows.append("transformers-5.99.0.dist-info/RECORD,,")
(di / "RECORD").write_text("\n".join(rows) + "\n")
(root / tv._LATEST_PIN_MARKER).write_text(
json.dumps({"version": "5.99.0", "packages": ["transformers==5.99.0"]})
)
return root
def _patch(
self,
monkeypatch,
live: Path,
is_child: bool = False,
):
"""Point the module at *live* and make a repair succeed without pip."""
import multiprocessing
import utils.transformers_version as tv
monkeypatch.setattr(tv, "_VENV_T5_LATEST_DIR", str(live))
# The other three tiers go to tmp_path as well, and this is not tidiness.
#
# _fake_install below writes a sidecar at whatever target it is handed, and
# _probe_tier walks _PROBE_TIER_ORDER provisioning each tier it tries. With only
# the latest dir redirected, the 530, 550 and 510 targets were the REAL ones under
# ~/.unsloth/studio, so running this file wrote a fake transformers 5.99.0 whose
# CONFIG_MAPPING_NAMES is {"brandnew": "C"} into the developer's own Studio.
#
# It then failed the next run of this same class: _lowest_tier_for("brandnew")
# found it in tier 530 and returned "530" where the test asserts "latest". Three
# tests, on a clean checkout of main, only on a machine that had run the suite
# before. That is also what CI reproduced, since a runner accumulates the same
# state within one session.
# Derived from the module rather than listed, because listing them is what went
# wrong: the list would have to be updated by whoever adds a tier, and the
# consequence of forgetting is invisible until a later run of an unrelated test.
# This also caught _VENV_T5_DIR, a fifth constant aliasing the 550 sidecar that a
# hand-written list of the three obvious ones missed.
for _tier_dir in [
name for name in dir(tv) if name.startswith("_VENV_T5_") and name.endswith("_DIR")
]:
if _tier_dir == "_VENV_T5_LATEST_DIR":
continue # already pointed at `live`, which is the sidecar under test
monkeypatch.setattr(tv, _tier_dir, str(live.parent / _tier_dir.lower().strip("_")))
monkeypatch.setattr(tv, "_latest_tier_disabled", lambda: False)
monkeypatch.setattr(tv, "_env_offline", lambda: False)
monkeypatch.setattr(tv, "_workers_active_for_repair", lambda: False)
monkeypatch.setattr(tv, "_config_mapping_cache", {})
monkeypatch.setattr(tv, "_latest_repair_failed_at", 0.0)
monkeypatch.setattr(
multiprocessing, "parent_process", lambda: object() if is_child else None
)
installs = []
def _fake_install(pkg, target):
installs.append(pkg)
self._sidecar(Path(target))
return True
monkeypatch.setattr(tv, "_install_to_dir", _fake_install)
return tv, installs
def test_every_sidecar_dir_is_redirected_away_from_the_real_studio_home(
self, monkeypatch, tmp_path
):
"""Adding a fifth tier must not silently start writing to the developer's home.
_fake_install writes a sidecar at whatever target it is handed, and _probe_tier
walks the tiers provisioning each one it tries. Redirecting only the latest dir
meant the other three targets were the real ones, and this file wrote a fake
transformers 5.99.0 into ~/.unsloth/studio, then failed its own next run.
So the assertion is over whatever tier constants the module defines, not over the
four that exist today: a new _VENV_T5_..._DIR that _patch does not redirect fails
here rather than in whichever unrelated test happens to read the tier mapping
next.
"""
import utils.transformers_version as tv
tiers = [name for name in dir(tv) if name.startswith("_VENV_T5_") and name.endswith("_DIR")]
assert tiers, "no tier directory constants found; this guard would pass on nothing"
self._patch(monkeypatch, self._sidecar(tmp_path / "venv_t5_latest"))
stray = {
name: getattr(tv, name)
for name in tiers
if not str(getattr(tv, name)).startswith(str(tmp_path))
}
assert not stray, (
f"_patch leaves {stray} pointing outside tmp_path. A repair or probe that "
f"provisions one of those writes test fixture data into the real Studio "
f"install, which poisons tier resolution for every later run on that machine."
)
def _damage(self, live: Path):
"""Disk-full shape: the file is still there, just short."""
(live / "transformers" / "__init__.py").write_text("x")
def _is_damaged(self, tv, live: Path) -> bool:
return bool(tv._sidecar_damaged_files(str(live)))
def test_damage_predating_the_process_heals_before_any_worker_starts(
self, monkeypatch, tmp_path
):
"""The parent resolves the mapping through _overlay_transformers_dir on a cache
miss, which is the one place that can afford the RECORD scan."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, installs = self._patch(monkeypatch, live)
self._damage(live)
assert tv._venv_dir_is_valid(
str(live), ("transformers==5.99.0",)
), "precondition: the package-level predicate cannot see this damage"
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
assert installs == ["transformers==5.99.0"], "the parent did not self-heal"
assert not self._is_damaged(tv, live)
def test_worker_child_flags_damage_instead_of_only_refusing(self, monkeypatch, tmp_path):
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, installs = self._patch(monkeypatch, live, is_child = True)
self._damage(live)
assert tv._ensure_venv_t5_latest_exists() is False, "a child must not repair"
assert installs == []
assert tv._latest_repair_requested(), "the child left no signal for the parent"
def test_flagged_damage_breaks_the_worker_retry_deadlock(self, monkeypatch, tmp_path):
"""The reported deadlock, end to end.
Damage that appears AFTER this process cached the latest mapping is invisible to
every parent-side predicate cheap enough to run per request, so the parent keeps
routing to 'latest' and the child keeps refusing. Pinned here is that the loop
terminates: the child's flag makes the next parent routing call repair, and the
retry after that activates.
"""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, installs = self._patch(monkeypatch, live)
# Warm the mapping cache from the healthy sidecar, as any earlier request does.
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
assert "latest" in tv._config_mapping_cache
installs.clear()
self._damage(live)
# Parent keeps routing to 'latest' off the cached mapping and never scans.
for _ in range(3):
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
assert installs == [], "the cached hot path must not pay for a scan"
assert self._is_damaged(tv, live)
# The worker child sees the damage, refuses, and flags it.
monkeypatch.setattr("multiprocessing.parent_process", lambda: object())
assert tv._ensure_venv_t5_latest_exists() is False
assert installs == []
# The next parent routing call repairs. Before the fix it never did, and the
# worker below failed forever.
monkeypatch.setattr("multiprocessing.parent_process", lambda: None)
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
assert installs == ["transformers==5.99.0"], "the parent never repaired"
assert not self._is_damaged(tv, live)
# The worker retry now succeeds, which is the property that was deadlocked.
monkeypatch.setattr("multiprocessing.parent_process", lambda: object())
assert tv._ensure_venv_t5_latest_exists() is True
assert not tv._latest_repair_requested(), "a satisfied request must not persist"
def test_stale_flag_on_a_healthy_sidecar_is_cleared_not_acted_on(self, monkeypatch, tmp_path):
"""A false positive here costs a several-hundred-MB re-download, so a flag a
clean scan contradicts must be dropped, and must not re-arm the branch."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, installs = self._patch(monkeypatch, live)
sentinel = live / "transformers" / "__init__.py"
tv._request_latest_repair()
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
assert installs == [], "a healthy sidecar must never be wiped on a stale flag"
assert sentinel.read_text() == "x" * 40
assert not tv._latest_repair_requested(), "the stale flag must not survive"
def test_cached_mapping_hit_never_scans_the_sidecar(self, monkeypatch, tmp_path):
"""The revalidation on a _config_mapping_cache HIT runs several times per
/validate request, so it must stay package level."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
assert tv._config_model_types("latest") == frozenset({"brandnew"})
monkeypatch.setattr(
tv,
"_sidecar_damaged_files",
lambda d, limit = 3: pytest.fail("the cached hot path must not scan RECORDs"),
)
for _ in range(5):
assert tv._config_model_types("latest") == frozenset({"brandnew"})
def test_repair_backoff_window_never_repeats_the_scan(self, monkeypatch, tmp_path):
"""A sidecar that cannot be repaired (offline, pip down) must not put a ~25 ms
scan on every routing call for the whole backoff window."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
monkeypatch.setattr(tv, "_install_to_dir", lambda pkg, target: False)
self._damage(live)
assert tv._overlay_transformers_dir("latest") is None # scans, tries, fails
monkeypatch.setattr(
tv,
"_sidecar_damaged_files",
lambda d, limit = 3: pytest.fail("the backoff window must not re-scan"),
)
for _ in range(5):
tv._overlay_transformers_dir("latest")
def test_backoff_window_keeps_the_damaged_sidecar_withheld(self, monkeypatch, tmp_path):
"""Backing off from pip must not double as declaring the sidecar usable: the
cheap predicate cannot see a truncated file, so without the marker every
routing call in the window routes latest-only models straight back into a
sidecar whose worker activation is known to fail."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
monkeypatch.setattr(tv, "_install_to_dir", lambda pkg, target: False)
self._damage(live)
assert tv._overlay_transformers_dir("latest") is None # scans, tries, fails
assert tv._venv_dir_is_valid(
str(live), ("transformers==5.99.0",)
), "precondition: only the scan can see this damage, not the cheap predicate"
monkeypatch.setattr(
tv,
"_sidecar_damaged_files",
lambda d, limit = 3: pytest.fail("the backoff window must not re-scan"),
)
for _ in range(5):
assert tv._overlay_transformers_dir("latest") is None
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) != "latest"
def test_every_unrepaired_exit_flags_the_damage(self, monkeypatch, tmp_path):
"""Offline, a child, and a swap already running each return before any repair.
Whichever one is taken, the damage has to be recorded, or the cheap predicate
keeps reporting the sidecar healthy and nothing ever revisits it."""
import multiprocessing
bailouts = {
"offline": lambda tv: monkeypatch.setattr(tv, "_env_offline", lambda: True),
"child": lambda tv: monkeypatch.setattr(
multiprocessing, "parent_process", lambda: object()
),
"swap_running": lambda tv: monkeypatch.setattr(
tv, "try_begin_sidecar_swap", lambda kind = None: False
),
}
for name, arrange in bailouts.items():
live = self._sidecar(tmp_path / f"venv_t5_latest_{name}")
tv, _ = self._patch(monkeypatch, live)
arrange(tv)
self._damage(live)
assert tv._ensure_venv_t5_latest_exists() is False, name
assert tv._latest_repair_requested(), f"{name} exited without recording the damage"
assert not tv._latest_sidecar_intact(), name
def test_cached_probe_result_does_not_survive_a_flagged_sidecar(self, monkeypatch, tmp_path):
"""_probe_tier caches per process. A result cached while the sidecar was healthy
must not keep sending workers into it once damage is known, the way the
config-mapping cache already refuses to."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
monkeypatch.setattr(tv, "_install_to_dir", lambda pkg, target: False)
key = f"{tv._probe_cache_key('some/model')}\0floor=default:def=1"
monkeypatch.setitem(tv._probe_tier_cache, key, "latest")
self._damage(live)
tv._request_latest_repair()
assert (
tv._probe_tier("some/model", None, "test", include_default = True, floor = "default")
!= "latest"
)
def test_file_check_kill_switch_suppresses_the_whole_handoff(self, monkeypatch, tmp_path):
"""UNSLOTH_SKIP_SIDECAR_FILE_CHECK is the escape hatch for a false positive; it
must leave no path that still wipes the sidecar."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, installs = self._patch(monkeypatch, live)
monkeypatch.setenv("UNSLOTH_SKIP_SIDECAR_FILE_CHECK", "1")
self._damage(live)
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
assert installs == []
monkeypatch.setattr("multiprocessing.parent_process", lambda: object())
assert tv._ensure_venv_t5_latest_exists() is True
assert not tv._latest_repair_requested()
def test_kill_switch_frees_the_sidecar_mid_backoff(self, monkeypatch, tmp_path):
"""The hatch exists for a false positive, so it has to work at the moment one is
being hit: with a marker already written and a repair already failed, it must
restore routing now rather than one backoff window later."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
monkeypatch.setattr(tv, "_install_to_dir", lambda pkg, target: False)
self._damage(live)
assert tv._overlay_transformers_dir("latest") is None # arms marker + backoff
assert tv._latest_repair_requested(), "precondition: the marker is armed"
monkeypatch.setenv("UNSLOTH_SKIP_SIDECAR_FILE_CHECK", "1")
assert tv._overlay_transformers_dir("latest") is not None
assert tv._tier_from_config_mapping({"model_type": "brandnew"}) == "latest"
def test_an_unreadable_file_is_not_reported_as_damage(self, monkeypatch, tmp_path):
"""A wipe costs a several-hundred-MB re-download, so the scan may only condemn
what it can prove: EIO or ESTALE on a flaky mount says the file could not be
read, never that it is gone."""
import errno
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
real_stat = Path.stat
def _flaky(self, *a, **k):
if self.name == "__init__.py":
raise OSError(errno.EIO, "Input/output error")
return real_stat(self, *a, **k)
monkeypatch.setattr(Path, "stat", _flaky)
assert tv._sidecar_damaged_files(str(live)) == []
assert tv._latest_sidecar_undamaged() is True
assert tv._overlay_transformers_dir("latest") is not None
def test_a_deleted_file_is_still_reported_as_damage(self, monkeypatch, tmp_path):
"""The counterpart: narrowing the caught errors must not blind the scan to a
genuinely missing file."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
(live / "transformers" / "__init__.py").unlink()
assert tv._sidecar_damaged_files(str(live)) == [
"transformers: transformers/__init__.py is missing"
]
def test_an_inconclusive_scan_does_not_retire_proven_damage(self, monkeypatch, tmp_path):
"""Skipping unreadable rows means an empty result can mean 'nothing proven', not
'proven healthy'. A worker already proved this sidecar damaged, and a scan that
could not read the file must not erase that and hand it back to routing."""
import errno
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
# A repair that cannot complete, so what is observed is the decision itself and
# not a rebuild papering over it.
monkeypatch.setattr(tv, "_install_to_dir", lambda pkg, target: False)
self._damage(live)
tv._request_latest_repair()
real_stat = Path.stat
def _flaky(self, *a, **k):
if self.name == "__init__.py":
raise OSError(errno.EIO, "Input/output error")
return real_stat(self, *a, **k)
monkeypatch.setattr(Path, "stat", _flaky)
assert tv._sidecar_damaged_files(str(live)) == [] # the scan sees nothing
assert tv._sidecar_scan(str(live)) == ([], True) # but knows it is blind
assert tv._ensure_venv_t5_latest_exists() is False
assert tv._latest_repair_requested(), "an unreadable scan cannot disprove damage"
def test_a_clean_scan_still_retires_a_stale_marker(self, monkeypatch, tmp_path):
"""The counterpart: a scan that really did read every row must still clear the
marker, or a request left by a lost race loops forever."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
tv._request_latest_repair()
assert tv._ensure_venv_t5_latest_exists() is True
assert not tv._latest_repair_requested()
def test_a_cached_probe_does_not_survive_the_pin_being_deleted(self, monkeypatch, tmp_path):
"""Deleting the pin takes the repair marker with it, so the marker cannot carry
this case. Activation short-circuits on a missing pin before it can signal
anything, so a cached latest would fail every retry until the process restarts."""
live = self._sidecar(tmp_path / "venv_t5_latest")
tv, _ = self._patch(monkeypatch, live)
import shutil
key = f"{tv._probe_cache_key('some/model')}\0floor=default:def=1"
monkeypatch.setitem(tv._probe_tier_cache, key, "latest")
shutil.rmtree(live)
assert (
tv._probe_tier("some/model", None, "test", include_default = True, floor = "default")
!= "latest"
)
Reference in a new issue View git blame Copy permalink