/tests/config/composable_parallel/test_apply.py

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Tests for applying strategy specs onto merged stage configs."""

from __future__ import annotations

import pytest

from vllm_omni.config.composable_parallel import (
    Broadcast,
    FanInByStage,
    MeshAxisSpec,
    RouteByStage,
    StrategyApplyError,
    StrategySpec,
    TakeRank,
    apply_strategy_specs,
)
from vllm_omni.config.stage_config import StageConfig

pytestmark = [pytest.mark.core_model, pytest.mark.cpu]


def _tp(size: int) -> StrategySpec:
    return StrategySpec("tp", MeshAxisSpec("tp", size), Broadcast(), TakeRank())


def _stage_replica(size: int, policy: str = "round_robin") -> StrategySpec:
    return StrategySpec("stage_replica", MeshAxisSpec("stage_replica", size), RouteByStage(policy), FanInByStage())


def _stage(stage_id: int, model_stage: str, engine_args=None, runtime=None) -> StageConfig:
    return StageConfig(
        stage_id=stage_id,
        model_stage=model_stage,
        yaml_engine_args=dict(engine_args or {}),
        yaml_runtime=dict(runtime or {"num_replicas": 1}),
    )


def _qwen_stages() -> list[StageConfig]:
    return [
        _stage(0, "thinker"),
        _stage(1, "talker"),
        _stage(2, "code2wav"),
    ]


def test_apply_tp_by_role():
    stages = _qwen_stages()
    apply_strategy_specs(stages, {"thinker": [_tp(2)]})
    assert stages[0].yaml_engine_args["tensor_parallel_size"] == 2
    # untouched roles keep their config
    assert "tensor_parallel_size" not in stages[1].yaml_engine_args


def test_apply_by_model_stage():
    stages = _qwen_stages()
    apply_strategy_specs(stages, {"thinker": [_tp(2)]})
    assert stages[0].yaml_engine_args["tensor_parallel_size"] == 2


def test_apply_stage_replica_sets_num_replicas_and_surfaces_lb():
    stages = _qwen_stages()
    result = apply_strategy_specs(stages, {"talker": [_stage_replica(2, "round_robin")]})
    assert stages[1].yaml_runtime["num_replicas"] == 2
    assert result.omni_lb_policy == "round-robin"


def test_only_declared_axes_are_written():
    stages = _qwen_stages()
    # strategy declares only stage_replica -> tp must not be forced.
    apply_strategy_specs(stages, {"talker": [_stage_replica(2)]})
    assert "tensor_parallel_size" not in stages[1].yaml_engine_args


def test_conflict_on_explicit_tp():
    stages = _qwen_stages()
    stages[0].yaml_engine_args["tensor_parallel_size"] = 4
    with pytest.raises(StrategyApplyError):
        apply_strategy_specs(stages, {"thinker": [_tp(2)]})


def test_equal_explicit_value_is_noop():
    stages = _qwen_stages()
    stages[0].yaml_engine_args["tensor_parallel_size"] = 2
    apply_strategy_specs(stages, {"thinker": [_tp(2)]})
    assert stages[0].yaml_engine_args["tensor_parallel_size"] == 2


def test_explicit_none_conflicts_with_derived_value():
    # An explicit YAML ``null`` (``tensor_parallel_size: null``) is a *present*
    # value, not a missing key, so a strategy deriving a non-None size must raise
    # rather than silently clobber the explicit None.
    stages = _qwen_stages()
    stages[0].yaml_engine_args["tensor_parallel_size"] = None
    with pytest.raises(StrategyApplyError):
        apply_strategy_specs(stages, {"thinker": [_tp(2)]})


def test_missing_key_is_filled_without_conflict():
    # A genuinely absent key (never set in the YAML) is filled by the strategy
    # and must NOT raise — the contrast case to an explicit None.
    stages = _qwen_stages()
    assert "tensor_parallel_size" not in stages[0].yaml_engine_args
    apply_strategy_specs(stages, {"thinker": [_tp(2)]})
    assert stages[0].yaml_engine_args["tensor_parallel_size"] == 2


def test_num_replicas_conflict():
    stages = _qwen_stages()
    stages[1].yaml_runtime["num_replicas"] = 3
    with pytest.raises(StrategyApplyError):
        apply_strategy_specs(stages, {"talker": [_stage_replica(2)]})


def test_device_count_ok_template():
    stages = _qwen_stages()
    stages[0].yaml_runtime["devices"] = "0,1"
    apply_strategy_specs(stages, {"thinker": [_tp(2)]})
    assert stages[0].yaml_engine_args["tensor_parallel_size"] == 2


def test_device_count_ok_pool():
    # tp=2 -> world=2; 2 replicas -> pool of 4 device ids is valid.
    stages = _qwen_stages()
    stages[1].yaml_runtime["devices"] = "0,1,2,3"
    apply_strategy_specs(stages, {"talker": [_tp(2), _stage_replica(2)]})
    assert stages[1].yaml_runtime["num_replicas"] == 2


def test_device_count_mismatch():
    stages = _qwen_stages()
    stages[0].yaml_runtime["devices"] = "0,1,2"
    with pytest.raises(StrategyApplyError):
        apply_strategy_specs(stages, {"thinker": [_tp(2)]})


def test_unknown_role_raises():
    stages = _qwen_stages()
    with pytest.raises(StrategyApplyError):
        apply_strategy_specs(stages, {"nonexistent": [_tp(2)]})


def test_conflicting_lb_policy_across_roles():
    stages = _qwen_stages()
    with pytest.raises(StrategyApplyError):
        apply_strategy_specs(
            stages,
            {
                "talker": [_stage_replica(2, "round_robin")],
                "code2wav": [_stage_replica(2, "least_queue")],
            },
        )