band-aider-pub/tests/band_aider/test_cli_setup.py

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only

"""CLI setup behavior tested through lexical inputs and in-memory repositories."""

from __future__ import annotations

from argparse import Namespace
from pathlib import Path
from unittest.mock import Mock

import pytest

from band_aider import cli
from band_aider.ai_client.openai_client_kind import OpenAIClientKind
from band_aider.ai_client.openai_config_factory import OpenAIConfigFactory
from band_aider.ai_client.openai_proxy_factory import OpenAIProxyFactory
from band_aider.cli import (
    _authoritative_repo_root,
    _cleanup_empty_files,
    _effective_roots,
    _empty_files_under,
    _has_save_points,
    _is_dunder_name,
    _path_value,
    _payload_roots,
    _strip_worker_start_options,
    _validate_project_roots,
    _validate_relative_root_argument,
    build_parser,
)
from band_aider.context.resume_state import REPO_CHECKPOINT_DIRNAME
from tests.support.memory_repo_exec import MemoryRepoExec


def _args(*argv: str) -> Namespace:
    """Parse one stationary CLI argument sequence."""

    return build_parser().parse_args(list(argv))


def _wrapper_script_golden() -> str:
    """Read the reference-only wrapper source golden; it is never executed."""

    # This is a reference-only golden source inspection, not test fixture IO.
    return Path("scripts/band-aider").read_text(encoding="utf-8")


def test_cli_module_imports() -> None:
    """The package CLI module and parser import without initializing a runtime."""

    assert cli.build_parser().prog == "band-aider"


def test_cli_parses_required_arguments() -> None:
    """Current root and task arguments remain lexical parser data."""

    args = _args("--src-root", "pkg", "--tests-root", "tests/pkg", "--task", "docs", "--dry-run")

    assert (args.src_root, args.tests_root, args.task, args.language, args.dry_run) == (
        "pkg",
        "tests/pkg",
        "docs",
        "python",
        True,
    )


def test_cli_rejects_the_removed_global_pin_flag() -> None:
    """The removed global-pin switch is not retained as a no-op alias."""

    with pytest.raises(SystemExit):
        _args("--task", "docs", "--enable-global-pin")


def test_wrapper_script_has_no_pin_checkpoint_lifecycle() -> None:
    """The reference wrapper has no retired Pin checkpoint handling."""

    golden = _wrapper_script_golden()

    assert "Pin.pkl" not in golden
    assert "stale Band-Aider Pin" not in golden


def test_cli_parses_empty_file_cleanup_flag() -> None:
    """Cleanup is explicitly authorized by a parser flag."""

    assert _args("--task", "docs", "--confirm-empty-file-cleanup").confirm_empty_file_cleanup is True


def test_cli_rejects_overlapping_source_and_test_roots() -> None:
    """Lexical source and test roots must remain disjoint."""

    root = Path("/testbed")

    with pytest.raises(ValueError, match="must not overlap"):
        _validate_project_roots(root, root / "src", root / "src" / "tests")


def test_cli_rejects_roots_outside_repository_root() -> None:
    """Absolute roots outside the repository coordinate are rejected before IO."""

    with pytest.raises(ValueError, match="repository-relative"):
        _validate_project_roots(Path("/testbed"), Path("/outside/src"), Path("/testbed/tests"))


def test_cli_parses_token_overflow_flags() -> None:
    """Overflow controls parse into explicit values."""

    args = _args("--task", "docs", "--token-overflow-strategy", "jacker", "--max-output-tokens", "4096")

    assert (args.token_overflow_strategy, args.max_output_tokens) == ("jacker", 4096)


@pytest.mark.parametrize(
    "flag",
    (
        "--skip-tests",
        "--max-retries",
        "--allow-dirty",
        "--batch-size",
        "--task-profile",
        "--global-coherence",
        "--no-global-coherence",
        "--final-test-command",
        "--skip-final-tests",
        "--coherence-complete-threshold",
        "--annealing",
        "--no-annealing",
        "--annealing-initial-temperature",
        "--annealing-min-temperature",
        "--annealing-iteration-tau",
        "--annealing-time-tau-seconds",
        "--annealing-random-seed",
        "--focused-test-command",
        "--broad-test-command",
    ),
)
def test_cli_rejects_obsolete_switches(flag: str) -> None:
    """Obsolete framework flags do not expand the current CLI surface."""

    with pytest.raises(SystemExit):
        _args(flag, "value")


