qwen-code/packages/sdk-python/tests/unit/test_validation.py

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any, cast, get_args

import pytest
from qwen_code_sdk.errors import ValidationError
from qwen_code_sdk.types import PermissionMode, QueryOptions, TimeoutOptions
from qwen_code_sdk.validation import validate_query_options

VALID_UUID = "123e4567-e89b-12d3-a456-426614174000"


def test_rejects_resume_with_continue_session() -> None:
    with pytest.raises(ValidationError, match="resume together with continue_session"):
        validate_query_options(
            QueryOptions(
                resume=VALID_UUID,
                continue_session=True,
            )
        )


def test_rejects_session_id_with_resume() -> None:
    with pytest.raises(ValidationError, match="Cannot use session_id with resume"):
        validate_query_options(
            QueryOptions(
                session_id=VALID_UUID,
                resume="223e4567-e89b-12d3-a456-426614174000",
            )
        )


def test_rejects_invalid_session_id() -> None:
    with pytest.raises(ValidationError, match="Invalid session_id"):
        validate_query_options(QueryOptions(session_id="not-a-uuid"))


def test_rejects_invalid_resume() -> None:
    with pytest.raises(ValidationError, match="Invalid resume"):
        validate_query_options(QueryOptions(resume="not-a-uuid"))


@pytest.mark.parametrize(
    "value",
    [
        "{12345678-1234-4234-8234-123456781234}",
        "urn:uuid:12345678-1234-4234-8234-123456781234",
        "12345678123442348234123456781234",
    ],
)
def test_rejects_non_canonical_session_id(value: str) -> None:
    # uuid.UUID() accepts these spellings, but the value is passed to the CLI
    # verbatim as --session-id, so only the canonical 8-4-4-4-12 form works.
    with pytest.raises(ValidationError, match="canonical"):
        validate_query_options(QueryOptions(session_id=value))


@pytest.mark.parametrize(
    "value",
    [
        "{12345678-1234-4234-8234-123456781234}",
        "urn:uuid:12345678-1234-4234-8234-123456781234",
        "12345678123442348234123456781234",
    ],
)
def test_rejects_non_canonical_resume(value: str) -> None:
    with pytest.raises(ValidationError, match="canonical"):
        validate_query_options(QueryOptions(resume=value))


def test_accepts_canonical_session_id_in_either_case() -> None:
    # Case is not part of canonical form; an uppercase UUID is valid input.
    validate_query_options(QueryOptions(session_id=VALID_UUID))
    validate_query_options(QueryOptions(session_id=VALID_UUID.upper()))


@pytest.mark.parametrize(
    "mode",
    get_args(PermissionMode),
)
def test_accepts_valid_permission_modes(mode: str) -> None:
    validate_query_options(QueryOptions.from_mapping({"permission_mode": mode}))


def test_permission_modes_match_core_contract() -> None:
    contract_path = (
        Path(__file__).parents[3] / "core" / "src" / "config" / "approval-modes.json"
    )
    expected = set(json.loads(contract_path.read_text(encoding="utf-8")))

    assert set(get_args(PermissionMode)) == expected


def test_rejects_invalid_permission_mode() -> None:
    with pytest.raises(
        ValidationError,
        match=re.escape(
            "Invalid permission_mode: 'unsafe-mode'. "
            "Expected one of: default, plan, auto-edit, auto, yolo."
        ),
    ):
        validate_query_options(
            QueryOptions.from_mapping({"permission_mode": "unsafe-mode"})
        )


def test_rejects_invalid_auth_type() -> None:
    with pytest.raises(ValidationError, match="Invalid auth_type"):
        validate_query_options(QueryOptions.from_mapping({"auth_type": "custom"}))


@pytest.mark.parametrize(
    "auth_type",
    ["openai", "openai-responses", "anthropic", "qwen-oauth", "gemini", "vertex-ai"],
)
def test_accepts_every_cli_auth_type(auth_type: str) -> None:
    # Pins the SDK's auth-type set against the CLI's --auth-type choices. An
    # auth type the CLI accepts but the SDK omits is rejected here with no
    # other signal, so enumerate the whole set rather than spot-checking one.
    validate_query_options(QueryOptions.from_mapping({"auth_type": auth_type}))


