187144d4e7Trim and tighten code comments and docstrings across the repository. Comment-only: every changed file verified code-identical to main via AST/token comparison.
305 lines
11 KiB
Python
305 lines
11 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""Tests for the llama.cpp prebuilt freshness check.
|
|
|
|
Pins the marker parser, disk+memory cache, stale-decision matrix, and
|
|
fail-open behaviour on missing data.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import types as _types
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
|
|
if _BACKEND_DIR not in sys.path:
|
|
sys.path.insert(0, _BACKEND_DIR)
|
|
|
|
_loggers_stub = _types.ModuleType("loggers")
|
|
_loggers_stub.get_logger = lambda name: __import__("logging").getLogger(name)
|
|
sys.modules.setdefault("loggers", _loggers_stub)
|
|
|
|
_structlog_stub = _types.ModuleType("structlog")
|
|
_structlog_stub.get_logger = lambda *a, **k: __import__("logging").getLogger("stub")
|
|
sys.modules.setdefault("structlog", _structlog_stub)
|
|
|
|
import pytest
|
|
|
|
from utils import llama_cpp_freshness as fr
|
|
|
|
|
|
# Helpers.
|
|
|
|
|
|
def _write_marker(install_dir: Path, **overrides) -> Path:
|
|
payload = {
|
|
"requested_tag": "latest",
|
|
"tag": "b9190",
|
|
"release_tag": "b9190",
|
|
"published_repo": "unslothai/llama.cpp",
|
|
"asset": "app-b9190-linux-x64-cuda13-newer.tar.gz",
|
|
"asset_sha256": None,
|
|
"source": "published",
|
|
"installed_at_utc": (datetime.now(tz = timezone.utc) - timedelta(days = 1))
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
}
|
|
payload.update(overrides)
|
|
install_dir.mkdir(parents = True, exist_ok = True)
|
|
(install_dir / "UNSLOTH_PREBUILT_INFO.json").write_text(json.dumps(payload))
|
|
return install_dir / "UNSLOTH_PREBUILT_INFO.json"
|
|
|
|
|
|
def _fake_binary(install_dir: Path, *, layout: str = "cmake") -> Path:
|
|
"""Stub llama-server under a supported install layout."""
|
|
if layout == "cmake":
|
|
bin_dir = install_dir / "build" / "bin"
|
|
bin_name = "llama-server"
|
|
elif layout == "root":
|
|
bin_dir = install_dir
|
|
bin_name = "llama-server"
|
|
elif layout == "windows":
|
|
bin_dir = install_dir / "build" / "bin" / "Release"
|
|
bin_name = "llama-server.exe"
|
|
else:
|
|
raise ValueError(f"unknown layout {layout}")
|
|
bin_dir.mkdir(parents = True, exist_ok = True)
|
|
bin_path = bin_dir / bin_name
|
|
bin_path.write_text("stub\n")
|
|
return bin_path
|
|
|
|
|
|
@pytest.fixture(autouse = True)
|
|
def _reset(monkeypatch, tmp_path):
|
|
# Isolate disk cache per-test; never touch the real cache.
|
|
monkeypatch.setattr(fr, "_cache_dir", lambda: tmp_path / ".freshness")
|
|
fr.reset_caches()
|
|
yield
|
|
fr.reset_caches()
|
|
|
|
|
|
# read_install_marker.
|
|
|
|
|
|
def test_read_install_marker_finds_cmake_layout(tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(install_dir, tag = "b9190")
|
|
bin_path = _fake_binary(install_dir, layout = "cmake")
|
|
marker = fr.read_install_marker(str(bin_path))
|
|
assert marker is not None
|
|
assert marker["tag"] == "b9190"
|
|
assert marker["published_repo"] == "unslothai/llama.cpp"
|
|
|
|
|
|
def test_read_install_marker_finds_root_layout(tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(install_dir, tag = "b9999")
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
marker = fr.read_install_marker(str(bin_path))
|
|
assert marker is not None
|
|
assert marker["tag"] == "b9999"
|
|
|
|
|
|
def test_read_install_marker_finds_windows_cmake_layout(tmp_path):
|
|
# Windows cmake puts the .exe under build/bin/Release/, so the marker
|
|
# is four levels above the binary.
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(install_dir, tag = "b8888")
|
|
bin_path = _fake_binary(install_dir, layout = "windows")
|
|
marker = fr.read_install_marker(str(bin_path))
|
|
assert marker is not None
|
|
assert marker["tag"] == "b8888"
|
|
|
|
|
|
@pytest.mark.parametrize("repo", ["unslothai/llama.cpp", "ggml-org/llama.cpp"])
|
|
def test_read_install_marker_carries_published_repo_dynamically(tmp_path, repo):
|
|
# The freshness check queries whichever release repo the marker records,
|
|
# so CUDA (unslothai), CPU/macOS (ggml-org), and ROCm all get the right
|
|
# "latest" tag.
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(install_dir, tag = "b9000", published_repo = repo)
|
|
bin_path = _fake_binary(install_dir, layout = "cmake")
|
|
marker = fr.read_install_marker(str(bin_path))
|
|
assert marker is not None
|
|
assert marker["published_repo"] == repo
|
|
|
|
|
|
def test_read_install_marker_missing_returns_none(tmp_path):
|
|
bin_path = _fake_binary(tmp_path / "no_marker", layout = "root")
|
|
assert fr.read_install_marker(str(bin_path)) is None
|
|
|
|
|
|
def test_read_install_marker_handles_invalid_json(tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
install_dir.mkdir(parents = True)
|
|
(install_dir / "UNSLOTH_PREBUILT_INFO.json").write_text("not json")
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
assert fr.read_install_marker(str(bin_path)) is None
|
|
|
|
|
|
def test_read_install_marker_handles_none_path():
|
|
assert fr.read_install_marker(None) is None
|
|
|
|
|
|
# latest_published_release (with monkeypatched fetcher).
|
|
|
|
|
|
def test_latest_published_release_uses_disk_cache(monkeypatch):
|
|
calls = []
|
|
|
|
def _fake_fetch(repo, timeout = 5.0):
|
|
calls.append(repo)
|
|
return "b9999"
|
|
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", _fake_fetch)
|
|
first = fr.latest_published_release("unslothai/llama.cpp")
|
|
second = fr.latest_published_release("unslothai/llama.cpp")
|
|
assert first == "b9999"
|
|
assert second == "b9999"
|
|
# Memo + disk cache -> only one fetch.
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_latest_published_release_returns_none_on_network_failure(monkeypatch):
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: None)
|
|
assert fr.latest_published_release("unslothai/llama.cpp") is None
|
|
|
|
|
|
def test_latest_published_release_keeps_old_cache_on_transient_failure(monkeypatch, tmp_path):
|
|
# Disk entry older than TTL + network fail -> return cached value.
|
|
cache_dir = tmp_path / ".freshness"
|
|
cache_dir.mkdir()
|
|
cache_file = cache_dir / "unslothai__llama.cpp.json"
|
|
yesterday = time.time() - 25 * 60 * 60 # > 24h
|
|
cache_file.write_text(json.dumps({"fetched_at": yesterday, "latest_tag": "b9000"}))
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: None)
|
|
assert fr.latest_published_release("unslothai/llama.cpp") == "b9000"
|
|
|
|
|
|
# check_prebuilt_freshness end-to-end.
|
|
|
|
|
|
def test_check_prebuilt_freshness_reports_stale_when_old_and_behind(monkeypatch, tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(
|
|
install_dir,
|
|
tag = "b9190",
|
|
installed_at_utc = (datetime.now(tz = timezone.utc) - timedelta(days = 5))
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
)
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: "b9300")
|
|
info = fr.check_prebuilt_freshness(str(bin_path))
|
|
assert info["has_marker"] is True
|
|
assert info["stale"] is True
|
|
assert info["installed_tag"] == "b9190"
|
|
assert info["latest_tag"] == "b9300"
|
|
assert info["age_days"] == 5
|
|
assert info["published_repo"] == "unslothai/llama.cpp"
|
|
|
|
|
|
def test_check_prebuilt_freshness_not_stale_when_tag_matches(monkeypatch, tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(
|
|
install_dir,
|
|
tag = "b9300",
|
|
installed_at_utc = (datetime.now(tz = timezone.utc) - timedelta(days = 30))
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
)
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: "b9300")
|
|
info = fr.check_prebuilt_freshness(str(bin_path))
|
|
assert info["stale"] is False
|
|
assert info["installed_tag"] == "b9300"
|
|
assert info["latest_tag"] == "b9300"
|
|
|
|
|
|
def test_check_prebuilt_freshness_not_stale_within_threshold(monkeypatch, tmp_path):
|
|
# Behind by tag but within the 3-day grace window.
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(
|
|
install_dir,
|
|
tag = "b9190",
|
|
installed_at_utc = (datetime.now(tz = timezone.utc) - timedelta(days = 1))
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
)
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: "b9300")
|
|
info = fr.check_prebuilt_freshness(str(bin_path))
|
|
assert info["stale"] is False
|
|
assert info["age_days"] == 1
|
|
|
|
|
|
def test_check_prebuilt_freshness_fails_open_without_marker(tmp_path):
|
|
bin_path = _fake_binary(tmp_path / "custom_build", layout = "root")
|
|
info = fr.check_prebuilt_freshness(str(bin_path))
|
|
assert info["has_marker"] is False
|
|
assert info["stale"] is False
|
|
|
|
|
|
def test_check_prebuilt_freshness_fails_open_when_github_unreachable(monkeypatch, tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(
|
|
install_dir,
|
|
tag = "b9190",
|
|
installed_at_utc = (datetime.now(tz = timezone.utc) - timedelta(days = 10))
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
)
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: None)
|
|
info = fr.check_prebuilt_freshness(str(bin_path))
|
|
assert info["has_marker"] is True
|
|
assert info["stale"] is False
|
|
assert info["latest_tag"] is None
|
|
|
|
|
|
def test_check_prebuilt_freshness_handles_unparseable_install_timestamp(monkeypatch, tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(install_dir, tag = "b9190", installed_at_utc = "not-a-date")
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: "b9300")
|
|
info = fr.check_prebuilt_freshness(str(bin_path))
|
|
assert info["stale"] is False
|
|
assert info["age_days"] is None
|
|
|
|
|
|
def test_check_prebuilt_freshness_respects_custom_threshold(monkeypatch, tmp_path):
|
|
install_dir = tmp_path / "llama.cpp"
|
|
_write_marker(
|
|
install_dir,
|
|
tag = "b9190",
|
|
installed_at_utc = (datetime.now(tz = timezone.utc) - timedelta(days = 2))
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
)
|
|
bin_path = _fake_binary(install_dir, layout = "root")
|
|
monkeypatch.setattr(fr, "_fetch_latest_release_tag", lambda repo, timeout = 5.0: "b9300")
|
|
info = fr.check_prebuilt_freshness(str(bin_path), threshold_days = 1)
|
|
assert info["stale"] is True
|
|
|
|
|
|
# format_stale_warning.
|
|
|
|
|
|
def test_format_stale_warning_contains_actionable_command():
|
|
msg = fr.format_stale_warning({"installed_tag": "b9190", "latest_tag": "b9300", "age_days": 5})
|
|
assert "b9190" in msg
|
|
assert "b9300" in msg
|
|
assert "5 days" in msg
|
|
assert "unsloth studio update" in msg
|
|
|
|
|
|
def test_format_stale_warning_singular_day():
|
|
msg = fr.format_stale_warning({"installed_tag": "b9190", "latest_tag": "b9300", "age_days": 1})
|
|
assert "1 day" in msg
|
|
assert "1 days" not in msg
|