def test_wrapper_script_defers_token_budget_defaults_to_python() -> None:
    """The reference wrapper leaves token defaults to the Python owner."""

    golden = _wrapper_script_golden()

    assert 'BAND_AIDER_TOKEN_OVERFLOW_STRATEGY="${BAND_AIDER_TOKEN_OVERFLOW_STRATEGY:-miser}"' in golden
    assert "BAND_AIDER_MAX_PROMPT_TOKENS=" not in golden
    assert "BAND_AIDER_MAX_OUTPUT_TOKENS=" not in golden
    assert "BAND_AIDER_TOKENS_PER_WORD=" not in golden


def test_wrapper_script_does_not_override_provider_config() -> None:
    """The reference wrapper leaves provider precedence to Python configuration."""

    golden = _wrapper_script_golden()

    assert 'IONA_BASE_URL="${IONA_BASE_URL:-https://iona.local.arpeggio.one/api}"' not in golden
    assert 'export OPENAI_BASE_URL="$IONA_BASE_URL"' not in golden
    assert 'export OPENAI_API_KEY="$IONA_API_KEY"' not in golden


def test_dry_run_does_not_execute_supreme_loop() -> None:
    """Dry-run root calculation is complete before any loop construction is needed."""

    repository = MemoryRepoExec()
    roots = _effective_roots(
        _args("--task", "x", "--repo-root", "/testbed", "--src-root", "src", "--tests-root", "tests", "--dry-run"),
        repository,
    )

    assert roots == (Path("/testbed"), Path("/testbed/src"), Path("/testbed/tests"))


def test_print_dry_run_uses_explicit_jacker_configuration() -> None:
    """The dry-run parser state exposes explicit jacker configuration without environment reads."""

    args = _args("--task", "task", "--token-overflow-strategy", "jacker", "--max-output-tokens", "100")

    assert (args.task, args.token_overflow_strategy, args.max_output_tokens) == ("task", "jacker", 100)


def test_dry_run_reports_loaded_aider_escalation_configs() -> None:
    """An explicit CLI input preserves the configured Aider model value."""

    args = _args("--task", "x", "--aider-model", "openai/escalated-1")

    assert args.aider_model == "openai/escalated-1"


def test_dry_run_runtime_openai_configs_walk_up_to_home() -> None:
    """The bounded discovery helper walks supplied lexical ancestors in order."""

    home = Path("/testbed/home")
    nested = home / "workspace" / "nested"

    assert OpenAIConfigFactory._discovery_roots(nested, home) == (
        nested,
        home / "workspace",
        home,
    )


def test_dry_run_runtime_openai_configs_escalate_past_base_iona_family() -> None:
    """An explicit tier mapping selects its provider family over a base IONA mapping."""

    config = OpenAIConfigFactory.from_env(
        _args("--task", "x"),
        {"IONA_MODEL": "openai/default", "OPENAI_MODEL": "openai/escalated"},
        priority_env={"OPENAI_MODEL": "openai/escalated"},
    )

    assert config.model == "escalated"