def test_from_mapping_rejects_non_callable_can_use_tool() -> None:
    with pytest.raises(TypeError, match="can_use_tool must be callable"):
        QueryOptions.from_mapping({"can_use_tool": "bad"})


def test_from_mapping_rejects_non_callable_stderr() -> None:
    with pytest.raises(TypeError, match="stderr must be callable"):
        QueryOptions.from_mapping({"stderr": "bad"})


def test_validation_rejects_non_callable_can_use_tool() -> None:
    with pytest.raises(ValidationError, match="can_use_tool must be callable"):
        validate_query_options(QueryOptions(can_use_tool=cast(Any, "bad")))


def test_validation_rejects_non_callable_stderr() -> None:
    with pytest.raises(ValidationError, match="stderr must be callable"):
        validate_query_options(QueryOptions(stderr=cast(Any, "bad")))


def test_from_mapping_rejects_sync_can_use_tool() -> None:
    def can_use_tool(  # type: ignore[no-untyped-def]
        tool_name, tool_input, context
    ):
        return {"behavior": "deny", "message": "bad"}

    with pytest.raises(TypeError, match="can_use_tool must be an async callable"):
        QueryOptions.from_mapping({"can_use_tool": can_use_tool})


def test_validation_rejects_sync_can_use_tool() -> None:
    def can_use_tool(  # type: ignore[no-untyped-def]
        tool_name, tool_input, context
    ):
        return {"behavior": "deny", "message": "bad"}

    with pytest.raises(ValidationError, match="can_use_tool must be an async callable"):
        validate_query_options(QueryOptions(can_use_tool=cast(Any, can_use_tool)))


def test_from_mapping_rejects_can_use_tool_with_wrong_arity() -> None:
    async def can_use_tool(
        tool_name: str,
        tool_input: dict[str, Any],
    ) -> dict[str, str]:
        return {"behavior": "deny"}

    with pytest.raises(
        TypeError, match="can_use_tool must accept exactly 3 positional arguments"
    ):
        QueryOptions.from_mapping({"can_use_tool": can_use_tool})


def test_validation_rejects_can_use_tool_with_wrong_arity() -> None:
    async def can_use_tool(
        tool_name: str,
        tool_input: dict[str, Any],
    ) -> dict[str, str]:
        return {"behavior": "deny"}

    with pytest.raises(
        ValidationError,
        match="can_use_tool must accept exactly 3 positional arguments",
    ):
        validate_query_options(QueryOptions(can_use_tool=cast(Any, can_use_tool)))


def test_from_mapping_rejects_stderr_with_wrong_arity() -> None:
    def stderr() -> None:
        return None

    with pytest.raises(
        TypeError, match="stderr must accept exactly 1 positional argument"
    ):
        QueryOptions.from_mapping({"stderr": stderr})


def test_validation_rejects_stderr_with_wrong_arity() -> None:
    def stderr() -> None:
        return None

    with pytest.raises(
        ValidationError, match="stderr must accept exactly 1 positional argument"
    ):
        validate_query_options(QueryOptions(stderr=cast(Any, stderr)))


def test_rejects_invalid_max_session_turns() -> None:
    with pytest.raises(ValidationError, match="max_session_turns"):
        validate_query_options(QueryOptions(max_session_turns=-2))


def test_rejects_fractional_max_session_turns() -> None:
    with pytest.raises(ValidationError, match="max_session_turns"):
        validate_query_options(QueryOptions(max_session_turns=cast(Any, 0.5)))


@pytest.mark.parametrize("value", [True, False])
def test_rejects_bool_max_session_turns(value: bool) -> None:
    with pytest.raises(ValidationError, match="max_session_turns"):
        validate_query_options(QueryOptions(max_session_turns=cast(Any, value)))


def test_rejects_empty_qwen_executable_path() -> None:
    with pytest.raises(
        ValidationError, match="path_to_qwen_executable cannot be empty"
    ):
        validate_query_options(QueryOptions(path_to_qwen_executable="   "))


def test_timeout_rejects_non_numeric_value() -> None:
    with pytest.raises(TypeError, match=r"timeout\.can_use_tool must be a positive"):
        TimeoutOptions.from_mapping({"can_use_tool": "fast"})


