lightseekorg--tokenspeed/test/cli/test_argsplit.py

# Copyright (c) 2026 LightSeek Foundation
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from __future__ import annotations

import pytest

from tokenspeed.cli._argsplit import SplitResult, split_argv


def _split(argv: list[str]) -> SplitResult:
    return split_argv(argv)


def test_orchestrator_only_flags_are_consumed():
    r = _split(["--engine-startup-timeout", "300"])
    assert r.engine == []
    assert r.gateway == []
    assert r.opts.engine_startup_timeout == 300


def test_orchestrator_default_timeouts():
    r = _split([])
    assert r.opts.engine_startup_timeout == 1800
    assert r.opts.gateway_startup_timeout == 60
    assert r.opts.drain_timeout == 30


def test_model_fans_out_to_both():
    r = _split(["--model", "/models/qwen3-4b"])
    assert "--model" in r.engine and "/models/qwen3-4b" in r.engine
    assert "--model" in r.gateway and "/models/qwen3-4b" in r.gateway


def test_model_path_alias_also_fans_out():
    """``--model-path`` is accepted as an alias and normalizes to ``--model``."""
    r = _split(["--model-path", "/models/qwen3-4b"])
    assert "--model" in r.engine and "/models/qwen3-4b" in r.engine
    assert "--model" in r.gateway and "/models/qwen3-4b" in r.gateway
    # The alias name must not leak through.
    assert "--model-path" not in r.engine
    assert "--model-path" not in r.gateway


def test_user_host_port_go_to_gateway_only():
    r = _split(["--host", "0.0.0.0", "--port", "8000"])
    assert r.engine == []  # engine binds 127.0.0.1:<auto> internally
    assert r.gateway == ["--host", "0.0.0.0", "--port", "8000"]


def test_chat_template_overrides_to_gateway_only():
    """Both prepare_server_args and smg accept this flag; spec says gateway."""
    r = _split(["--chat-template", "/some/template.jinja"])
    assert r.engine == []
    assert r.gateway == ["--chat-template", "/some/template.jinja"]


def test_tool_call_parser_overrides_to_gateway_only():
    r = _split(["--tool-call-parser", "hermes"])
    assert r.engine == []
    assert r.gateway == ["--tool-call-parser", "hermes"]


def test_reasoning_parser_fans_out_to_both():
    # Gateway uses it for post-gen reasoning parsing; engine uses it to
    # defer JSON grammars past the reasoning channel.
    r = _split(["--reasoning-parser", "qwen3"])
    assert r.engine == ["--reasoning-parser", "qwen3"]
    assert r.gateway == ["--reasoning-parser", "qwen3"]


def test_tp_alias_routes_to_engine():
    r = _split(["--tp", "2"])
    assert r.engine == ["--tensor-parallel-size", "2"]
    assert r.gateway == []


def test_tensor_parallel_size_routes_to_engine():
    r = _split(["--tensor-parallel-size", "4"])
    assert r.engine == ["--tensor-parallel-size", "4"]
    assert r.gateway == []


def test_engine_only_flag_routes_to_engine():
    """Anything prepare_server_args accepts but smg does not."""
    r = _split(["--sampling-backend", "flashinfer"])
    assert r.engine == ["--sampling-backend", "flashinfer"]
    assert r.gateway == []


def test_dp_sampling_routes_to_engine():
    r = _split(["--dp-sampling"])
    assert r.engine == ["--dp-sampling"]
    assert r.gateway == []


def test_unknown_flag_falls_through_to_gateway():
    """smg's clap is the final authority on these."""
    r = _split(["--policy", "cache_aware"])
    assert r.engine == []
    assert r.gateway == ["--policy", "cache_aware"]


