ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot] 2026-06-13 03:13:42 +00:00
commit d6f1075812
460 changed files with 13446 additions and 4512 deletions
Download patch file Download diff file Expand all files Collapse all files

8
tests/python/conftest.py
View file

@ -2,5 +2,9 @@
def pytest_configure(config):
config.addinivalue_line("markers", "server: heavyweight tests requiring studio venv")
config.addinivalue_line("markers", "e2e: end-to-end tests requiring network and venv creation")
config.addinivalue_line(
"markers", "server: heavyweight tests requiring studio venv"
)
config.addinivalue_line(
"markers", "e2e: end-to-end tests requiring network and venv creation"
)

10
tests/python/test_cross_platform_parity.py
View file

@ -27,11 +27,17 @@ class TestNoTorchBackendAutoInInstallSh:
for i, line in enumerate(lines):
if fallback_start is None and "GPU detection failed" in line:
fallback_start = i
elif fallback_start is not None and fallback_end is None and line.strip() == "fi":
elif (
fallback_start is not None
and fallback_end is None
and line.strip() == "fi"
):
fallback_end = i
break
fallback_range = (
range(fallback_start or 0, (fallback_end or 0) + 1) if fallback_start else range(0)
range(fallback_start or 0, (fallback_end or 0) + 1)
if fallback_start
else range(0)
)
matches = [

60
tests/python/test_e2e_no_torch_sandbox.py
View file

@ -235,8 +235,12 @@ class TestBeforeAfterImportChain:
exec(source)
""")
result = _run_in_sandbox(no_torch_venv, code)
assert result.returncode != 0, "BEFORE chat_templates.py should crash without torch"
assert b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
assert (
result.returncode != 0
), "BEFORE chat_templates.py should crash without torch"
assert (
b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
)
def test_before_data_collators_crashes(self, no_torch_venv, sandbox_dir):
"""BEFORE: data_collators.py with top-level 'import torch' crashes."""
@ -254,8 +258,12 @@ class TestBeforeAfterImportChain:
exec(open({str(before_file)!r}).read())
""")
result = _run_in_sandbox(no_torch_venv, code)
assert result.returncode != 0, "BEFORE data_collators.py should crash without torch"
assert b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
assert (
result.returncode != 0
), "BEFORE data_collators.py should crash without torch"
assert (
b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
)
def test_before_full_import_chain_crashes(self, no_torch_venv, sandbox_dir):
"""BEFORE: full utils/datasets/ package with top-level torch imports crashes."""
@ -300,8 +308,12 @@ class TestBeforeAfterImportChain:
from utils.datasets import detect_dataset_format
""")
result = _run_in_sandbox(no_torch_venv, code)
assert result.returncode != 0, "BEFORE full import chain should crash without torch"
assert b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
assert (
result.returncode != 0
), "BEFORE full import chain should crash without torch"
assert (
b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
)
# -- AFTER: succeeds --
@ -515,7 +527,9 @@ class TestEdgeCasesBrokenTorch:
print("OK: data_collators works despite broken torch on sys.path")
""")
result = _run_in_sandbox(no_torch_venv, code)
assert result.returncode == 0, f"Should work with broken torch:\n{result.stderr.decode()}"
assert (
result.returncode == 0
), f"Should work with broken torch:\n{result.stderr.decode()}"
assert b"OK:" in result.stdout
def test_torch_import_error_hardware_fallback(self, no_torch_venv, sandbox_dir):
@ -578,10 +592,14 @@ class TestEdgeCasesBrokenTorch:
print("OK: detect_hardware returned CPU with fake torch (no CUDA)")
""")
result = _run_in_sandbox(no_torch_venv, code)
assert result.returncode == 0, f"Should fall back to CPU:\n{result.stderr.decode()}"
assert (
result.returncode == 0
), f"Should fall back to CPU:\n{result.stderr.decode()}"
assert b"OK:" in result.stdout
def test_lazy_torch_fails_at_call_time_not_import_time(self, no_torch_venv, sandbox_dir):
def test_lazy_torch_fails_at_call_time_not_import_time(
self, no_torch_venv, sandbox_dir
):
"""apply_chat_template_to_dataset is importable without torch.
Calling the alpaca branch triggers the lazy 'from torch.utils.data' inside
@ -627,7 +645,9 @@ class TestEdgeCasesBrokenTorch:
print("OK: call succeeded (unexpected but not a crash)")
""")
result = _run_in_sandbox(no_torch_venv, code)
assert result.returncode == 0, f"Should not crash at import time:\n{result.stderr.decode()}"
assert (
result.returncode == 0
), f"Should not crash at import time:\n{result.stderr.decode()}"
assert b"OK: import succeeded" in result.stdout
@ -979,7 +999,9 @@ class TestInstallPythonStackFiltering:
source = Path(ips.__file__).read_text(encoding = "utf-8")
# NO_TORCH guard before overrides
assert "if NO_TORCH:" in source, "NO_TORCH guard not found in install_python_stack.py"
assert (
"if NO_TORCH:" in source
), "NO_TORCH guard not found in install_python_stack.py"
# macOS guard for triton
assert (
@ -1082,7 +1104,9 @@ class TestLiveServerStartup:
for _ in range(30):
time.sleep(1)
try:
resp = urllib.request.urlopen(f"http://127.0.0.1:{port}/api/health", timeout = 2)
resp = urllib.request.urlopen(
f"http://127.0.0.1:{port}/api/health", timeout = 2
)
if resp.status == 200:
ready = True
break
@ -1106,8 +1130,12 @@ class TestLiveServerStartup:
capture_output = True,
timeout = 300,
)
server_output = stdout.decode(errors = "replace") + stderr.decode(errors = "replace")
pytest.skip(f"Server failed to start within 30 seconds. Output:\n{server_output}")
server_output = stdout.decode(errors = "replace") + stderr.decode(
errors = "replace"
)
pytest.skip(
f"Server failed to start within 30 seconds. Output:\n{server_output}"
)
yield proc, port
@ -1151,7 +1179,9 @@ class TestLiveServerStartup:
import urllib.request
_, port = server_process
resp = urllib.request.urlopen(f"http://127.0.0.1:{port}/openapi.json", timeout = 5)
resp = urllib.request.urlopen(
f"http://127.0.0.1:{port}/openapi.json", timeout = 5
)
spec = json.loads(resp.read())
assert (
len(spec.get("paths", {})) >= 20

19
tests/python/test_fast_language_model_text_only.py
View file

@ -126,7 +126,9 @@ def test_fast_language_model_forwards_text_only_to_fast_model():
# text_only defaults False (opt-in, not forced True), and both FastModel
# delegations forward it.
text_only_default = _param_default(method, "text_only")
assert isinstance(text_only_default, ast.Constant) and text_only_default.value is False
assert (
isinstance(text_only_default, ast.Constant) and text_only_default.value is False
)
fast_model_calls = [
node
@ -150,13 +152,16 @@ def test_fast_model_text_only_does_not_override_explicit_auto_model():
method = _class_method(ast.parse(source), "FastModel", "from_pretrained")
text_only_default = _param_default(method, "text_only")
assert isinstance(text_only_default, ast.Constant) and text_only_default.value is False
assert (
isinstance(text_only_default, ast.Constant) and text_only_default.value is False
)
# load_text_only is text_only AND a check that the caller did not pass auto_model.
def _is_guarded_bool(value):
names = _names_in(value)
has_none_check = any(
isinstance(n, ast.Compare) and any(isinstance(op, (ast.Is, ast.IsNot)) for op in n.ops)
isinstance(n, ast.Compare)
and any(isinstance(op, (ast.Is, ast.IsNot)) for op in n.ops)
for n in ast.walk(value)
)
return "text_only" in names and "auto_model" in names and has_none_check
@ -192,7 +197,9 @@ def test_fast_base_model_text_only_bypasses_vision_auto_model():
method = _class_method(ast.parse(source), "FastBaseModel", "from_pretrained")
text_only_default = _param_default(method, "text_only")
assert isinstance(text_only_default, ast.Constant) and text_only_default.value is False
assert (
isinstance(text_only_default, ast.Constant) and text_only_default.value is False
)
assert _assigns_name(
method,
@ -327,7 +334,9 @@ def test_text_only_key_mapping_targets_published_prefixes():
# transformers >=5 (on 4.x base_model_prefix handles it and a mapping hurts).
transformers = pytest.importorskip("transformers")
get_key_mapping = _load_util_func("_get_text_only_key_mapping")
mapping = get_key_mapping(transformers.Gemma3Config(), transformers.Gemma3TextConfig())
mapping = get_key_mapping(
transformers.Gemma3Config(), transformers.Gemma3TextConfig()
)
if int(transformers.__version__.split(".")[0]) < 5:
assert mapping is None
else:

11
tests/python/test_fast_model_config_passthrough.py
View file

@ -29,7 +29,8 @@ def _assigns_from_kwargs_pop(method, target_name, key_name):
if not isinstance(node, ast.Assign):
continue
if not any(
isinstance(target, ast.Name) and target.id == target_name for target in node.targets
isinstance(target, ast.Name) and target.id == target_name
for target in node.targets
):
continue
value = node.value
@ -50,7 +51,9 @@ def _assigns_from_kwargs_pop(method, target_name, key_name):
def _calls_name(method, name):
return any(
isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == name
isinstance(node, ast.Call)
and isinstance(node.func, ast.Name)
and node.func.id == name
for node in ast.walk(method)
)
@ -130,7 +133,9 @@ def test_fast_model_uses_user_config_num_labels_for_task_model_selection():
def test_fast_model_captures_user_config_num_labels_before_text_only_switch():
source = _source(LOADER_PATH)
fallback = source.index("task_config_attrs = _get_user_task_config_attrs(user_config)")
fallback = source.index(
"task_config_attrs = _get_user_task_config_attrs(user_config)"
)
text_only_switch = source.index("model_config = text_config")
assert fallback < text_only_switch

12
tests/python/test_fast_sentence_transformer_redirect_lifecycle.py
View file

@ -127,10 +127,18 @@ def _build_driver(transformer_class):
return model if is_requested_model_name(a, kw) else original_model(*a, **kw)
def return_existing_tokenizer(*a, **kw):
return tokenizer if is_requested_model_name(a, kw) else original_tokenizer(*a, **kw)
return (
tokenizer
if is_requested_model_name(a, kw)
else original_tokenizer(*a, **kw)
)
def return_existing_processor(*a, **kw):
return tokenizer if is_requested_model_name(a, kw) else original_processor(*a, **kw)
return (
tokenizer
if is_requested_model_name(a, kw)
else original_processor(*a, **kw)
)
try:
AutoModel.from_pretrained = return_existing_model

78
tests/python/test_flash_attn_install_python_stack.py
View file

@ -33,42 +33,64 @@ class TestHasBlackwellGpu:
def test_returns_true_for_sm_100(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(wheel_utils.subprocess, "run", return_value = _smi_result("10.0\n")),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess, "run", return_value = _smi_result("10.0\n")
),
):
assert wheel_utils.has_blackwell_gpu() is True
def test_returns_true_for_sm_120(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(wheel_utils.subprocess, "run", return_value = _smi_result("12.0\n")),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess, "run", return_value = _smi_result("12.0\n")
),
):
assert wheel_utils.has_blackwell_gpu() is True
def test_returns_true_for_sm_121(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(wheel_utils.subprocess, "run", return_value = _smi_result("12.1\n")),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess, "run", return_value = _smi_result("12.1\n")
),
):
assert wheel_utils.has_blackwell_gpu() is True
def test_returns_false_for_sm_90(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(wheel_utils.subprocess, "run", return_value = _smi_result("9.0\n")),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess, "run", return_value = _smi_result("9.0\n")
),
):
assert wheel_utils.has_blackwell_gpu() is False
def test_returns_false_for_sm_89(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(wheel_utils.subprocess, "run", return_value = _smi_result("8.9\n")),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess, "run", return_value = _smi_result("8.9\n")
),
):
assert wheel_utils.has_blackwell_gpu() is False
def test_mixed_gpus_with_one_blackwell_returns_true(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess,
"run",
@ -79,7 +101,9 @@ class TestHasBlackwellGpu:
def test_returns_false_when_nvidia_smi_fails(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess,
"run",
@ -90,7 +114,9 @@ class TestHasBlackwellGpu:
def test_returns_false_on_subprocess_timeout(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess,
"run",
@ -101,7 +127,9 @@ class TestHasBlackwellGpu:
def test_returns_false_on_malformed_output(self):
with (
mock.patch.object(wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"),
mock.patch.object(
wheel_utils.shutil, "which", return_value = "/usr/bin/nvidia-smi"
),
mock.patch.object(
wheel_utils.subprocess,
"run",
@ -133,7 +161,10 @@ class TestFlashAttnWheelSelection:
)
assert url is not None
assert "v2.8.1" in url
assert "flash_attn-2.8.1+cu12torch2.10cxx11abiTRUE-cp313-cp313-linux_x86_64.whl" in url
assert (
"flash_attn-2.8.1+cu12torch2.10cxx11abiTRUE-cp313-cp313-linux_x86_64.whl"
in url
)
def test_missing_cuda_major_disables_wheel_lookup(self):
assert (
@ -316,7 +347,10 @@ class TestEnsureFlashAttn:
ips._ensure_flash_attn()
mock_install_wheel.assert_not_called()
assert ("warning", "No published flash-attn prebuilt wheel found") in step_messages
assert (
"warning",
"No published flash-attn prebuilt wheel found",
) in step_messages
def test_skip_env_disables_setup_install(self):
with (
@ -357,7 +391,9 @@ class TestEnsureFlashAttn:
mock_probe.assert_not_called()
mock_install_wheel.assert_not_called()
assert any(label == "warning" and "Blackwell" in msg for label, msg in step_messages)
assert any(
label == "warning" and "Blackwell" in msg for label, msg in step_messages
)
def test_blackwell_gpu_on_windows_emits_blackwell_warning(self):
step_messages: list[tuple[str, str]] = []
@ -383,7 +419,9 @@ class TestEnsureFlashAttn:
mock_probe.assert_not_called()
mock_install_wheel.assert_not_called()
assert any(label == "warning" and "Blackwell" in msg for label, msg in step_messages)
assert any(
label == "warning" and "Blackwell" in msg for label, msg in step_messages
)
def test_non_blackwell_windows_does_not_emit_blackwell_warning(self):
step_messages: list[tuple[str, str]] = []
@ -435,7 +473,9 @@ class TestInstallPythonStackFlashAttnIntegration:
mock.patch("subprocess.run", side_effect = fake_run),
mock.patch.object(ips, "_has_usable_nvidia_gpu", return_value = False),
mock.patch.object(ips, "_has_rocm_gpu", return_value = False),
mock.patch.object(ips, "LOCAL_DD_UNSTRUCTURED_PLUGIN", Path("/fake/plugin")),
mock.patch.object(
ips, "LOCAL_DD_UNSTRUCTURED_PLUGIN", Path("/fake/plugin")
),
mock.patch("pathlib.Path.is_dir", return_value = True),
mock.patch("pathlib.Path.is_file", return_value = True),
mock.patch.dict(os.environ, {"SKIP_STUDIO_BASE": "1"}, clear = False),

4
tests/python/test_gpu_init_ldconfig_guard.py
View file

@ -19,7 +19,9 @@ def _find_geteuid_guard(tree: ast.AST):
def test_gpu_init_has_geteuid_guard():
tree = ast.parse(GPU_INIT.read_text())
guard = _find_geteuid_guard(tree)
assert guard is not None, "_gpu_init.py must guard ldconfig recovery on os.geteuid()"
assert (
guard is not None
), "_gpu_init.py must guard ldconfig recovery on os.geteuid()"
def test_ldconfig_calls_only_inside_geteuid_guard():

84
tests/python/test_no_torch_filtering.py
View file

@ -156,7 +156,9 @@ class TestFilterRequirements:
)
# First filter Windows packages, then NO_TORCH packages
intermediate = ips._filter_requirements(req, ips.WINDOWS_SKIP_PACKAGES)
result = ips._filter_requirements(Path(intermediate), ips.NO_TORCH_SKIP_PACKAGES)
result = ips._filter_requirements(
Path(intermediate), ips.NO_TORCH_SKIP_PACKAGES
)
lines = Path(result).read_text(encoding = "utf-8").splitlines()
non_blank = [l.strip() for l in lines if l.strip()]
assert non_blank == [
@ -175,7 +177,9 @@ class TestFilterRequirements:
result = ips._filter_requirements(req, ips.NO_TORCH_SKIP_PACKAGES)
lines = Path(result).read_text(encoding = "utf-8").splitlines()
non_blank = [l.strip() for l in lines if l.strip()]
assert non_blank == ["numpy"], f"VCS URL line should be filtered, got: {non_blank}"
assert non_blank == [
"numpy"
], f"VCS URL line should be filtered, got: {non_blank}"
def test_env_marker_line_filtered(self, tmp_path):
"""Package lines with env markers are still filtered by prefix."""
@ -189,7 +193,9 @@ class TestFilterRequirements:
result = ips._filter_requirements(req, ips.NO_TORCH_SKIP_PACKAGES)
lines = Path(result).read_text(encoding = "utf-8").splitlines()
non_blank = [l.strip() for l in lines if l.strip()]
assert non_blank == ["numpy"], f"Env marker line should be filtered, got: {non_blank}"
assert non_blank == [
"numpy"
], f"Env marker line should be filtered, got: {non_blank}"
def test_git_plus_url_not_over_matched(self, tmp_path):
"""A git+ URL whose path contains a skip package name but does NOT start with it."""
@ -241,7 +247,9 @@ class TestRealRequirementsFiltering:
expected = [
l
for l in original
if not any(l.strip().lower().startswith(p) for p in ips.NO_TORCH_SKIP_PACKAGES)
if not any(
l.strip().lower().startswith(p) for p in ips.NO_TORCH_SKIP_PACKAGES
)
]
assert filtered == expected, (
f"Filtered extras.txt should match expected.\n"
@ -251,7 +259,9 @@ class TestRealRequirementsFiltering:
def test_extras_no_deps_txt_torchcodec_and_dlpack_removed(self):
"""extras-no-deps.txt: torchcodec and torch-c-dlpack-ext must be removed."""
result = ips._filter_requirements(EXTRAS_NO_DEPS_TXT, ips.NO_TORCH_SKIP_PACKAGES)
result = ips._filter_requirements(
EXTRAS_NO_DEPS_TXT, ips.NO_TORCH_SKIP_PACKAGES
)
filtered = self._non_blank_non_comment(Path(result))
original = self._non_blank_non_comment(EXTRAS_NO_DEPS_TXT)
@ -263,7 +273,9 @@ class TestRealRequirementsFiltering:
expected = [
l
for l in original
if not any(l.strip().lower().startswith(p) for p in ips.NO_TORCH_SKIP_PACKAGES)
if not any(
l.strip().lower().startswith(p) for p in ips.NO_TORCH_SKIP_PACKAGES
)
]
assert filtered == expected
@ -279,7 +291,9 @@ class TestRealRequirementsFiltering:
def test_extras_no_deps_txt_trl_preserved(self):
"""trl should survive NO_TORCH filtering in extras-no-deps.txt."""
result = ips._filter_requirements(EXTRAS_NO_DEPS_TXT, ips.NO_TORCH_SKIP_PACKAGES)
result = ips._filter_requirements(
EXTRAS_NO_DEPS_TXT, ips.NO_TORCH_SKIP_PACKAGES
)
filtered_text = Path(result).read_text(encoding = "utf-8").lower()
assert "trl" in filtered_text, "trl should survive NO_TORCH filtering"
@ -390,7 +404,9 @@ class TestInstallPythonStackSubprocessMock:
captured_cmds: list[list[str]] = []
def mock_run(cmd, **kw):
captured_cmds.append(list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)])
captured_cmds.append(
list(cmd) if isinstance(cmd, (list, tuple)) else [str(cmd)]
)
return subprocess.CompletedProcess(cmd, 0, b"", b"")
env = {"SKIP_STUDIO_BASE": "1"} if skip_base else {}
@ -407,7 +423,9 @@ class TestInstallPythonStackSubprocessMock:
mock.patch.object(ips, "_has_rocm_gpu", return_value = False),
mock.patch("subprocess.run", side_effect = mock_run),
mock.patch.object(ips, "_bootstrap_uv", return_value = True),
mock.patch.object(ips, "LOCAL_DD_UNSTRUCTURED_PLUGIN", Path("/fake/plugin")),
mock.patch.object(
ips, "LOCAL_DD_UNSTRUCTURED_PLUGIN", Path("/fake/plugin")
),
mock.patch("pathlib.Path.is_dir", return_value = True),
mock.patch("pathlib.Path.is_file", return_value = True),
):
@ -450,7 +468,9 @@ class TestInstallPythonStackSubprocessMock:
has_extras_nd = self._cmds_contain_file(cmds, "extras-no-deps.txt") or any(
"-r" in cmd and "tmp" in cmd.lower() for cmd in cmds
)
assert has_extras_nd, "extras-no-deps.txt (or its filtered temp) should be called"
assert (
has_extras_nd
), "extras-no-deps.txt (or its filtered temp) should be called"
# -- IS_WINDOWS=True + NO_TORCH=True (stacked) --
@ -549,13 +569,17 @@ class TestOverridesSkip:
def test_no_torch_guard_exists_in_source(self):
"""The install_python_stack source must contain a NO_TORCH guard around overrides."""
source = Path(ips.__file__).read_text(encoding = "utf-8")
assert "if NO_TORCH:" in source, "NO_TORCH guard not found in install_python_stack.py"
assert (
"if NO_TORCH:" in source
), "NO_TORCH guard not found in install_python_stack.py"
def test_overrides_skipped_when_no_torch(self):
"""With NO_TORCH=True on the module, pip_install should NOT be called for overrides."""
source = Path(ips.__file__).read_text(encoding = "utf-8")
overrides_match = re.search(r"if NO_TORCH:.*?overrides", source, re.DOTALL)
assert overrides_match is not None, "Expected NO_TORCH conditional before overrides install"
assert (
overrides_match is not None
), "Expected NO_TORCH conditional before overrides install"
# ── install.sh --no-torch flag tests ──────────────────────────────────
@ -574,21 +598,33 @@ class TestInstallShNoTorchFlag:
def test_no_torch_flag_in_case_statement(self):
"""--no-torch must appear in the flag parser case statement."""
assert "--no-torch)" in self.source, "--no-torch not found in install.sh flag parser"
assert (
"--no-torch)" in self.source
), "--no-torch not found in install.sh flag parser"
def test_no_torch_flag_variable_initialized(self):
"""_NO_TORCH_FLAG must be initialized to false."""
assert "_NO_TORCH_FLAG=false" in self.source, "_NO_TORCH_FLAG=false not found in install.sh"
assert (
"_NO_TORCH_FLAG=false" in self.source
), "_NO_TORCH_FLAG=false not found in install.sh"
def test_skip_torch_variable_exists(self):
"""SKIP_TORCH variable must be defined."""
assert "SKIP_TORCH=false" in self.source, "SKIP_TORCH=false not found in install.sh"
assert "SKIP_TORCH=true" in self.source, "SKIP_TORCH=true not found in install.sh"
assert (
"SKIP_TORCH=false" in self.source
), "SKIP_TORCH=false not found in install.sh"
assert (
"SKIP_TORCH=true" in self.source
), "SKIP_TORCH=true not found in install.sh"
def test_skip_torch_driven_by_flag_and_mac_intel(self):
"""SKIP_TORCH must check both _NO_TORCH_FLAG and MAC_INTEL."""
assert "_NO_TORCH_FLAG" in self.source, "_NO_TORCH_FLAG not referenced in SKIP_TORCH logic"
assert "MAC_INTEL" in self.source, "MAC_INTEL not referenced in SKIP_TORCH logic"
assert (
"_NO_TORCH_FLAG" in self.source
), "_NO_TORCH_FLAG not referenced in SKIP_TORCH logic"
assert (
"MAC_INTEL" in self.source
), "MAC_INTEL not referenced in SKIP_TORCH logic"
def test_unsloth_no_torch_uses_skip_torch(self):
"""UNSLOTH_NO_TORCH must reference $SKIP_TORCH, not $MAC_INTEL."""
@ -596,12 +632,18 @@ class TestInstallShNoTorchFlag:
matches = re.findall(r'UNSLOTH_NO_TORCH="\$(\w+)"', self.source)
for var in matches:
assert var == "SKIP_TORCH", f"UNSLOTH_NO_TORCH references ${var} instead of $SKIP_TORCH"
assert (
var == "SKIP_TORCH"
), f"UNSLOTH_NO_TORCH references ${var} instead of $SKIP_TORCH"
def test_cpu_hint_message_exists(self):
"""CPU hint message must exist in install.sh."""
assert "No GPU detected" in self.source, "CPU hint message not found in install.sh"
assert "--no-torch" in self.source, "--no-torch suggestion not found in CPU hint"
assert (
"No GPU detected" in self.source
), "CPU hint message not found in install.sh"
assert (
"--no-torch" in self.source
), "--no-torch suggestion not found in CPU hint"
def test_no_torch_flag_parsing_subprocess(self):
"""--no-torch flag sets _NO_TORCH_FLAG=true (subprocess test)."""

35
tests/python/test_studio_import_no_torch.py
View file

@ -23,9 +23,15 @@ from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parents[2]
DATA_COLLATORS = REPO_ROOT / "studio" / "backend" / "utils" / "datasets" / "data_collators.py"
CHAT_TEMPLATES = REPO_ROOT / "studio" / "backend" / "utils" / "datasets" / "chat_templates.py"
FORMAT_CONVERSION = REPO_ROOT / "studio" / "backend" / "utils" / "datasets" / "format_conversion.py"
DATA_COLLATORS = (
REPO_ROOT / "studio" / "backend" / "utils" / "datasets" / "data_collators.py"
)
CHAT_TEMPLATES = (
REPO_ROOT / "studio" / "backend" / "utils" / "datasets" / "chat_templates.py"
)
FORMAT_CONVERSION = (
REPO_ROOT / "studio" / "backend" / "utils" / "datasets" / "format_conversion.py"
)
def _has_uv() -> bool:
@ -65,7 +71,9 @@ def no_torch_venv(request, tmp_path_factory):
[str(venv_python), "-c", "import torch"],
capture_output = True,
)
assert check.returncode != 0, f"torch should NOT be importable in fresh {py_version} venv"
assert (
check.returncode != 0
), f"torch should NOT be importable in fresh {py_version} venv"
return str(venv_python)
@ -214,7 +222,9 @@ class TestDataCollatorsNoTorchVenv:
capture_output = True,
timeout = 30,
)
assert result.returncode == 0, f"DeepSeekOCRDataCollator failed:\n{result.stderr.decode()}"
assert (
result.returncode == 0
), f"DeepSeekOCRDataCollator failed:\n{result.stderr.decode()}"
assert b"OK: DeepSeekOCRDataCollator instantiated" in result.stdout
def test_dataclass_vlm_collator_instantiable(self, no_torch_venv):
@ -235,7 +245,9 @@ class TestDataCollatorsNoTorchVenv:
capture_output = True,
timeout = 30,
)
assert result.returncode == 0, f"VLMDataCollator failed:\n{result.stderr.decode()}"
assert (
result.returncode == 0
), f"VLMDataCollator failed:\n{result.stderr.decode()}"
assert b"OK: VLMDataCollator instantiated" in result.stdout
@ -516,9 +528,12 @@ class TestNegativeControls:
capture_output = True,
timeout = 30,
)
assert result.returncode != 0, "Expected failure when 'import torch' is prepended"
assert (
b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
result.returncode != 0
), "Expected failure when 'import torch' is prepended"
assert (
b"ModuleNotFoundError" in result.stderr
or b"ImportError" in result.stderr
), f"Expected ImportError, got:\n{result.stderr.decode()}"
finally:
os.unlink(temp_file)
@ -561,4 +576,6 @@ class TestNegativeControls:
timeout = 30,
)
assert result.returncode != 0, "import torch should fail in no-torch venv"
assert b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
assert (
b"ModuleNotFoundError" in result.stderr or b"ImportError" in result.stderr
)

40
tests/python/test_tokenizers_and_torch_constraint.py
View file

@ -17,7 +17,9 @@ _TESTS_DIR = pathlib.Path(__file__).resolve().parent.parent # tests/
_REPO_ROOT = _TESTS_DIR.parent # unsloth/
_INSTALL_SH = _REPO_ROOT / "install.sh"
_INSTALL_PS1 = _REPO_ROOT / "install.ps1"
_NO_TORCH_RT = _REPO_ROOT / "studio" / "backend" / "requirements" / "no-torch-runtime.txt"
_NO_TORCH_RT = (
_REPO_ROOT / "studio" / "backend" / "requirements" / "no-torch-runtime.txt"
)
def _read(path: pathlib.Path) -> str:
@ -42,23 +44,30 @@ class TestStructuralTokenizers:
def test_tokenizers_present(self):
"""tokenizers must be a standalone package line."""
pkgs = _lines(_NO_TORCH_RT)
bare_names = [p.split(">")[0].split("<")[0].split("!")[0].split("=")[0] for p in pkgs]
bare_names = [
p.split(">")[0].split("<")[0].split("!")[0].split("=")[0] for p in pkgs
]
assert "tokenizers" in bare_names
def test_tokenizers_before_transformers(self):
"""tokenizers should appear before transformers (install order intent)."""
pkgs = _lines(_NO_TORCH_RT)
bare_names = [p.split(">")[0].split("<")[0].split("!")[0].split("=")[0] for p in pkgs]
bare_names = [
p.split(">")[0].split("<")[0].split("!")[0].split("=")[0] for p in pkgs
]
idx_tok = bare_names.index("tokenizers")
idx_tf = bare_names.index("transformers")
assert idx_tok < idx_tf, (
f"tokenizers at index {idx_tok} should appear before " f"transformers at index {idx_tf}"
f"tokenizers at index {idx_tok} should appear before "
f"transformers at index {idx_tf}"
)
def test_torch_not_in_no_torch_file(self):
"""torch itself must NOT be listed in the no-torch requirements."""
pkgs = _lines(_NO_TORCH_RT)
bare_names = [p.split(">")[0].split("<")[0].split("!")[0].split("=")[0] for p in pkgs]
bare_names = [
p.split(">")[0].split("<")[0].split("!")[0].split("=")[0] for p in pkgs
]
assert "torch" not in bare_names
@ -399,7 +408,9 @@ class TestE2ETokenizersFix:
r = self._pip_install(venv, "--no-deps", "-r", str(_NO_TORCH_RT))
assert r.returncode == 0, f"Install failed: {r.stderr}"
result = self._run_python(venv, "from transformers import AutoConfig; print('OK')")
result = self._run_python(
venv, "from transformers import AutoConfig; print('OK')"
)
assert (
result.returncode == 0
), f"AutoConfig import failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"
@ -429,15 +440,22 @@ class TestE2ETokenizersFix:
req_no_tokenizers = tmp_path / "no-tokenizers.txt"
req_no_tokenizers.write_text(
"\n".join(
line for line in _read(_NO_TORCH_RT).splitlines() if line.strip() != "tokenizers"
line
for line in _read(_NO_TORCH_RT).splitlines()
if line.strip() != "tokenizers"
),
encoding = "utf-8",
)
r = self._pip_install(venv, "--no-deps", "-r", str(req_no_tokenizers))
assert r.returncode == 0, f"Install failed: {r.stderr}"
result = self._run_python(venv, "from transformers import AutoConfig")
assert result.returncode != 0, "AutoConfig should fail without tokenizers installed"
assert "tokenizers" in result.stderr.lower() or "ModuleNotFoundError" in result.stderr
assert (
result.returncode != 0
), "AutoConfig should fail without tokenizers installed"
assert (
"tokenizers" in result.stderr.lower()
or "ModuleNotFoundError" in result.stderr
)
# ======================================================================
@ -516,7 +534,9 @@ class TestE2EFullNoTorchSandbox:
venv = self._create_venv(tmp_path, "full-no-torch")
r = self._pip_install(venv, "--no-deps", "-r", str(_NO_TORCH_RT))
assert r.returncode == 0, f"Install failed: {r.stderr}"
result = self._run_python(venv, "from transformers import AutoConfig; print('OK')")
result = self._run_python(
venv, "from transformers import AutoConfig; print('OK')"
)
assert (
result.returncode == 0
), f"AutoConfig failed:\nstdout: {result.stdout}\nstderr: {result.stderr}"

8
tests/python/test_unsloth_run_tool_policy_resolver.py
View file

@ -141,11 +141,15 @@ class TestZeroHost:
class TestIsExternalHost:
@pytest.mark.parametrize("host", ["127.0.0.1", "localhost", "::1", "LOCALHOST", "Localhost"])
@pytest.mark.parametrize(
"host", ["127.0.0.1", "localhost", "::1", "LOCALHOST", "Localhost"]
)
def test_loopback_aliases_are_local(self, host):
assert is_external_host(host) is False
@pytest.mark.parametrize("host", ["0.0.0.0", "::", "192.168.1.5", "10.0.0.1", "example.com"])
@pytest.mark.parametrize(
"host", ["0.0.0.0", "::", "192.168.1.5", "10.0.0.1", "example.com"]
)
def test_non_loopback_is_external(self, host):
assert is_external_host(host) is True