Add tests for cache case resolution (from PR #4822) (#4823)

Tests for resolve_cached_repo_id_case and get_model_config case
resolution, separated from the runtime changes in PR #4822.
This commit is contained in:
Daniel Han 2026-04-03 13:58:26 -07:00 committed by GitHub
commit 4020a70a93
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 201 additions and 0 deletions

View file

@ -0,0 +1,120 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
from pathlib import Path
import sys
import types
# Keep this test runnable in lightweight environments where optional logging
# deps are not installed.
if "structlog" not in sys.modules:
class _DummyLogger:
def __getattr__(self, _name):
return lambda *args, **kwargs: None
sys.modules["structlog"] = types.SimpleNamespace(
BoundLogger = _DummyLogger,
get_logger = lambda *args, **kwargs: _DummyLogger(),
)
from utils.paths.path_utils import (
resolve_cached_repo_id_case,
get_cache_case_resolution_stats,
reset_cache_case_resolution_state,
)
import utils.paths.path_utils as path_utils
def _mk_cache_repo(cache_root: Path, repo_id: str) -> Path:
repo_dir = cache_root / f"models--{repo_id.replace('/', '--')}"
repo_dir.mkdir(parents = True, exist_ok = True)
return repo_dir
def test_resolve_cached_repo_id_case_exact_hit(tmp_path, monkeypatch):
reset_cache_case_resolution_state()
_mk_cache_repo(tmp_path, "Org/Model")
monkeypatch.setattr(path_utils, "_hf_hub_cache_dir", lambda: tmp_path)
resolved = resolve_cached_repo_id_case("Org/Model")
assert resolved == "Org/Model"
stats = get_cache_case_resolution_stats()
assert stats["calls"] == 1
assert stats["exact_hits"] == 1
assert stats["variant_hits"] == 0
def test_resolve_cached_repo_id_case_variant_hit(tmp_path, monkeypatch):
reset_cache_case_resolution_state()
_mk_cache_repo(tmp_path, "Org/Model")
monkeypatch.setattr(path_utils, "_hf_hub_cache_dir", lambda: tmp_path)
resolved = resolve_cached_repo_id_case("org/model")
assert resolved == "Org/Model"
stats = get_cache_case_resolution_stats()
assert stats["variant_hits"] == 1
assert stats["tie_breaks"] == 0
def test_resolve_cached_repo_id_case_tie_break_deterministic(tmp_path, monkeypatch):
reset_cache_case_resolution_state()
_mk_cache_repo(tmp_path, "Org/Model")
_mk_cache_repo(tmp_path, "org/model")
monkeypatch.setattr(path_utils, "_hf_hub_cache_dir", lambda: tmp_path)
resolved = resolve_cached_repo_id_case("oRg/mOdEl")
# Deterministic rule: lexical sort of candidate repo ids.
assert resolved == "Org/Model"
stats = get_cache_case_resolution_stats()
assert stats["variant_hits"] == 1
assert stats["tie_breaks"] == 1
def test_resolve_cached_repo_id_case_no_cache_fallback(tmp_path, monkeypatch):
reset_cache_case_resolution_state()
monkeypatch.setattr(path_utils, "_hf_hub_cache_dir", lambda: tmp_path)
resolved = resolve_cached_repo_id_case("Org/Missing")
assert resolved == "Org/Missing"
stats = get_cache_case_resolution_stats()
assert stats["fallbacks"] == 1
assert stats["variant_hits"] == 0
assert stats["exact_hits"] == 0
def test_resolve_cached_repo_id_case_memoization(tmp_path, monkeypatch):
reset_cache_case_resolution_state()
_mk_cache_repo(tmp_path, "Org/Model")
monkeypatch.setattr(path_utils, "_hf_hub_cache_dir", lambda: tmp_path)
first = resolve_cached_repo_id_case("org/model")
second = resolve_cached_repo_id_case("org/model")
assert first == "Org/Model"
assert second == "Org/Model"
stats = get_cache_case_resolution_stats()
assert stats["calls"] == 2
assert stats["variant_hits"] == 1
assert stats["memo_hits"] == 1
def test_resolve_cached_repo_id_case_late_cache_population(tmp_path, monkeypatch):
"""Regression guard: memoized fallback should not hide a later cache variant."""
reset_cache_case_resolution_state()
monkeypatch.setattr(path_utils, "_hf_hub_cache_dir", lambda: tmp_path)
first = resolve_cached_repo_id_case("org/model")
assert first == "org/model"
# Simulate cache being populated after first miss (e.g. another code path/download).
_mk_cache_repo(tmp_path, "Org/Model")
second = resolve_cached_repo_id_case("org/model")
# Desired behavior: second lookup should pick up the now-existing variant.
assert second == "Org/Model"

View file

@ -0,0 +1,81 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
import asyncio
import sys
import types
# Keep this test runnable in lightweight environments where optional logging
# deps are not installed.
if "structlog" not in sys.modules:
class _DummyLogger:
def __getattr__(self, _name):
return lambda *args, **kwargs: None
sys.modules["structlog"] = types.SimpleNamespace(
BoundLogger = _DummyLogger,
get_logger = lambda *args, **kwargs: _DummyLogger(),
)
import routes.models as models_route
import utils.models.model_config as model_config_module
def test_get_model_config_resolves_cached_case_before_model_checks(monkeypatch):
calls: dict[str, str] = {}
class _DummyModelConfig:
is_lora = False
base_model = None
def _record_load(model_name):
calls["load_model_defaults"] = model_name
return {}
def _record_vision(model_name, hf_token = None):
calls["is_vision_model"] = model_name
return False
def _record_embedding(model_name, hf_token = None):
calls["is_embedding_model"] = model_name
return False
def _record_audio(model_name, hf_token = None):
calls["detect_audio_type"] = model_name
return None
def _record_from_identifier(cls, model_name):
calls["from_identifier"] = model_name
return _DummyModelConfig()
monkeypatch.setattr(models_route, "is_local_path", lambda _: False)
monkeypatch.setattr(
models_route, "resolve_cached_repo_id_case", lambda _: "Org/Model"
)
monkeypatch.setattr(models_route, "load_model_defaults", _record_load)
monkeypatch.setattr(models_route, "is_vision_model", _record_vision)
monkeypatch.setattr(models_route, "is_embedding_model", _record_embedding)
monkeypatch.setattr(model_config_module, "detect_audio_type", _record_audio)
monkeypatch.setattr(
models_route.ModelConfig,
"from_identifier",
classmethod(_record_from_identifier),
)
monkeypatch.setattr(models_route, "_get_max_position_embeddings", lambda _: 4096)
monkeypatch.setattr(models_route, "_get_model_size_bytes", lambda *_args, **_kw: 0)
result = asyncio.run(
models_route.get_model_config(
model_name = "org/model",
hf_token = None,
current_subject = "test-subject",
)
)
assert result.model_name == "Org/Model"
assert calls["load_model_defaults"] == "Org/Model"
assert calls["is_vision_model"] == "Org/Model"
assert calls["is_embedding_model"] == "Org/Model"
assert calls["detect_audio_type"] == "Org/Model"
assert calls["from_identifier"] == "Org/Model"