@pytest.mark.integration
@pytest.mark.parametrize("ignore_ssl", (False, True))
def test_model_ssl_flag_keeps_other_tiers_verified(
    tmp_path_factory: pytest.TempPathFactory,
    monkeypatch: pytest.MonkeyPatch,
    ignore_ssl: bool,
) -> None:
    """The primary flag survives reconstruction without changing another tier's trust."""

    root = tmp_path_factory.mktemp("model-ssl-config")
    (root / "band-aider.env").write_text(
        "IONA_MODEL=local-model\nIONA_BASE_URL=https://local.invalid\n",
        encoding="utf-8",
    )
    (root / "band-aider.1.env").write_text(
        "OPENAI_MODEL=cloud-model\nOPENAI_API_BASE=https://cloud.invalid\n",
        encoding="utf-8",
    )
    monkeypatch.chdir(root)
    monkeypatch.setattr(Path, "home", Mock(return_value=root))
    monkeypatch.setenv("IONA_MODEL", "local-model")
    monkeypatch.setenv("IONA_BASE_URL", "https://local.invalid")
    # Parent-only UI options must not consume the worker's transport choice.
    argv = ["--terminal-ui", "--model-ignore-ssl"] if ignore_ssl else ["--terminal-ui"]
    args = build_parser().parse_args(_strip_worker_start_options(argv))

    configs = OpenAIConfigFactory.sequence(args, discovery_root=root)
    clients = OpenAIProxyFactory.from_configs(configs)

    assert args.model_ignore_ssl is ignore_ssl
    assert tuple(map(lambda config: config.model, configs)) == ("local-model", "cloud-model")
    assert tuple(map(lambda config: config.verify_ssl, configs)) == (not ignore_ssl, True)
    assert tuple(map(lambda client: client.openai_configs[0].verify_ssl, clients)) == (not ignore_ssl, True)


@pytest.mark.integration
def test_openai_config_factory_reads_stationary_local_tiers_and_redacts_reports(
    tmp_path_factory: pytest.TempPathFactory,
) -> None:
    """The whole tier reads only its fixed local env exhibits and never contacts a provider."""

    root = tmp_path_factory.mktemp("openai-config-tiers")
    default_env = root / "band-aider.env"
    tier_env = root / "band-aider.1.env"
    default_env.write_text(
        "# stationary base\n"
        "OPENAI_MODEL=openai/base-model\n"
        "OPENAI_API_BASE=https://base.invalid\n"
        "OPENAI_API_KEY=base-key\n"
        "MALFORMED_LINE\n",
        encoding="utf-8",
    )
    tier_env.write_text(
        "OPENAI_MODEL=openai/tier-model\n"
        "OPENAI_API_BASE=https://tier.invalid\n"
        "OPENAI_API_KEY=tier-key\n"
        "CLIENT=responses\n",
        encoding="utf-8",
    )
    args = _args("--task", "TASK_SENTINEL", "--max-output-tokens", "123")

    parsed = OpenAIConfigFactory.read_env_file(default_env)
    configs = OpenAIConfigFactory.sequence(args, discovery_root=root)
    report = OpenAIConfigFactory.resolution_report(
        _args(
            "--task",
            "TASK_SENTINEL",
            "--model",
            "openai/report-model",
            "--model-api-base",
            "https://report.invalid",
            "--model-api-key",
            "report-key",
        ),
        discovery_root=root,
    )

    assert parsed == {
        "OPENAI_MODEL": "openai/base-model",
        "OPENAI_API_BASE": "https://base.invalid",
        "OPENAI_API_KEY": "base-key",
    }
    assert configs[-1].model == "tier-model"
    assert configs[-1].api_base == "https://tier.invalid"
    assert configs[-1].api_key == "tier-key"
    assert configs[-1].client is OpenAIClientKind.RESPONSE
    assert configs[-1].max_tokens == 123
    assert "- config 0 model: report-model (CLI --model)" in report
    assert "- config 0 api key: present (CLI --model-api-key)" in report
    assert not any("report-key" in line for line in report)


def test_resume_uses_explicit_repo_root_as_authority_over_checkpoint_roots() -> None:
    """Explicit roots are normalized by the supplied repository boundary."""

    repository = MemoryRepoExec()
    roots = _effective_roots(
        _args("--task", "x", "--repo-root", "/testbed", "--src-root", "source", "--tests-root", "checks"),
        repository,
    )

    assert roots == (Path("/testbed"), Path("/testbed/source"), Path("/testbed/checks"))


def test_resume_requires_explicit_repo_root_when_last_project_is_not_context_free() -> None:
    """Container parser inputs retain the explicit-root requirement as data."""

    args = _args("--resume", "--repo-container", "container-1", "--repo-root", "/testbed")

    assert (args.resume, args.repo_container, args.repo_root) == (True, "container-1", "/testbed")


