mirror of https://github.com/razor-ai/soup.git
791 lines
28 KiB
Python
791 lines
28 KiB
Python
"""Tests for v0.27.0 Multi-GPU Mastery.
|
|
|
|
Covers: topology, launcher, ZeRO++, FSDP+compile, MII, pipeline, recipes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
# =============================================================================
|
|
# Part A: Topology detector + --gpus flag
|
|
# =============================================================================
|
|
|
|
class TestResolveNumGpus:
|
|
"""Test --gpus resolution: int, 'auto', None."""
|
|
|
|
def test_resolve_integer(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
assert resolve_num_gpus("4") == 4
|
|
assert resolve_num_gpus(2) == 2
|
|
|
|
def test_resolve_auto_uses_detected(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
with patch(
|
|
"soup_cli.utils.topology._detected_gpu_count", return_value=8
|
|
):
|
|
assert resolve_num_gpus("auto") == 8
|
|
|
|
def test_resolve_auto_cpu_returns_zero(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
with patch(
|
|
"soup_cli.utils.topology._detected_gpu_count", return_value=0
|
|
):
|
|
assert resolve_num_gpus("auto") == 0
|
|
|
|
def test_resolve_none_returns_none(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
assert resolve_num_gpus(None) is None
|
|
|
|
def test_resolve_invalid_negative(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
with pytest.raises(ValueError, match="must be"):
|
|
resolve_num_gpus("-1")
|
|
|
|
def test_resolve_invalid_zero(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
with pytest.raises(ValueError, match="must be"):
|
|
resolve_num_gpus("0")
|
|
|
|
def test_resolve_invalid_string(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
with pytest.raises(ValueError, match="Invalid"):
|
|
resolve_num_gpus("all")
|
|
|
|
def test_resolve_out_of_bounds(self):
|
|
from soup_cli.utils.topology import resolve_num_gpus
|
|
|
|
with pytest.raises(ValueError, match="exceeds"):
|
|
resolve_num_gpus("1024")
|
|
|
|
|
|
class TestDetectTopology:
|
|
"""Test topology sniffing (NVLink / PCIe) — all mocked."""
|
|
|
|
def test_no_cuda_returns_none_interconnect(self):
|
|
from soup_cli.utils.topology import detect_topology
|
|
|
|
with patch(
|
|
"soup_cli.utils.topology._detected_gpu_count", return_value=0
|
|
):
|
|
topo = detect_topology()
|
|
assert topo["gpu_count"] == 0
|
|
assert topo["interconnect"] == "none"
|
|
assert topo["nvlink_pairs"] == 0
|
|
|
|
def test_single_gpu_has_no_interconnect(self):
|
|
from soup_cli.utils.topology import detect_topology
|
|
|
|
with patch(
|
|
"soup_cli.utils.topology._detected_gpu_count", return_value=1
|
|
):
|
|
topo = detect_topology()
|
|
assert topo["gpu_count"] == 1
|
|
assert topo["interconnect"] == "single"
|
|
assert topo["nvlink_pairs"] == 0
|
|
|
|
def test_multi_gpu_nvlink_detected(self):
|
|
from soup_cli.utils.topology import detect_topology
|
|
|
|
# Mock: 4 GPUs, all connected via NVLink
|
|
with (
|
|
patch(
|
|
"soup_cli.utils.topology._detected_gpu_count", return_value=4
|
|
),
|
|
patch(
|
|
"soup_cli.utils.topology._count_nvlink_pairs", return_value=6
|
|
),
|
|
):
|
|
topo = detect_topology()
|
|
assert topo["gpu_count"] == 4
|
|
assert topo["interconnect"] == "nvlink"
|
|
assert topo["nvlink_pairs"] == 6
|
|
|
|
def test_multi_gpu_pcie_fallback(self):
|
|
from soup_cli.utils.topology import detect_topology
|
|
|
|
# 2 GPUs, no NVLink
|
|
with (
|
|
patch(
|
|
"soup_cli.utils.topology._detected_gpu_count", return_value=2
|
|
),
|
|
patch(
|
|
"soup_cli.utils.topology._count_nvlink_pairs", return_value=0
|
|
),
|
|
):
|
|
topo = detect_topology()
|
|
assert topo["gpu_count"] == 2
|
|
assert topo["interconnect"] == "pcie"
|
|
|
|
def test_nccl_env_suggestions(self):
|
|
from soup_cli.utils.topology import suggest_nccl_env
|
|
|
|
env = suggest_nccl_env(gpu_count=4, interconnect="nvlink")
|
|
assert env["NCCL_P2P_DISABLE"] == "0"
|
|
assert env["NCCL_IB_DISABLE"] == "1" # no IB in local multi-GPU
|
|
|
|
env_pcie = suggest_nccl_env(gpu_count=2, interconnect="pcie")
|
|
assert env_pcie["NCCL_P2P_DISABLE"] == "0"
|
|
|
|
env_single = suggest_nccl_env(gpu_count=1, interconnect="single")
|
|
assert env_single == {}
|
|
|
|
|
|
class TestSuggestStrategy:
|
|
"""Test distributed-strategy suggestion based on model size + GPU count."""
|
|
|
|
def test_single_gpu_no_strategy(self):
|
|
from soup_cli.utils.topology import suggest_strategy
|
|
|
|
assert suggest_strategy(gpu_count=1, model_size_b=7.0)["strategy"] == "single"
|
|
|
|
def test_small_model_multi_gpu_ddp(self):
|
|
from soup_cli.utils.topology import suggest_strategy
|
|
|
|
rec = suggest_strategy(gpu_count=2, model_size_b=3.0)
|
|
assert rec["strategy"] in ("ddp", "zero2")
|
|
|
|
def test_mid_model_multi_gpu_zero2(self):
|
|
from soup_cli.utils.topology import suggest_strategy
|
|
|
|
rec = suggest_strategy(gpu_count=4, model_size_b=13.0)
|
|
assert rec["strategy"] in ("zero2", "zero3", "fsdp_full_shard")
|
|
|
|
def test_huge_model_zero3_or_fsdp(self):
|
|
from soup_cli.utils.topology import suggest_strategy
|
|
|
|
rec = suggest_strategy(gpu_count=8, model_size_b=70.0)
|
|
assert rec["strategy"] in ("zero3", "fsdp_full_shard", "zero2_offload")
|
|
|
|
def test_cpu_suggests_none(self):
|
|
from soup_cli.utils.topology import suggest_strategy
|
|
|
|
rec = suggest_strategy(gpu_count=0, model_size_b=7.0)
|
|
assert rec["strategy"] == "none"
|
|
|
|
|
|
class TestTrainGpusFlag:
|
|
"""Test --gpus flag in train command."""
|
|
|
|
def test_train_help_shows_gpus(self):
|
|
from typer.testing import CliRunner
|
|
|
|
from soup_cli.cli import app
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(app, ["train", "--help"])
|
|
# Strip ANSI + whitespace per v0.26.0 CI hardening pattern
|
|
out = re.sub(r"\x1b\[[0-9;]*m", "", result.output)
|
|
out = re.sub(r"\s+", " ", out)
|
|
assert "--gpus" in out
|
|
|
|
|
|
# =============================================================================
|
|
# Part B: Accelerate launcher wrapper
|
|
# =============================================================================
|
|
|
|
class TestLauncherArgv:
|
|
"""Test accelerate-launch argv construction."""
|
|
|
|
def test_single_process_no_launcher(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
# num_processes=1 → no wrapper needed
|
|
argv = build_accelerate_argv(num_processes=1, script_args=["soup", "train"])
|
|
assert argv == ["soup", "train"]
|
|
|
|
def test_multi_process_wraps(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
argv = build_accelerate_argv(num_processes=4, script_args=["soup", "train"])
|
|
assert argv[0] == "accelerate"
|
|
assert argv[1] == "launch"
|
|
assert "--num_processes" in argv
|
|
# num_processes value follows the flag
|
|
idx = argv.index("--num_processes")
|
|
assert argv[idx + 1] == "4"
|
|
|
|
def test_multi_process_mixed_precision(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
argv = build_accelerate_argv(
|
|
num_processes=4, script_args=["soup", "train"], mixed_precision="bf16"
|
|
)
|
|
assert "--mixed_precision" in argv
|
|
idx = argv.index("--mixed_precision")
|
|
assert argv[idx + 1] == "bf16"
|
|
|
|
def test_invalid_num_processes(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
with pytest.raises(ValueError, match="num_processes"):
|
|
build_accelerate_argv(num_processes=0, script_args=["soup", "train"])
|
|
|
|
def test_invalid_mixed_precision(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
with pytest.raises(ValueError, match="mixed_precision"):
|
|
build_accelerate_argv(
|
|
num_processes=2, script_args=["soup", "train"], mixed_precision="int8"
|
|
)
|
|
|
|
def test_invalid_num_machines_low(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
with pytest.raises(ValueError, match="num_machines"):
|
|
build_accelerate_argv(
|
|
num_processes=2, script_args=["soup", "train"], num_machines=0
|
|
)
|
|
|
|
def test_invalid_num_machines_high(self):
|
|
from soup_cli.utils.launcher import build_accelerate_argv
|
|
|
|
with pytest.raises(ValueError, match="num_machines"):
|
|
build_accelerate_argv(
|
|
num_processes=2, script_args=["soup", "train"], num_machines=10_000
|
|
)
|
|
|
|
|
|
class TestLauncherDetectActive:
|
|
"""Test detection of whether we're running under accelerate/torchrun.
|
|
|
|
Uses ``clear=False`` + explicit key deletion so Windows CI keeps
|
|
SYSTEMROOT / PATH / TEMP available for the rest of the process.
|
|
"""
|
|
|
|
_DIST_KEYS = (
|
|
"RANK",
|
|
"WORLD_SIZE",
|
|
"LOCAL_RANK",
|
|
"ACCELERATE_MIXED_PRECISION",
|
|
"ACCELERATE_USE_DEEPSPEED",
|
|
"ACCELERATE_USE_FSDP",
|
|
)
|
|
|
|
def test_not_in_distributed(self):
|
|
from soup_cli.utils.launcher import is_in_distributed
|
|
|
|
# Delete any pre-existing distributed markers without wiping env.
|
|
stripped_env = {
|
|
key: val for key, val in os.environ.items() if key not in self._DIST_KEYS
|
|
}
|
|
with patch.dict(os.environ, stripped_env, clear=True):
|
|
assert is_in_distributed() is False
|
|
|
|
def test_in_torchrun(self):
|
|
from soup_cli.utils.launcher import is_in_distributed
|
|
|
|
env = {"RANK": "0", "WORLD_SIZE": "4", "LOCAL_RANK": "0"}
|
|
with patch.dict(os.environ, env, clear=False):
|
|
assert is_in_distributed() is True
|
|
|
|
def test_in_accelerate(self):
|
|
from soup_cli.utils.launcher import is_in_distributed
|
|
|
|
env = {"ACCELERATE_MIXED_PRECISION": "bf16", "RANK": "0", "WORLD_SIZE": "2"}
|
|
with patch.dict(os.environ, env, clear=False):
|
|
assert is_in_distributed() is True
|
|
|
|
|
|
class TestLauncherAdvice:
|
|
"""Test that launcher prints useful advice message when N>1 and not distributed."""
|
|
|
|
def test_format_advice_command(self):
|
|
from soup_cli.utils.launcher import format_advice
|
|
|
|
text = format_advice(num_processes=4, script_args=["soup", "train", "-c", "soup.yaml"])
|
|
assert "accelerate launch" in text
|
|
assert "--num_processes 4" in text
|
|
assert "soup train" in text
|
|
|
|
|
|
# =============================================================================
|
|
# Part C: ZeRO++ DeepSpeed template
|
|
# =============================================================================
|
|
|
|
class TestZeroPlusPlus:
|
|
"""Test ZeRO++ (zero_plus_plus / zero++) config template."""
|
|
|
|
def test_zero_pp_config_exists(self):
|
|
from soup_cli.utils.deepspeed import CONFIGS
|
|
|
|
assert "zero++" in CONFIGS or "zero_pp" in CONFIGS
|
|
|
|
def test_zero_pp_has_stage_3_base(self):
|
|
from soup_cli.utils.deepspeed import get_deepspeed_config
|
|
|
|
cfg = get_deepspeed_config("zero++")
|
|
assert cfg["zero_optimization"]["stage"] == 3
|
|
|
|
def test_zero_pp_has_hierarchical_comms(self):
|
|
from soup_cli.utils.deepspeed import get_deepspeed_config
|
|
|
|
cfg = get_deepspeed_config("zero++")
|
|
zopt = cfg["zero_optimization"]
|
|
# ZeRO++ distinguishing features: quantized weights/gradients + hpz
|
|
assert "zero_quantized_weights" in zopt
|
|
assert "zero_hpz_partition_size" in zopt
|
|
assert "zero_quantized_gradients" in zopt
|
|
|
|
def test_zero_pp_write_file(self):
|
|
from soup_cli.utils.deepspeed import write_deepspeed_config
|
|
|
|
path = write_deepspeed_config("zero++")
|
|
try:
|
|
assert os.path.exists(path)
|
|
with open(path) as f:
|
|
cfg = json.load(f)
|
|
assert cfg["zero_optimization"]["zero_quantized_weights"] is True
|
|
finally:
|
|
if os.path.exists(path):
|
|
os.unlink(path)
|
|
|
|
def test_resolve_zero_pp_in_train(self):
|
|
from soup_cli.commands.train import _resolve_deepspeed
|
|
|
|
path = _resolve_deepspeed("zero++")
|
|
try:
|
|
assert os.path.exists(path)
|
|
finally:
|
|
if os.path.exists(path):
|
|
os.unlink(path)
|
|
|
|
|
|
# =============================================================================
|
|
# Part D: FSDP2 + torch.compile
|
|
# =============================================================================
|
|
|
|
class TestFsdp2Compile:
|
|
"""Test FSDP2 + torch.compile integration."""
|
|
|
|
def test_config_has_fsdp2_compile_field(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
cfg = TrainingConfig(use_fsdp2_compile=True)
|
|
assert cfg.use_fsdp2_compile is True
|
|
|
|
def test_config_default_false(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
cfg = TrainingConfig()
|
|
assert cfg.use_fsdp2_compile is False
|
|
|
|
def test_validate_compile_requires_fsdp(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=True, fsdp_preset=None, backend="transformers", device="cuda"
|
|
)
|
|
assert any("FSDP" in e for e in errors)
|
|
|
|
def test_validate_compile_requires_cuda(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=True, fsdp_preset="full_shard", backend="transformers", device="cpu"
|
|
)
|
|
assert any("CUDA" in e for e in errors)
|
|
|
|
def test_validate_compile_requires_transformers_backend(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=True, fsdp_preset="full_shard", backend="unsloth", device="cuda"
|
|
)
|
|
assert any("unsloth" in e for e in errors)
|
|
|
|
def test_validate_compile_ok(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
with patch("soup_cli.utils.fsdp.is_fsdp_available", return_value=True):
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=True,
|
|
fsdp_preset="full_shard",
|
|
backend="transformers",
|
|
device="cuda",
|
|
)
|
|
assert errors == []
|
|
|
|
def test_validate_compile_disabled_noop(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
# When flag is off, no errors ever.
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=False, fsdp_preset=None, backend="unsloth", device="cpu"
|
|
)
|
|
assert errors == []
|
|
|
|
|
|
# =============================================================================
|
|
# Part E: DeepSpeed-MII serve backend
|
|
# =============================================================================
|
|
|
|
class TestMiiBackend:
|
|
"""Test DeepSpeed-MII detection + runtime creation."""
|
|
|
|
def test_is_mii_available_false_when_stubbed_to_none(self):
|
|
from soup_cli.utils.mii import is_mii_available
|
|
|
|
with patch.dict("sys.modules", {"mii": None}):
|
|
assert is_mii_available() is False
|
|
|
|
def test_is_mii_available_false_when_key_absent(self):
|
|
"""Cover the ImportError branch — no ``mii`` key in sys.modules at all."""
|
|
from soup_cli.utils.mii import is_mii_available
|
|
|
|
saved = sys.modules.pop("mii", "_SENTINEL")
|
|
try:
|
|
# Force the import path to fail by making the real module look gone.
|
|
with patch.dict("sys.modules", {"mii": None}, clear=False):
|
|
assert is_mii_available() is False
|
|
finally:
|
|
if saved != "_SENTINEL":
|
|
sys.modules["mii"] = saved
|
|
|
|
def test_is_mii_available_true_when_importable(self):
|
|
from soup_cli.utils.mii import is_mii_available
|
|
|
|
fake_mii = MagicMock()
|
|
with patch.dict("sys.modules", {"mii": fake_mii}):
|
|
assert is_mii_available() is True
|
|
|
|
def test_create_mii_pipeline_mocked(self):
|
|
from soup_cli.utils.mii import create_mii_pipeline
|
|
|
|
fake_mii = MagicMock()
|
|
fake_pipe = MagicMock()
|
|
fake_mii.pipeline.return_value = fake_pipe
|
|
with patch.dict("sys.modules", {"mii": fake_mii}):
|
|
pipe = create_mii_pipeline(
|
|
model_path="/models/llama", tensor_parallel=2, max_length=4096
|
|
)
|
|
assert pipe is fake_pipe
|
|
fake_mii.pipeline.assert_called_once()
|
|
kwargs = fake_mii.pipeline.call_args.kwargs
|
|
assert kwargs.get("tensor_parallel") == 2
|
|
|
|
def test_create_mii_raises_when_missing(self):
|
|
from soup_cli.utils.mii import create_mii_pipeline
|
|
|
|
with patch.dict("sys.modules", {"mii": None}):
|
|
with pytest.raises(ImportError, match="deepspeed-mii"):
|
|
create_mii_pipeline(model_path="/models/llama")
|
|
|
|
def test_serve_help_shows_mii(self):
|
|
from typer.testing import CliRunner
|
|
|
|
from soup_cli.cli import app
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(app, ["serve", "--help"])
|
|
import re
|
|
out = re.sub(r"\x1b\[[0-9;]*m", "", result.output)
|
|
out = re.sub(r"\s+", " ", out)
|
|
assert "mii" in out.lower()
|
|
|
|
|
|
# =============================================================================
|
|
# Part F: Pipeline parallelism config
|
|
# =============================================================================
|
|
|
|
class TestPipelineParallelConfig:
|
|
"""Test pipeline-parallel config fields + validation."""
|
|
|
|
def test_default_parallelism_data(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
cfg = TrainingConfig()
|
|
assert cfg.parallelism == "data"
|
|
assert cfg.pipeline_stages == 1
|
|
|
|
def test_set_pipeline(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
cfg = TrainingConfig(parallelism="pipeline", pipeline_stages=4)
|
|
assert cfg.parallelism == "pipeline"
|
|
assert cfg.pipeline_stages == 4
|
|
|
|
def test_pipeline_stages_bounds_low(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
with pytest.raises(ValueError):
|
|
TrainingConfig(pipeline_stages=0)
|
|
|
|
def test_pipeline_stages_bounds_high(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
with pytest.raises(ValueError):
|
|
TrainingConfig(pipeline_stages=64)
|
|
|
|
def test_parallelism_invalid_literal(self):
|
|
from soup_cli.config.schema import TrainingConfig
|
|
|
|
with pytest.raises(ValueError):
|
|
TrainingConfig(parallelism="hybrid")
|
|
|
|
def test_validate_pipeline_requires_stages(self):
|
|
from soup_cli.utils.pipeline import validate_pipeline_config
|
|
|
|
errors = validate_pipeline_config(
|
|
parallelism="pipeline", pipeline_stages=1, device="cuda", gpu_count=4,
|
|
)
|
|
assert any("pipeline_stages" in e for e in errors)
|
|
|
|
def test_validate_pipeline_requires_cuda(self):
|
|
from soup_cli.utils.pipeline import validate_pipeline_config
|
|
|
|
errors = validate_pipeline_config(
|
|
parallelism="pipeline", pipeline_stages=4, device="cpu", gpu_count=0,
|
|
)
|
|
assert any("CUDA" in e for e in errors)
|
|
|
|
def test_validate_pipeline_requires_enough_gpus(self):
|
|
from soup_cli.utils.pipeline import validate_pipeline_config
|
|
|
|
errors = validate_pipeline_config(
|
|
parallelism="pipeline", pipeline_stages=4, device="cuda", gpu_count=2,
|
|
)
|
|
assert any("GPUs" in e for e in errors)
|
|
|
|
def test_validate_data_parallel_ok(self):
|
|
from soup_cli.utils.pipeline import validate_pipeline_config
|
|
|
|
errors = validate_pipeline_config(
|
|
parallelism="data", pipeline_stages=1, device="cpu", gpu_count=0,
|
|
)
|
|
assert errors == []
|
|
|
|
|
|
# =============================================================================
|
|
# Part G: Multi-GPU recipes
|
|
# =============================================================================
|
|
|
|
class TestMultiGpuRecipes:
|
|
"""Test new multi-GPU recipes in catalog."""
|
|
|
|
def test_llama3_70b_fsdp2_recipe_exists(self):
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
assert "llama3-70b-fsdp2" in RECIPES
|
|
|
|
def test_qwen3_32b_zeropp_recipe_exists(self):
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
assert "qwen3-32b-zeropp" in RECIPES
|
|
|
|
def test_deepseek_v3_pipeline_recipe_exists(self):
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
assert "deepseek-v3-pipeline" in RECIPES
|
|
|
|
def test_all_new_recipes_load_as_config(self):
|
|
from soup_cli.config.loader import load_config_from_string
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
for name in ("llama3-70b-fsdp2", "qwen3-32b-zeropp", "deepseek-v3-pipeline"):
|
|
meta = RECIPES[name]
|
|
cfg = load_config_from_string(meta.yaml_str)
|
|
# Exact-match check — substring allowed false positives between
|
|
# ``llama3-8b`` / ``llama3-70b`` recipes.
|
|
assert cfg.base == meta.model
|
|
|
|
def test_llama3_70b_fsdp2_has_fsdp2_compile(self):
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
meta = RECIPES["llama3-70b-fsdp2"]
|
|
doc = yaml.safe_load(meta.yaml_str)
|
|
assert doc["training"].get("use_fsdp2_compile") is True
|
|
|
|
def test_qwen3_32b_zeropp_tags(self):
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
meta = RECIPES["qwen3-32b-zeropp"]
|
|
# Recipe is multi-GPU focused
|
|
assert "qwen3-32" in meta.model.lower()
|
|
|
|
def test_deepseek_v3_pipeline_stages(self):
|
|
from soup_cli.recipes.catalog import RECIPES
|
|
|
|
meta = RECIPES["deepseek-v3-pipeline"]
|
|
doc = yaml.safe_load(meta.yaml_str)
|
|
assert doc["training"].get("parallelism") == "pipeline"
|
|
assert doc["training"].get("pipeline_stages") >= 2
|
|
|
|
|
|
# =============================================================================
|
|
# Integration: config field → trainer wiring
|
|
# =============================================================================
|
|
|
|
class TestFsdp2CompileHelper:
|
|
"""Behavioral tests for apply_fsdp_training_kwargs (extracted helper).
|
|
|
|
Tests the actual dict mutation that reaches ``TrainingArguments`` —
|
|
no source grep, no brittle string matching.
|
|
"""
|
|
|
|
def test_no_fsdp_leaves_kwargs_untouched(self):
|
|
from soup_cli.utils.fsdp import apply_fsdp_training_kwargs
|
|
|
|
kwargs = {"learning_rate": 2e-4}
|
|
result = apply_fsdp_training_kwargs(
|
|
kwargs, fsdp_config=None, use_fsdp2_compile=True
|
|
)
|
|
assert result is kwargs
|
|
assert "torch_compile" not in result
|
|
assert "fsdp" not in result
|
|
|
|
def test_fsdp_without_compile_flag(self):
|
|
from soup_cli.utils.fsdp import apply_fsdp_training_kwargs
|
|
|
|
kwargs = {}
|
|
apply_fsdp_training_kwargs(
|
|
kwargs,
|
|
fsdp_config={"fsdp": "full_shard auto_wrap", "fsdp_config": {}},
|
|
use_fsdp2_compile=False,
|
|
)
|
|
assert kwargs["fsdp"] == "full_shard auto_wrap"
|
|
assert "torch_compile" not in kwargs
|
|
|
|
def test_fsdp_with_compile_flag_true(self):
|
|
from soup_cli.utils.fsdp import apply_fsdp_training_kwargs
|
|
|
|
kwargs = {}
|
|
apply_fsdp_training_kwargs(
|
|
kwargs,
|
|
fsdp_config={"fsdp": "full_shard auto_wrap", "fsdp_config": {}},
|
|
use_fsdp2_compile=True,
|
|
)
|
|
assert kwargs["torch_compile"] is True
|
|
|
|
def test_unexpected_fsdp_keys_rejected(self):
|
|
from soup_cli.utils.fsdp import apply_fsdp_training_kwargs
|
|
|
|
with pytest.raises(ValueError, match="Unexpected FSDP config keys"):
|
|
apply_fsdp_training_kwargs(
|
|
{},
|
|
fsdp_config={"fsdp": "x", "fsdp_config": {}, "shady": 1},
|
|
use_fsdp2_compile=False,
|
|
)
|
|
|
|
|
|
class TestFsdp2CompileValidatorDeepSpeed:
|
|
"""validate_fsdp2_compile_config rejects DeepSpeed + torch.compile."""
|
|
|
|
def test_deepspeed_and_compile_rejected(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=True,
|
|
fsdp_preset="full_shard",
|
|
backend="transformers",
|
|
device="cuda",
|
|
deepspeed_config="/tmp/ds.json",
|
|
)
|
|
assert any("deepspeed" in err.lower() for err in errors)
|
|
|
|
def test_no_deepspeed_no_error(self):
|
|
from soup_cli.utils.fsdp import validate_fsdp2_compile_config
|
|
|
|
with patch("soup_cli.utils.fsdp.is_fsdp_available", return_value=True):
|
|
errors = validate_fsdp2_compile_config(
|
|
use_compile=True,
|
|
fsdp_preset="full_shard",
|
|
backend="transformers",
|
|
device="cuda",
|
|
deepspeed_config=None,
|
|
)
|
|
assert errors == []
|
|
|
|
|
|
class TestNcclEnvApplication:
|
|
"""Verify train.py actually calls os.environ.setdefault in distributed mode
|
|
and skips the call on single-GPU / CPU runs."""
|
|
|
|
def test_setdefault_does_not_overwrite_user_override(self):
|
|
from soup_cli.utils.topology import suggest_nccl_env
|
|
|
|
env = suggest_nccl_env(gpu_count=4, interconnect="nvlink")
|
|
user = {"NCCL_P2P_DISABLE": "1"}
|
|
for key, val in env.items():
|
|
user.setdefault(key, val)
|
|
# User value preserved
|
|
assert user["NCCL_P2P_DISABLE"] == "1"
|
|
# At least one new key added (non-brittle: the exact key set may
|
|
# evolve, we only need the non-overwrite invariant + "something new
|
|
# came in" guarantee).
|
|
new_keys = set(user) - {"NCCL_P2P_DISABLE"}
|
|
assert new_keys, "suggest_nccl_env produced no new keys"
|
|
|
|
def test_single_gpu_returns_empty_env(self):
|
|
"""Single-GPU path returns an empty dict (so setdefault is a no-op)."""
|
|
from soup_cli.utils.topology import suggest_nccl_env
|
|
|
|
assert suggest_nccl_env(gpu_count=1, interconnect="single") == {}
|
|
|
|
|
|
class TestTrainValidatorGating:
|
|
"""CLI-level tests proving the new validators actually block `soup train`
|
|
when misconfigured (not silently continue)."""
|
|
|
|
def test_use_fsdp2_compile_on_cpu_blocks_train(self, tmp_path):
|
|
from typer.testing import CliRunner
|
|
|
|
from soup_cli.cli import app
|
|
|
|
# Build a config that asks for use_fsdp2_compile but runs on CPU (no
|
|
# --fsdp + no --gpus). Validators should fire before anything heavy
|
|
# runs, so we don't need a real model here.
|
|
data_file = tmp_path / "train.jsonl"
|
|
data_file.write_text('{"prompt": "x", "completion": "y"}\n')
|
|
cfg_file = tmp_path / "soup.yaml"
|
|
cfg_file.write_text(
|
|
"base: test-model\n"
|
|
"task: sft\n"
|
|
f"data:\n train: {data_file.as_posix()}\n"
|
|
"training:\n use_fsdp2_compile: true\n"
|
|
f"output: {(tmp_path / 'out').as_posix()}\n"
|
|
)
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
app, ["train", "--config", str(cfg_file), "--yes", "--dry-run"]
|
|
)
|
|
# Exit non-zero, with a clear message mentioning the flag.
|
|
assert result.exit_code != 0, (result.output, repr(result.exception))
|
|
assert "fsdp2_compile" in result.output.lower() or "fsdp" in result.output.lower()
|
|
|
|
def test_pipeline_parallel_on_cpu_blocks_train(self, tmp_path):
|
|
from typer.testing import CliRunner
|
|
|
|
from soup_cli.cli import app
|
|
|
|
data_file = tmp_path / "train.jsonl"
|
|
data_file.write_text('{"prompt": "x", "completion": "y"}\n')
|
|
cfg_file = tmp_path / "soup.yaml"
|
|
cfg_file.write_text(
|
|
"base: test-model\n"
|
|
"task: sft\n"
|
|
f"data:\n train: {data_file.as_posix()}\n"
|
|
"training:\n parallelism: pipeline\n pipeline_stages: 4\n"
|
|
f"output: {(tmp_path / 'out').as_posix()}\n"
|
|
)
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
app, ["train", "--config", str(cfg_file), "--yes", "--dry-run"]
|
|
)
|
|
assert result.exit_code != 0, (result.output, repr(result.exception))
|
|
assert "pipeline" in result.output.lower() or "cuda" in result.output.lower()
|