def test_combined_real_invocation():
    argv = [
        "--model",
        "Qwen/Qwen3-30B-A3B",
        "--port",
        "8000",
        "--tp",
        "2",
        "--tool-call-parser",
        "hermes",
        "--reasoning-parser",
        "qwen3",
        "--sampling-backend",
        "flashinfer",
        "--policy",
        "cache_aware",
    ]
    r = _split(argv)
    # Engine: --model (fan-out), --tp (normalized), --reasoning-parser
    # (fan-out), --sampling-backend. Argv order is preserved.
    assert r.engine == [
        "--model",
        "Qwen/Qwen3-30B-A3B",
        "--tensor-parallel-size",
        "2",
        "--reasoning-parser",
        "qwen3",
        "--sampling-backend",
        "flashinfer",
    ]
    # Gateway: --model (fan-out), --port, parsers, --policy.
    assert r.gateway == [
        "--model",
        "Qwen/Qwen3-30B-A3B",
        "--port",
        "8000",
        "--tool-call-parser",
        "hermes",
        "--reasoning-parser",
        "qwen3",
        "--policy",
        "cache_aware",
    ]


def test_equals_form_is_normalized():
    r = _split(["--model=/models/qwen3", "--port=8000"])
    assert r.engine == ["--model", "/models/qwen3"]
    assert r.gateway == ["--model", "/models/qwen3", "--port", "8000"]


def test_valueless_flag_followed_by_equals_form():
    """Regression: --foo --bar=x must not glue =x to --bar's name."""
    r = _split(["--enable-foo", "--port=8000"])
    # --enable-foo is unknown to engine introspection, so falls through.
    assert r.gateway == ["--enable-foo", "--port", "8000"]


def test_valueless_flag_followed_by_value_flag():
    """Regression: --foo --bar value must not lose 'value'."""
    r = _split(["--enable-foo", "--port", "8000"])
    assert r.gateway == ["--enable-foo", "--port", "8000"]


def test_timeout_with_non_integer_raises():
    with pytest.raises(ValueError, match="not a valid integer"):
        _split(["--engine-startup-timeout", "abc"])


def test_timeout_with_empty_value_raises():
    with pytest.raises(ValueError, match="requires a positive integer"):
        _split(["--engine-startup-timeout="])


def test_timeout_with_zero_raises():
    with pytest.raises(ValueError, match="must be positive"):
        _split(["--engine-startup-timeout", "0"])


def test_leading_positional_is_treated_as_model():
    r = _split(
        [
            "openai/gpt-oss-20b",
            "--host",
            "0.0.0.0",
            "--port",
            "8000",
            "--tensor-parallel-size",
            "1",
        ]
    )
    assert "--model" in r.engine and "openai/gpt-oss-20b" in r.engine
    assert "--model" in r.gateway and "openai/gpt-oss-20b" in r.gateway
    assert r.gateway[r.gateway.index("--host") + 1] == "0.0.0.0"
    assert r.gateway[r.gateway.index("--port") + 1] == "8000"
    assert r.engine[r.engine.index("--tensor-parallel-size") + 1] == "1"


def test_positional_model_matches_explicit_model_flag():
    """Both invocation styles must produce the same split."""
    positional = _split(["openai/gpt-oss-20b", "--host", "0.0.0.0"])
    explicit = _split(["--model", "openai/gpt-oss-20b", "--host", "0.0.0.0"])
    assert positional == explicit


def test_positional_and_model_flag_conflict_raises():
    with pytest.raises(ValueError, match="positional argument and via"):
        _split(["openai/gpt-oss-20b", "--model", "other/model"])


def test_positional_and_model_path_alias_conflict_raises():
    with pytest.raises(ValueError, match="positional argument and via"):
        _split(["openai/gpt-oss-20b", "--model-path", "other/model"])


def test_positional_and_model_equals_form_conflict_raises():
    with pytest.raises(ValueError, match="positional argument and via"):
        _split(["openai/gpt-oss-20b", "--model=other/model"])


def test_trailing_positional_still_raises():
    """Only the FIRST argument may be positional; later ones remain illegal."""
    with pytest.raises(ValueError, match="unexpected positional arg"):
        _split(["--host", "0.0.0.0", "stray-positional"])