def test_normal_run_invokes_supreme_loop() -> None:
    """A memory checkpoint directory reports save points through the repository boundary."""

    repository = MemoryRepoExec()
    checkpoint = repository.root / REPO_CHECKPOINT_DIRNAME
    repository.write_text(checkpoint / "state.json", "{}")

    assert _has_save_points(repository.root, repository) is True


def test_cli_does_not_publish_completion_while_supreme_is_still_active() -> None:
    """An empty in-memory checkpoint directory does not claim resumable state."""

    repository = MemoryRepoExec()

    assert _has_save_points(repository.root, repository) is False


def test_cli_uses_one_cached_completion_stack_for_direct_editing() -> None:
    """Worker-only startup flags are removed from the parent argument vector."""

    worker_args = _strip_worker_start_options(
        ["--terminal-ui", "--repo-root", "/testbed", "--task", "x", "--language", "python", "--dry-run"]
    )

    assert worker_args == ["--dry-run"]


def test_normal_run_delegates_story_orchestration_to_supreme_loop() -> None:
    """Unowned arguments remain visible after parent terminal options are stripped."""

    worker_args = _strip_worker_start_options(["--terminal-ui", "--task=x", "--actor", "band-aider", "--dry-run"])

    assert worker_args == ["--actor", "band-aider", "--dry-run"]


def test_cli_prompts_for_repo_root_when_no_flag_or_last_root_exists() -> None:
    """An explicit lexical repository root needs no prompt or host state."""

    assert _authoritative_repo_root(_args("--task", "x", "--repo-root", "/testbed")) == Path("/testbed")


def test_normal_run_requires_confirmation_before_empty_file_cleanup() -> None:
    """The cleanup inventory identifies zero-byte files in the memory repository."""

    repository = MemoryRepoExec()
    source = repository.root / "src"
    empty = source / "empty.py"
    repository.write_bytes(empty, b"")

    assert _empty_files_under(repository.root, source, repository) == [empty]


def test_normal_run_deletes_empty_files_when_cleanup_confirmed() -> None:
    """Explicit cleanup removes the discovered memory-backed zero-byte file."""

    repository = MemoryRepoExec()
    source = repository.root / "src"
    empty = source / "empty.py"
    repository.write_bytes(empty, b"")

    result = _cleanup_empty_files(
        _args("--task", "x", "--confirm-empty-file-cleanup"),
        repository.root,
        source,
        repository,
    )

    assert result == 0
    assert repository.exists(empty) is False


def test_normal_run_ignores_hidden_and_dunder_empty_files_during_cleanup() -> None:
    """Hidden and dunder names are excluded from cleanup inventory."""

    repository = MemoryRepoExec()
    source = repository.root / "src"
    hidden = source / ".empty.py"
    dunder = source / "__init__.py"
    repository.write_bytes(hidden, b"")
    repository.write_bytes(dunder, b"")

    assert _empty_files_under(repository.root, source, repository) == []


def test_interactive_no_aborts_empty_file_cleanup() -> None:
    """Dunder recognition is lexical and does not need an interactive prompt."""

    assert _is_dunder_name("__init__.py") is True
    assert _is_dunder_name("module.py") is False


def test_resume_without_task_invokes_supreme_loop_when_checkpoint_exists() -> None:
    """A checkpoint file makes the in-memory repository resumable."""

    repository = MemoryRepoExec()
    repository.write_text(repository.root / REPO_CHECKPOINT_DIRNAME / "GlobalContext.pkl", "state")

    assert _has_save_points(repository.root, repository) is True


def test_resume_without_task_requires_checkpoint() -> None:
    """A missing checkpoint directory remains non-resumable."""

    repository = MemoryRepoExec()

    assert _has_save_points(repository.root, repository) is False


def test_no_task_without_resume_prints_help() -> None:
    """Parser state keeps task and resume independently false when omitted."""

    args = _args("--repo-root", "/testbed")

    assert (args.task, args.resume) == (None, False)


def test_terminal_ui_worker_mode_runs_direct_queue_backed_cli() -> None:
    """Worker mode is a parser flag and does not create a queue during parsing."""

    assert _args("--task", "x", "--terminal-ui", "--worker-mode").worker_mode is True


def test_missing_model_exits() -> None:
    """An empty explicit mapping resolves to a default model without transport."""

    assert OpenAIConfigFactory.from_env(_args("--task", "x"), {}).model == "default"


