mirror of https://github.com/razor-ai/soup.git
554 lines
19 KiB
Python
554 lines
19 KiB
Python
"""Subprocess CLI tests — real process execution to catch platform-specific bugs.
|
|
|
|
These tests run `soup` (or `python -m soup_cli`) as a real subprocess,
|
|
catching issues that in-process CliRunner misses:
|
|
- encoding bugs (cp1251/cp1252 on Windows)
|
|
- path separator issues
|
|
- pipe buffering / stdout encoding
|
|
- entry-point script resolution
|
|
- exit code propagation through the OS shell
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
# Use `python -m soup_cli` for reliability across platforms (no need for
|
|
# the entry-point script to be installed or on PATH).
|
|
SOUP_CMD = [sys.executable, "-m", "soup_cli"]
|
|
|
|
# Timeout for all subprocess calls (seconds). Tests should be fast —
|
|
# these are smoke-level checks, not training runs.
|
|
TIMEOUT = 30
|
|
|
|
|
|
def run_soup(
|
|
*args: str, timeout: int = TIMEOUT, env: dict | None = None,
|
|
) -> subprocess.CompletedProcess:
|
|
"""Run soup CLI as a subprocess and return CompletedProcess."""
|
|
merged_env = {**os.environ, **(env or {})}
|
|
# Force UTF-8 in the child process (affects print / Rich Console output)
|
|
merged_env["PYTHONIOENCODING"] = "utf-8"
|
|
merged_env["PYTHONLEGACYWINDOWSSTDIO"] = "0"
|
|
return subprocess.run(
|
|
[*SOUP_CMD, *args],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=timeout,
|
|
env=merged_env,
|
|
# Decode as UTF-8 on the *parent* side too — avoids cp1251/cp1252
|
|
# failing on Rich box-drawing characters.
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Basic entry-point tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEntryPoint:
|
|
"""Verify the CLI launches correctly as a real process."""
|
|
|
|
def test_no_args_shows_help(self):
|
|
result = run_soup()
|
|
# Typer with no_args_is_help=True returns exit code 0 or 2
|
|
assert result.returncode in (0, 2)
|
|
combined = (result.stdout or "") + (result.stderr or "")
|
|
assert "Fine-tune" in combined or "Usage" in combined
|
|
|
|
def test_help_flag(self):
|
|
result = run_soup("--help")
|
|
assert result.returncode == 0
|
|
assert "Fine-tune" in result.stdout
|
|
|
|
def test_version(self):
|
|
from soup_cli import __version__
|
|
|
|
result = run_soup("version")
|
|
assert result.returncode == 0
|
|
assert __version__ in result.stdout
|
|
|
|
def test_version_full(self):
|
|
result = run_soup("version", "--full")
|
|
assert result.returncode == 0
|
|
assert "Python" in result.stdout
|
|
|
|
def test_unknown_command_fails(self):
|
|
result = run_soup("nonexistent_command_xyz")
|
|
assert result.returncode != 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command --help tests (every command must print help without crashing)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
ALL_COMMANDS = [
|
|
"init",
|
|
"train",
|
|
"chat",
|
|
"push",
|
|
"export",
|
|
"merge",
|
|
"eval",
|
|
"serve",
|
|
"sweep",
|
|
"diff",
|
|
"infer",
|
|
"doctor",
|
|
"quickstart",
|
|
"ui",
|
|
"version",
|
|
]
|
|
|
|
ALL_DATA_SUBCOMMANDS = [
|
|
"inspect",
|
|
"validate",
|
|
"convert",
|
|
"merge",
|
|
"dedup",
|
|
"stats",
|
|
"generate",
|
|
]
|
|
|
|
ALL_RUNS_SUBCOMMANDS = [
|
|
"show",
|
|
"compare",
|
|
"delete",
|
|
]
|
|
|
|
|
|
class TestCommandHelp:
|
|
"""Every registered command must respond to --help without crashing."""
|
|
|
|
@pytest.mark.parametrize("cmd", ALL_COMMANDS)
|
|
def test_command_help(self, cmd):
|
|
result = run_soup(cmd, "--help")
|
|
assert result.returncode == 0, (
|
|
f"`soup {cmd} --help` failed (rc={result.returncode}):\n{result.stderr}"
|
|
)
|
|
# Should contain some usage info
|
|
assert len(result.stdout) > 20, f"`soup {cmd} --help` output too short"
|
|
|
|
@pytest.mark.parametrize("subcmd", ALL_DATA_SUBCOMMANDS)
|
|
def test_data_subcommand_help(self, subcmd):
|
|
result = run_soup("data", subcmd, "--help")
|
|
assert result.returncode == 0, (
|
|
f"`soup data {subcmd} --help` failed (rc={result.returncode}):\n{result.stderr}"
|
|
)
|
|
|
|
@pytest.mark.parametrize("subcmd", ALL_RUNS_SUBCOMMANDS)
|
|
def test_runs_subcommand_help(self, subcmd):
|
|
result = run_soup("runs", subcmd, "--help")
|
|
assert result.returncode == 0, (
|
|
f"`soup runs {subcmd} --help` failed (rc={result.returncode}):\n{result.stderr}"
|
|
)
|
|
|
|
def test_data_no_args_shows_help(self):
|
|
result = run_soup("data")
|
|
# Typer returns 0 or 2 depending on no_args_is_help vs missing args
|
|
assert result.returncode in (0, 2)
|
|
combined = (result.stdout or "") + (result.stderr or "")
|
|
assert "inspect" in combined.lower() or "usage" in combined.lower()
|
|
|
|
def test_runs_lists_or_shows_help(self):
|
|
result = run_soup("runs")
|
|
# runs without subcommand lists experiments (may be empty) — exit 0
|
|
assert result.returncode == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Error handling / exit code tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestErrorHandling:
|
|
"""Verify errors produce correct exit codes and readable messages."""
|
|
|
|
def test_train_missing_config(self):
|
|
result = run_soup("train", "--config", "nonexistent_file.yaml")
|
|
assert result.returncode == 1
|
|
|
|
def test_chat_missing_model(self):
|
|
result = run_soup("chat", "--model", "nonexistent_model_path")
|
|
assert result.returncode == 1
|
|
|
|
def test_push_missing_model(self):
|
|
result = run_soup(
|
|
"push", "--model", "nonexistent_model_path", "--repo", "user/model"
|
|
)
|
|
assert result.returncode == 1
|
|
|
|
def test_init_unknown_template(self):
|
|
result = run_soup("init", "--template", "nonexistent_template")
|
|
assert result.returncode == 1
|
|
|
|
def test_export_missing_model(self):
|
|
result = run_soup("export", "--model", "nonexistent_model_path")
|
|
assert result.returncode == 1
|
|
|
|
def test_merge_missing_adapter(self):
|
|
result = run_soup("merge", "--adapter", "nonexistent_adapter_path")
|
|
assert result.returncode == 1
|
|
|
|
def test_infer_missing_files(self):
|
|
result = run_soup(
|
|
"infer",
|
|
"--model", "nonexistent_model",
|
|
"--input", "nonexistent.jsonl",
|
|
"--output", "out.jsonl",
|
|
)
|
|
assert result.returncode == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Encoding / Unicode safety on all platforms
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEncoding:
|
|
"""Verify output contains no problematic Unicode on any platform."""
|
|
|
|
PROBLEMATIC_CHARS = [
|
|
"\u2192", # → right arrow
|
|
"\u2014", # — em dash
|
|
"\u2018", # ' left single quote
|
|
"\u2019", # ' right single quote
|
|
"\u201c", # " left double quote
|
|
"\u201d", # " right double quote
|
|
]
|
|
|
|
def test_version_output_is_ascii_safe(self):
|
|
result = run_soup("version")
|
|
assert result.returncode == 0
|
|
for char in self.PROBLEMATIC_CHARS:
|
|
assert char not in result.stdout, (
|
|
f"Version output contains problematic Unicode char: {repr(char)}"
|
|
)
|
|
|
|
def test_help_output_is_ascii_safe(self):
|
|
result = run_soup("--help")
|
|
assert result.returncode == 0
|
|
for char in self.PROBLEMATIC_CHARS:
|
|
assert char not in result.stdout, (
|
|
f"Help output contains problematic Unicode char: {repr(char)}"
|
|
)
|
|
|
|
def test_error_output_is_ascii_safe(self):
|
|
"""Error messages must be safe for Windows terminals."""
|
|
result = run_soup("train", "--config", "nonexistent.yaml")
|
|
combined = result.stdout + result.stderr
|
|
for char in self.PROBLEMATIC_CHARS:
|
|
assert char not in combined, (
|
|
f"Error output contains problematic Unicode char: {repr(char)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data command — real file I/O
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDataCommands:
|
|
"""Test data subcommands with actual file operations."""
|
|
|
|
def test_data_inspect_jsonl(self, tmp_path):
|
|
"""Inspect a small JSONL file via subprocess."""
|
|
data_file = tmp_path / "test.jsonl"
|
|
rows = [
|
|
{"messages": [
|
|
{"role": "user", "content": "Hello"},
|
|
{"role": "assistant", "content": "Hi there!"},
|
|
]}
|
|
for _ in range(5)
|
|
]
|
|
data_file.write_text(
|
|
"\n".join(json.dumps(r) for r in rows), encoding="utf-8"
|
|
)
|
|
result = run_soup("data", "inspect", str(data_file))
|
|
assert result.returncode == 0
|
|
assert "5" in result.stdout # row count
|
|
|
|
def test_data_inspect_nonexistent_file(self):
|
|
result = run_soup("data", "inspect", "nonexistent_file.jsonl")
|
|
assert result.returncode == 1
|
|
|
|
def test_data_validate_jsonl(self, tmp_path):
|
|
"""Validate a well-formed JSONL file."""
|
|
data_file = tmp_path / "valid.jsonl"
|
|
rows = [
|
|
{"messages": [
|
|
{"role": "user", "content": f"Q{i}"},
|
|
{"role": "assistant", "content": f"A{i}"},
|
|
]}
|
|
for i in range(3)
|
|
]
|
|
data_file.write_text(
|
|
"\n".join(json.dumps(r) for r in rows), encoding="utf-8"
|
|
)
|
|
result = run_soup("data", "validate", str(data_file))
|
|
assert result.returncode == 0
|
|
|
|
def test_data_stats_jsonl(self, tmp_path):
|
|
"""Stats command should work without encoding errors on any OS."""
|
|
data_file = tmp_path / "stats.jsonl"
|
|
rows = [
|
|
{"messages": [
|
|
{"role": "user", "content": f"Question {i}"},
|
|
{"role": "assistant", "content": f"Answer {i} " + "x" * (i * 10)},
|
|
]}
|
|
for i in range(10)
|
|
]
|
|
data_file.write_text(
|
|
"\n".join(json.dumps(r) for r in rows), encoding="utf-8"
|
|
)
|
|
result = run_soup("data", "stats", str(data_file))
|
|
assert result.returncode == 0, (
|
|
f"data stats failed (rc={result.returncode}):\n{result.stderr}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Init command — file creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestInitCommand:
|
|
"""Test init command creates files correctly."""
|
|
|
|
TEMPLATES = ["chat", "code", "medical", "reasoning", "vision"]
|
|
|
|
@pytest.mark.parametrize("template", TEMPLATES)
|
|
def test_init_creates_config_file(self, tmp_path, template):
|
|
"""soup init --template X should create a valid YAML config."""
|
|
result = run_soup(
|
|
"init",
|
|
"--template", template,
|
|
"--output", str(tmp_path / f"{template}.yaml"),
|
|
)
|
|
assert result.returncode == 0, (
|
|
f"init --template {template} failed:\n{result.stderr}"
|
|
)
|
|
config_file = tmp_path / f"{template}.yaml"
|
|
assert config_file.exists(), f"Config file not created for template {template}"
|
|
content = config_file.read_text(encoding="utf-8")
|
|
assert "base:" in content
|
|
|
|
def test_init_default_output(self, tmp_path, monkeypatch):
|
|
"""soup init --template chat writes soup.yaml in cwd."""
|
|
monkeypatch.chdir(tmp_path)
|
|
result = run_soup("init", "--template", "chat")
|
|
assert result.returncode == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Doctor command — system check
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDoctorCommand:
|
|
"""Doctor should always succeed (even without GPU)."""
|
|
|
|
def test_doctor_runs(self):
|
|
# doctor imports torch and checks GPU — allow extra time
|
|
result = run_soup("doctor", timeout=120)
|
|
assert result.returncode == 0
|
|
# Should mention Python at minimum
|
|
assert "python" in result.stdout.lower() or "Python" in result.stdout
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Verbose flag
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVerboseFlag:
|
|
"""--verbose / -V flag should not crash."""
|
|
|
|
def test_verbose_with_version(self):
|
|
result = run_soup("--verbose", "version")
|
|
assert result.returncode == 0
|
|
|
|
def test_verbose_short_flag(self):
|
|
result = run_soup("-V", "version")
|
|
assert result.returncode == 0
|
|
|
|
def test_verbose_error_shows_traceback(self):
|
|
result = run_soup("--verbose", "train", "--config", "nonexistent.yaml")
|
|
assert result.returncode == 1
|
|
# Verbose mode should show traceback details
|
|
combined = result.stdout + result.stderr
|
|
assert len(combined) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Platform-specific regression tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPlatformRegression:
|
|
"""Regressions that appeared on specific platforms."""
|
|
|
|
def test_stdout_encoding_no_crash(self):
|
|
"""Verify stdout doesn't crash with encoding issues (Windows cp1252 bug)."""
|
|
result = run_soup("version", "--full")
|
|
assert result.returncode == 0
|
|
# Output should be decodable text, no mojibake
|
|
assert result.stdout.isprintable() or "\n" in result.stdout
|
|
|
|
def test_path_with_spaces(self, tmp_path):
|
|
"""Paths with spaces must work on all OSes."""
|
|
spaced_dir = tmp_path / "path with spaces"
|
|
spaced_dir.mkdir()
|
|
data_file = spaced_dir / "test.jsonl"
|
|
data_file.write_text(
|
|
json.dumps({"messages": [
|
|
{"role": "user", "content": "Hi"},
|
|
{"role": "assistant", "content": "Hello"},
|
|
]}),
|
|
encoding="utf-8",
|
|
)
|
|
result = run_soup("data", "inspect", str(data_file))
|
|
assert result.returncode == 0
|
|
|
|
@pytest.mark.skipif(sys.platform != "win32", reason="Windows-only test")
|
|
def test_windows_long_path(self, tmp_path):
|
|
"""Long paths should work on Windows (>100 chars)."""
|
|
long_dir = tmp_path
|
|
for i in range(5):
|
|
long_dir = long_dir / f"subdirectory_level_{i}_name"
|
|
long_dir.mkdir()
|
|
data_file = long_dir / "data.jsonl"
|
|
data_file.write_text(
|
|
json.dumps({"messages": [
|
|
{"role": "user", "content": "Test"},
|
|
{"role": "assistant", "content": "OK"},
|
|
]}),
|
|
encoding="utf-8",
|
|
)
|
|
assert len(str(data_file)) > 100
|
|
result = run_soup("data", "inspect", str(data_file))
|
|
assert result.returncode == 0
|
|
|
|
def test_unicode_in_data_file(self, tmp_path):
|
|
"""Data files with Unicode content must load on all platforms."""
|
|
data_file = tmp_path / "unicode.jsonl"
|
|
rows = [
|
|
{"messages": [
|
|
{"role": "user", "content": "Wie geht es dir?"},
|
|
{"role": "assistant", "content": "Mir geht es gut, danke!"},
|
|
]},
|
|
{"messages": [
|
|
{"role": "user", "content": "Как дела?"},
|
|
{"role": "assistant", "content": "Хорошо, спасибо!"},
|
|
]},
|
|
{"messages": [
|
|
{"role": "user", "content": "お元気ですか？"},
|
|
{"role": "assistant", "content": "元気です、ありがとう！"},
|
|
]},
|
|
]
|
|
data_file.write_text(
|
|
"\n".join(json.dumps(r, ensure_ascii=False) for r in rows),
|
|
encoding="utf-8",
|
|
)
|
|
result = run_soup("data", "inspect", str(data_file))
|
|
assert result.returncode == 0
|
|
|
|
def test_empty_jsonl_file(self, tmp_path):
|
|
"""Empty data file should not crash (may show 0 rows)."""
|
|
data_file = tmp_path / "empty.jsonl"
|
|
data_file.write_text("", encoding="utf-8")
|
|
result = run_soup("data", "inspect", str(data_file))
|
|
# Should complete without crashing — exit 0 (0 rows) or 1 (error)
|
|
assert result.returncode in (0, 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Quickstart --dry-run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestQuickstart:
|
|
"""Quickstart command basic checks."""
|
|
|
|
def test_quickstart_help(self):
|
|
result = run_soup("quickstart", "--help")
|
|
assert result.returncode == 0
|
|
assert "dry" in result.stdout.lower() or "demo" in result.stdout.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sweep command
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSweepCommand:
|
|
"""Sweep command validation."""
|
|
|
|
def test_sweep_missing_config(self):
|
|
result = run_soup(
|
|
"sweep", "--config", "nonexistent.yaml",
|
|
"--param", "lr=1e-4,2e-4",
|
|
)
|
|
assert result.returncode == 1
|
|
|
|
def test_sweep_help(self):
|
|
result = run_soup("sweep", "--help")
|
|
assert result.returncode == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Diff command
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDiffCommand:
|
|
"""Diff command validation."""
|
|
|
|
def test_diff_help(self):
|
|
result = run_soup("diff", "--help")
|
|
assert result.returncode == 0
|
|
assert "model" in result.stdout.lower() or "compare" in result.stdout.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Eval command
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEvalCommand:
|
|
"""Eval command validation."""
|
|
|
|
def test_eval_help(self):
|
|
result = run_soup("eval", "--help")
|
|
assert result.returncode == 0
|
|
|
|
def test_eval_missing_model(self):
|
|
result = run_soup("eval", "benchmark", "--model", "nonexistent_model")
|
|
assert result.returncode == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Serve / UI commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestServeUI:
|
|
"""Serve and UI commands help checks (don't actually start servers)."""
|
|
|
|
def test_serve_help(self):
|
|
result = run_soup("serve", "--help")
|
|
assert result.returncode == 0
|
|
assert "backend" in result.stdout.lower() or "model" in result.stdout.lower()
|
|
|
|
def test_ui_help(self):
|
|
result = run_soup("ui", "--help")
|
|
assert result.returncode == 0
|