mirror of
https://github.com/cloudstack-llc/mlx-knife.git
synced 2026-07-21 01:55:25 -04:00
The BROKE Cluster Team
5751545b8b
- Server: exit codes, /v1/models crash fix, vision routing, MLXK2_MAX_TOKENS - Vision: true SSE streaming, hallucination fix (local numbering) - Workspace: list prefix-match, push ambiguous pattern handling - Docs: SERVER-HANDBOOK accuracy updates See CHANGELOG.md for details.
119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""CLI-arg tests for experimental push (offline)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
# Skip all tests if push is not enabled
|
|
# Push tests now run by default (alpha features included in standard test suite)
|
|
|
|
|
|
def _run_cli(argv: list[str], capsys):
|
|
from mlxk2.cli import main as cli_main
|
|
|
|
# Replace sys.argv and run
|
|
old_argv = sys.argv[:]
|
|
sys.argv = argv[:]
|
|
try:
|
|
with pytest.raises(SystemExit):
|
|
cli_main()
|
|
finally:
|
|
sys.argv = old_argv
|
|
captured = capsys.readouterr()
|
|
return captured.out, captured.err
|
|
|
|
|
|
def test_cli_push_missing_args_json_error(capsys, monkeypatch):
|
|
# Missing required positional args but with --json should emit JSON error
|
|
stdout, stderr = _run_cli(["mlxk2", "push", "--private", "--json"], capsys)
|
|
# JSON mode: all output to stdout (for scripting)
|
|
data = json.loads(stdout)
|
|
assert data["status"] == "error"
|
|
assert data["command"] is None
|
|
assert isinstance(data["error"], dict)
|
|
|
|
|
|
def test_cli_push_workspace_missing_json_error(tmp_path, monkeypatch, capsys):
|
|
# Provide missing workspace; ensure JSON error and specific error type
|
|
monkeypatch.setenv("HF_TOKEN", "dummy")
|
|
missing = str(tmp_path / "nope")
|
|
stdout, stderr = _run_cli(["mlxk2", "push", "--private", missing, "user/repo", "--json"], capsys)
|
|
# JSON mode: all output to stdout (for scripting)
|
|
data = json.loads(stdout)
|
|
assert data["status"] == "error"
|
|
assert data["command"] == "push"
|
|
assert data["error"]["type"] == "workspace_not_found"
|
|
|
|
|
|
def _install_fake_hf(monkeypatch, mode: str):
|
|
class _Errors:
|
|
class HfHubHTTPError(Exception):
|
|
pass
|
|
|
|
class RepositoryNotFoundError(Exception):
|
|
pass
|
|
|
|
class RevisionNotFoundError(Exception):
|
|
pass
|
|
|
|
class _Api:
|
|
def __init__(self, token=None):
|
|
self.token = token
|
|
|
|
def repo_info(self, repo_id: str, repo_type: str, revision: str):
|
|
return {"id": repo_id, "type": repo_type, "rev": revision}
|
|
|
|
def upload_folder(**kwargs): # type: ignore
|
|
if mode == "no_changes":
|
|
# Return an object without commit_id
|
|
return SimpleNamespace()
|
|
else:
|
|
return SimpleNamespace(commit_id="abcdef1234567890abcdef1234567890abcdef12")
|
|
|
|
fake = SimpleNamespace(HfApi=_Api, upload_folder=upload_folder, errors=_Errors)
|
|
# Use monkeypatch to ensure automatic restoration after each test
|
|
monkeypatch.setitem(sys.modules, "huggingface_hub", fake)
|
|
monkeypatch.setitem(sys.modules, "huggingface_hub.errors", _Errors)
|
|
|
|
|
|
def test_cli_push_no_changes_json_output(tmp_path, monkeypatch, capsys):
|
|
# Setup workspace
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
(ws / "x.txt").write_text("x")
|
|
monkeypatch.setenv("HF_TOKEN", "dummy")
|
|
|
|
_install_fake_hf(monkeypatch, mode="no_changes")
|
|
|
|
stdout, stderr = _run_cli(["mlxk2", "push", "--private", str(ws), "user/repo", "--json"], capsys)
|
|
# Success goes to stdout
|
|
data = json.loads(stdout)
|
|
assert data["status"] == "success"
|
|
assert data["command"] == "push"
|
|
assert data["data"]["no_changes"] is True
|
|
assert data["data"]["uploaded_files_count"] == 0
|
|
|
|
|
|
def test_cli_push_with_changes_json_output(tmp_path, monkeypatch, capsys):
|
|
# Setup workspace
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
(ws / "x.txt").write_text("x")
|
|
monkeypatch.setenv("HF_TOKEN", "dummy")
|
|
|
|
_install_fake_hf(monkeypatch, mode="with_changes")
|
|
|
|
stdout, stderr = _run_cli(["mlxk2", "push", "--private", str(ws), "user/repo", "--json"], capsys)
|
|
# Success goes to stdout
|
|
data = json.loads(stdout)
|
|
assert data["status"] == "success"
|
|
assert data["command"] == "push"
|
|
assert data["data"]["no_changes"] is False
|
|
assert isinstance(data["data"]["commit_sha"], str)
|