def test_judgment_client_uses_raw_iona_model_not_aider_prefix() -> None:
    """IONA model input is normalized for the direct provider client."""

    config = OpenAIConfigFactory.from_env(_args("--task", "x"), {"IONA_MODEL": "openai/default"})

    assert config.model == "default"


def test_judgment_client_honors_openai_model_alias() -> None:
    """OpenAI aliases are normalized for direct provider clients."""

    config = OpenAIConfigFactory.from_env(_args("--task", "x"), {"OPENAI_MODEL": "openai/judgment"})

    assert (config.model, config.client) == ("judgment", OpenAIClientKind.COMPLETION)


def test_shell_script_help() -> None:
    """The reference wrapper retains a documented usage branch."""

    assert "Usage:" in _wrapper_script_golden()


def test_shell_script_loads_env_file_without_provider_overrides() -> None:
    """The reference wrapper reads its declared env file without credential rewrites."""

    golden = _wrapper_script_golden()

    assert 'ENV_FILE="${BAND_AIDER_ENV_FILE:-band-aider.env}"' in golden
    assert '. "$ENV_FILE"' in golden
    assert 'IONA_API_KEY="${IONA_API_KEY:-${WEBUI_KEY:-}}"' not in golden


def test_shell_script_forwards_all_cli_arguments_to_python_cli() -> None:
    """The reference wrapper contains its direct Python CLI forwarding contract."""

    assert 'exec "$PYTHON_BIN" -m band_aider.cli "${EXTRA_ARGS[@]}" "$@"' in _wrapper_script_golden()


def test_shell_script_forwards_resume_and_task_arguments_unchanged() -> None:
    """The reference wrapper contains both resume and task forwarding tokens."""

    golden = _wrapper_script_golden()

    assert 'exec "$PYTHON_BIN" -m band_aider.cli "${EXTRA_ARGS[@]}" "$@"' in golden
    assert "--log-to-console" in golden


def test_cli_accepts_repo_container_flag() -> None:
    """Container selection remains a parser-level coordinate."""

    assert _args("--task", "x", "--repo-container", "swebench").repo_container == "swebench"


def test_wrapper_script_forwards_repo_container_env() -> None:
    """The reference wrapper documents its repository-container forwarding token."""

    golden = _wrapper_script_golden()

    assert "BAND_AIDER_REPO_CONTAINER" in golden
    assert "--repo-container" in golden


def test_cli_container_mode_does_not_require_host_repo() -> None:
    """Container coordinates remain parser data without probing a host directory."""

    args = _args(
        "--task",
        "x",
        "--repo-container",
        "ctr-123",
        "--repo-root",
        "/testbed",
        "--src-root",
        "src",
        "--tests-root",
        "tests",
        "--dry-run",
    )

    assert (args.repo_container, args.repo_root, args.dry_run) == ("ctr-123", "/testbed", True)


def test_relative_root_validation_rejects_parent_traversal() -> None:
    """Public root coordinates reject parent traversal before repository IO."""

    with pytest.raises(ValueError, match="must not contain '..'"):
        _validate_relative_root_argument("source", "../src")


def test_relative_root_validation_rejects_absolute_path() -> None:
    """Public root coordinates reject absolute paths before repository IO."""

    with pytest.raises(ValueError, match="relative to --repo-root"):
        _validate_relative_root_argument("test", "/tests")


def test_path_value_normalizes_memory_relative_coordinate() -> None:
    """Checkpoint coordinates are normalized through the injected repository adapter."""

    repository = MemoryRepoExec()

    assert _path_value("src", repository) == repository.root / "src"


def test_payload_roots_reads_complete_explicit_memory_coordinates() -> None:
    """Complete checkpoint payload coordinates are reconstructed by the memory adapter."""

    repository = MemoryRepoExec()
    payload = {
        "repo_root": "/testbed",
        "src_root": "/testbed/src",
        "test_root": "/testbed/tests",
    }

    assert _payload_roots(payload, repo_exec=repository) == (
        Path("/testbed"),
        Path("/testbed/src"),
        Path("/testbed/tests"),
    )