feat(v2): T11 打包分发 setup_runtime(下载/续传/解压/硬件档位)
This commit is contained in:
@@ -0,0 +1,123 @@
|
||||
"""T11 setup_runtime 单测(封闭:假 opener / 假 zip)。"""
|
||||
import io
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
from scripts.setup_runtime import (
|
||||
Downloader,
|
||||
extract_llama_server,
|
||||
manual_instructions,
|
||||
parse_size_from_length,
|
||||
validate_size,
|
||||
)
|
||||
|
||||
|
||||
class _FakeResp:
|
||||
def __init__(self, data, status=200):
|
||||
self._buf = io.BytesIO(data)
|
||||
self.status = status
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def read(self, n=-1):
|
||||
return self._buf.read(n)
|
||||
|
||||
|
||||
class _FakeOpener:
|
||||
def __init__(self, data=b"hello world"):
|
||||
self.data = data
|
||||
self.request_url = None
|
||||
self.request_headers = {}
|
||||
|
||||
def __call__(self):
|
||||
return self
|
||||
|
||||
def open(self, req, timeout=None):
|
||||
self.request_url = req.full_url
|
||||
self.request_headers = dict(req.headers)
|
||||
return _FakeResp(self.data)
|
||||
|
||||
|
||||
def test_parse_size_from_length():
|
||||
assert parse_size_from_length("1024") == 1024
|
||||
assert parse_size_from_length(" 42 ") == 42
|
||||
assert parse_size_from_length(None) is None
|
||||
assert parse_size_from_length("abc") is None
|
||||
|
||||
|
||||
def test_validate_size(tmp_path):
|
||||
p = tmp_path / "size.txt"
|
||||
p.write_bytes(b"x" * (2 * 1024 * 1024)) # 2MB
|
||||
ok, actual = validate_size(p, 2 * 1024 * 1024)
|
||||
assert ok is True and actual == 2 * 1024 * 1024
|
||||
ok, _ = validate_size(p, 100) # 2MB vs 100 超出 ±1MB 容差
|
||||
assert ok is False
|
||||
ok, _ = validate_size(p, 0) # expected=0 -> 仅存在性
|
||||
assert ok is True
|
||||
|
||||
|
||||
def test_downloader_writes_file(tmp_path):
|
||||
opener = _FakeOpener(b"ABCDEF")
|
||||
dl = Downloader(chunk=2, opener_factory=lambda: opener)
|
||||
dest = tmp_path / "out.bin"
|
||||
written, err = dl.download("https://x/y", dest)
|
||||
assert err is None
|
||||
assert written == 6
|
||||
assert dest.read_bytes() == b"ABCDEF"
|
||||
assert opener.request_headers.get("Range") is None # 无既有文件 -> 不带 Range
|
||||
|
||||
|
||||
def test_downloader_resumes(tmp_path):
|
||||
opener = _FakeOpener(b"CDEF")
|
||||
dl = Downloader(chunk=2, opener_factory=lambda: opener)
|
||||
dest = tmp_path / "out.bin"
|
||||
dest.write_bytes(b"AB") # 既有 2 字节 -> 应带 Range: bytes=2-
|
||||
written, err = dl.download("https://x/y", dest)
|
||||
assert err is None
|
||||
assert written == 6 # 2 + 4
|
||||
assert dest.read_bytes() == b"ABCDEF"
|
||||
assert opener.request_headers.get("Range") == "bytes=2-"
|
||||
|
||||
|
||||
def test_downloader_failure_returns_error(tmp_path):
|
||||
class Boom:
|
||||
def __call__(self):
|
||||
return self
|
||||
|
||||
def open(self, req, timeout=None):
|
||||
raise OSError("network down")
|
||||
|
||||
dl = Downloader(opener_factory=lambda: Boom())
|
||||
dest = tmp_path / "out.bin"
|
||||
written, err = dl.download("https://x/y", dest)
|
||||
assert err is not None
|
||||
assert "OSError" in err
|
||||
|
||||
|
||||
def test_extract_llama_server(tmp_path):
|
||||
zip_path = tmp_path / "llama.zip"
|
||||
bin_dir = tmp_path / "bin"
|
||||
with zipfile.ZipFile(zip_path, "w") as zf:
|
||||
zf.writestr("llama-b3662/bin/llama-server.exe", b"MZfake")
|
||||
zf.writestr("llama-b3662/README.md", "readme")
|
||||
err = extract_llama_server(zip_path, bin_dir)
|
||||
assert err is None
|
||||
assert (bin_dir / "llama-server.exe").exists()
|
||||
|
||||
|
||||
def test_extract_missing_exe(tmp_path):
|
||||
zip_path = tmp_path / "no.exe.zip"
|
||||
with zipfile.ZipFile(zip_path, "w") as zf:
|
||||
zf.writestr("a.txt", "x")
|
||||
err = extract_llama_server(zip_path, tmp_path / "bin")
|
||||
assert err is not None
|
||||
assert "未找到" in err
|
||||
|
||||
|
||||
def test_manual_instructions_nonempty():
|
||||
s = manual_instructions()
|
||||
assert "llama-server" in s and "GGUF" in s
|
||||
Reference in New Issue
Block a user