def test_timeout_rejects_negative_value() -> None:
    pattern = r"timeout\.control_request must be a positive"
    with pytest.raises(ValueError, match=pattern):
        TimeoutOptions.from_mapping({"control_request": -1})


def test_timeout_rejects_boolean_value() -> None:
    with pytest.raises(TypeError, match=r"timeout\.stream_close must be a positive"):
        TimeoutOptions.from_mapping({"stream_close": True})


def test_rejects_invalid_max_tool_calls() -> None:
    with pytest.raises(ValidationError, match="max_tool_calls"):
        validate_query_options(QueryOptions(max_tool_calls=-2))


def test_rejects_invalid_max_subagent_depth() -> None:
    with pytest.raises(ValidationError, match="max_subagent_depth"):
        validate_query_options(QueryOptions(max_subagent_depth=0))


@pytest.mark.parametrize("value", [True, False, 0.5])
def test_rejects_non_integer_max_tool_calls(value: object) -> None:
    # The error message promises "-1 or a non-negative integer", and the value
    # is stringified straight onto the CLI, so `True` would become
    # `--max-tool-calls True`.
    with pytest.raises(ValidationError, match="max_tool_calls"):
        validate_query_options(QueryOptions(max_tool_calls=cast(Any, value)))


@pytest.mark.parametrize("value", [True, False, 2.5])
def test_rejects_non_integer_max_subagent_depth(value: object) -> None:
    with pytest.raises(ValidationError, match="max_subagent_depth"):
        validate_query_options(QueryOptions(max_subagent_depth=cast(Any, value)))


def test_accepts_valid_integer_limits() -> None:
    # The in-range integers these options are documented to take must survive.
    validate_query_options(
        QueryOptions(max_tool_calls=-1, max_session_turns=-1, max_subagent_depth=1)
    )
    validate_query_options(
        QueryOptions(max_tool_calls=0, max_session_turns=0, max_subagent_depth=100)
    )


def test_rejects_agents_missing_required_fields() -> None:
    with pytest.raises(ValidationError, match="missing required field"):
        validate_query_options(QueryOptions(agents=[{"name": "test"}]))


def test_rejects_extra_args_with_reserved_flags() -> None:
    with pytest.raises(ValidationError, match="reserved flag"):
        validate_query_options(QueryOptions(extra_args=["--input-format"]))


@pytest.mark.parametrize(
    "flag",
    [
        "--model",
        "-m",
        "--auth-type",
        "--approval-mode",
        "--insecure",
        "--yolo",
        "-y",
        "--allowed-tools",
        "--exclude-tools",
        "--resume",
        "-r",
        "--continue",
        "-c",
        "--session-id",
        "--proxy",
        "--channel",
        "--output-format",
        "-o",
        "--openai-base-url",
        "--openai-api-key",
        "--mcp-config",
        "--prompt",
        "-p",
        "--prompt-interactive",
        "-i",
        "--add-dir",
        "--input-file",
        "--extensions",
        "-e",
        "--sandbox",
        "-s",
        "--no-sandbox",
        "--no-insecure",
        "--no-safe-mode",
        "--sandbox-image",
        "--fork-session",
        "--max-tool-calls",
        "--max-subagent-depth",
        "--max-session-turns",
        "--system-prompt",
        "--append-system-prompt",
        "--include-directories",
        "--allowed-mcp-server-names",
        "--disabled-slash-commands",
        "--include-partial-messages",
        "--chat-recording",
        "--openai-logging",
        "--openai-logging-dir",
        "--json-schema",
        "--json-fd",
        "--json-file",
    ],
)
def test_rejects_extra_args_with_security_sensitive_flags(flag: str) -> None:
    with pytest.raises(ValidationError, match="reserved flag"):
        validate_query_options(QueryOptions(extra_args=[flag]))


@pytest.mark.parametrize(
    "flag",
    [
        "--model=qwen-max",
        "--auth-type=openai",
        "--approval-mode=yolo",
        "--insecure=true",
        "--yolo=true",
        "--proxy=http://localhost:8080",
    ],
)
def test_rejects_extra_args_with_flag_value_syntax(flag: str) -> None:
    with pytest.raises(ValidationError, match="reserved flag"):
        validate_query_options(QueryOptions(extra_args=[flag]))


def test_accepts_extra_args_with_non_reserved_flags() -> None:
    validate_query_options(QueryOptions(extra_args=["--some-unknown-flag", "--value"]))


def test_rejects_fallback_model_exceeding_max() -> None:
    with pytest.raises(ValidationError, match="fallback_model supports a maximum of 3"):
        validate_query_options(QueryOptions(fallback_model=["a", "b", "c", "d"]))


def test_rejects_empty_proxy() -> None:
    with pytest.raises(ValidationError, match="proxy cannot be empty"):
        validate_query_options(QueryOptions(proxy="   "))


def test_rejects_fork_session_without_resume() -> None:
    with pytest.raises(ValidationError, match="fork_session requires resume"):
        validate_query_options(QueryOptions(fork_session=True))


def test_accepts_fork_session_with_resume() -> None:
    validate_query_options(
        QueryOptions(
            fork_session=True,
            resume="123e4567-e89b-12d3-a456-426614174000",
        )
    )


def test_rejects_invalid_effort() -> None:
    with pytest.raises(ValidationError, match="Invalid effort"):
        validate_query_options(QueryOptions(effort="invalid"))  # type: ignore[arg-type]


def test_accepts_valid_effort() -> None:
    for effort in ("low", "medium", "high", "xhigh", "max"):
        validate_query_options(QueryOptions(effort=effort))  # type: ignore[arg-type]


@pytest.mark.parametrize(
    "field_name",
    [
        "include_directories",
        "extensions",
        "allowed_mcp_server_names",
        "disabled_slash_commands",
        "fallback_model",
    ],
)
def test_rejects_comma_in_list_fields(field_name: str) -> None:
    with pytest.raises(ValidationError, match="cannot contain commas"):
        validate_query_options(QueryOptions(**{field_name: ["valid", "invalid,comma"]}))


def test_from_mapping_parses_all_new_fields() -> None:
    opts = QueryOptions.from_mapping(
        {
            "fork_session": True,
            "resume": VALID_UUID,
            "max_tool_calls": 50,
            "max_subagent_depth": 3,
            "include_directories": ["/dir1", "/dir2"],
            "extra_args": ["--verbose"],
            "extensions": ["ext1"],
            "allowed_mcp_server_names": ["server1"],
            "fallback_model": ["model-a", "model-b"],
            "proxy": "http://proxy:8080",
            "sandbox": True,
            "safe_mode": True,
            "insecure": True,
            "worktree": True,
            "disabled_slash_commands": ["/cmd1"],
            "agents": [{"name": "a", "description": "b", "systemPrompt": "c"}],
        }
    )
    assert opts.fork_session is True
    assert opts.resume == VALID_UUID
    assert opts.max_tool_calls == 50
    assert opts.max_subagent_depth == 3
    assert opts.include_directories == ["/dir1", "/dir2"]
    assert opts.extra_args == ["--verbose"]
    assert opts.extensions == ["ext1"]
    assert opts.allowed_mcp_server_names == ["server1"]
    assert opts.fallback_model == ["model-a", "model-b"]
    assert opts.proxy == "http://proxy:8080"
    assert opts.sandbox is True
    assert opts.safe_mode is True
    assert opts.insecure is True
    assert opts.worktree is True
    assert opts.disabled_slash_commands == ["/cmd1"]
    assert opts.agents == [{"name": "a", "description": "b", "systemPrompt": "c"}]


def test_from_mapping_defaults_new_fields_to_none() -> None:
    opts = QueryOptions.from_mapping({})
    assert opts.fork_session is False
    assert opts.max_tool_calls is None
    assert opts.max_subagent_depth is None
    assert opts.include_directories is None
    assert opts.extra_args is None
    assert opts.extensions is None
    assert opts.allowed_mcp_server_names is None
    assert opts.fallback_model is None
    assert opts.proxy is None
    assert opts.sandbox is False
    assert opts.safe_mode is False
    assert opts.insecure is False
    assert opts.worktree is False
    assert opts.disabled_slash_commands is None
    assert opts.agents is None