tracer-cloud--opensre/tests/cli/test_smoke.py

from __future__ import annotations

import errno
import json
import os
import re
import select
import shutil
import site
import subprocess
import sys
import sysconfig
import time
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from threading import Thread

import pytest

from config.constants.paths import REPO_ROOT
from config.version import get_opensre_version

_SCRIPT_NAME = "opensre.exe" if os.name == "nt" else "opensre"
_ANSI_RE = re.compile(r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
_CLEARED_ENV_KEYS = (
    "ANTHROPIC_API_KEY",
    "AWS_ACCESS_KEY_ID",
    "AWS_DEFAULT_REGION",
    "AWS_EXTERNAL_ID",
    "AWS_REGION",
    "AWS_ROLE_ARN",
    "AWS_SECRET_ACCESS_KEY",
    "AWS_SESSION_TOKEN",
    "CORALOGIX_API_KEY",
    "CORALOGIX_API_URL",
    "CORALOGIX_APPLICATION_NAME",
    "CORALOGIX_SUBSYSTEM_NAME",
    "DD_API_KEY",
    "DD_APP_KEY",
    "DD_SITE",
    "GEMINI_API_KEY",
    "GOOGLE_CREDENTIALS_FILE",
    "GOOGLE_DRIVE_FOLDER_ID",
    "GRAFANA_INSTANCE_URL",
    "GRAFANA_READ_TOKEN",
    "HONEYCOMB_API_KEY",
    "HONEYCOMB_API_URL",
    "HONEYCOMB_DATASET",
    "JWT_TOKEN",
    "NVIDIA_API_KEY",
    "OPENAI_API_KEY",
    "OPENROUTER_API_KEY",
    "OPENCLAW_MCP_ARGS",
    "OPENCLAW_MCP_AUTH_TOKEN",
    "OPENCLAW_MCP_COMMAND",
    "OPENCLAW_MCP_MODE",
    "OPENCLAW_MCP_URL",
    "OPENSRE_LLM_AUTH_METADATA_PATH",
    "OPENSRE_PROJECT_ENV_PATH",
    "OPENSRE_RELEASES_API_URL",
    "OPENSRE_WIZARD_STORE_PATH",
    "SLACK_WEBHOOK_URL",
    "TELEGRAM_BOT_TOKEN",
    "TELEGRAM_DEFAULT_CHAT_ID",
    "TRACER_API_URL",
    "TRACER_WEB_APP_URL",
)


@dataclass(frozen=True)
class CliResult:
    args: tuple[str, ...]
    exit_code: int
    stdout: str
    stderr: str = ""


@dataclass(frozen=True)
class PtyAction:
    #: Wait until this substring appears, or any substring if a tuple is given.
    expect: str | tuple[str, ...]
    send: bytes
    timeout: float = 10.0
    #: If > 0, send this many ``stagger_key`` keypresses one at a time
    #: (prompt_toolkit may coalesce a single burst), then send ``send``
    #: (usually ``\\r``).
    stagger_j: int = 0
    #: The single-keypress navigation byte sent ``stagger_j`` times. Defaults to
    #: ``j`` (vim-down). Use ``k`` (vim-up) to reach the last option in one step,
    #: since questionary select menus wrap around.
    stagger_key: bytes = b"j"


@dataclass
class CliSandbox:
    home: Path
    project_env_path: Path
    env: dict[str, str]

    @property
    def integration_store_path(self) -> Path:
        return self.home / ".opensre" / "integrations.json"

    @property
    def wizard_store_path(self) -> Path:
        return self.home / ".opensre" / "opensre.json"

    def seed_integrations(self, integrations: list[dict[str, object]]) -> None:
        self.integration_store_path.parent.mkdir(parents=True, exist_ok=True)
        self.integration_store_path.write_text(
            json.dumps({"version": 1, "integrations": integrations}, indent=2) + "\n",
            encoding="utf-8",
        )

    def seed_wizard_store(
        self,
        *,
        provider: str = "anthropic",
        model: str = "claude-opus-4-7",
    ) -> None:
        self.wizard_store_path.parent.mkdir(parents=True, exist_ok=True)
        self.wizard_store_path.write_text(
            json.dumps(
                {
                    "version": 1,
                    "wizard": {
                        "mode": "quickstart",
                        "configured_target": "local",
                        "updated_at": "2026-01-01T00:00:00+00:00",
                    },
                    "targets": {
                        "local": {
                            "provider": provider,
                            "model": model,
                            "api_key_env": f"{provider.upper()}_API_KEY",
                            "model_env": f"{provider.upper()}_REASONING_MODEL",
                            "updated_at": "2026-01-01T00:00:00+00:00",
                        }
                    },
                    "probes": {"local": {}, "remote": {}},
                },
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )

    def seed_project_env(
        self,
        *,
        provider: str = "anthropic",
        model: str = "claude-opus-4-7",
    ) -> None:
        model_env = f"{provider.upper()}_REASONING_MODEL"
        self.project_env_path.write_text(
            f"LLM_PROVIDER={provider}\n{model_env}={model}\n",
            encoding="utf-8",
        )

    def read_integrations(self) -> list[dict[str, object]]:
        if not self.integration_store_path.exists():
            return []
        payload = json.loads(self.integration_store_path.read_text(encoding="utf-8"))
        return list(payload.get("integrations", []))

    def read_project_env(self) -> str:
        if not self.project_env_path.exists():
            return ""
        return self.project_env_path.read_text(encoding="utf-8")

    def read_wizard_store(self) -> dict[str, object]:
        return json.loads(self.wizard_store_path.read_text(encoding="utf-8"))


def _clean_terminal_output(text: str) -> str:
    if not text:
        return ""
    cleaned = _ANSI_RE.sub("", text)
    cleaned = cleaned.replace("\r", "\n").replace("\x00", "")
    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
    return cleaned


def _opensre_executable() -> Path:
    """Return the ``opensre`` entrypoint from the active test interpreter's venv.

    Prefer the script adjacent to ``sys.executable`` over ``shutil.which`` so CI
    and local xdist workers never invoke a stale global ``opensre`` on ``PATH``
    that predates the ``remote`` command (smoke tests would otherwise see
    ``Error: No such command 'remote'``).
    """
    candidates: list[Path] = [
        Path(sys.executable).with_name(_SCRIPT_NAME),
        Path(sysconfig.get_path("scripts")) / _SCRIPT_NAME,
    ]
    resolved = shutil.which(_SCRIPT_NAME)
    if resolved:
        candidates.append(Path(resolved))
    for candidate in candidates:
        if candidate.exists():
            return candidate
    pytest.skip("installed opensre executable is unavailable in this environment")
    raise AssertionError("pytest.skip should have interrupted control flow")


def _is_python_script(path: Path) -> bool:
    """Return True when an executable should be launched via Python."""
    if path.suffix in {".py", ".pyw"}:
        return True
    try:
        first_line = path.read_text(encoding="utf-8", errors="ignore").splitlines()[0]
    except (OSError, IndexError):
        return False
    return first_line.startswith("#!") and "python" in first_line.lower()


def _cli_env(home: Path, project_env_path: Path) -> dict[str, str]:
    env = os.environ.copy()
    # Blank values block ``load_dotenv(override=False)`` from re-importing the repo
    # ``.env`` when subprocesses run with ``cwd=REPO_ROOT``.
    for key in _CLEARED_ENV_KEYS:
        env[key] = ""

    existing_pythonpath = env.get("PYTHONPATH", "")
    pythonpath_parts = [str(REPO_ROOT)]
    user_site = site.getusersitepackages()
    if user_site:
        pythonpath_parts.append(user_site)
    if existing_pythonpath:
        pythonpath_parts.append(existing_pythonpath)
    env["PYTHONPATH"] = os.pathsep.join(pythonpath_parts)
    env["HOME"] = str(home)
    env["USERPROFILE"] = str(home)
    env["OPENSRE_NO_TELEMETRY"] = "1"
    env["OPENSRE_PROJECT_ENV_PATH"] = str(project_env_path)
    env["PYTHONUTF8"] = "1"
    env["PYTHONIOENCODING"] = "utf-8"
    env["TERM"] = "xterm-256color"
    env.pop("OPENSRE_DISABLE_KEYRING", None)
    env["PYTHON_KEYRING_BACKEND"] = "tests.shared.keyring_backend.MemoryKeyring"
    return env


@pytest.fixture()
def cli_sandbox(tmp_path: Path) -> CliSandbox:
    home = tmp_path / "home"
    home.mkdir()
    project_env_path = tmp_path / "project.env"
    return CliSandbox(
        home=home,
        project_env_path=project_env_path,
        env=_cli_env(home, project_env_path),
    )


def _run_cli(
    sandbox: CliSandbox,
    *args: str,
    timeout: float = 15.0,
    extra_env: dict[str, str] | None = None,
) -> CliResult:
    executable = _opensre_executable()
    command = [str(executable), *args]
    if executable.suffix != ".exe" and _is_python_script(executable):
        command = [sys.executable, str(executable), *args]

    env = sandbox.env.copy()
    if extra_env:
        env.update(extra_env)

    completed = subprocess.run(
        command,
        cwd=REPO_ROOT,
        env=env,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=timeout,
        check=False,
    )
    return CliResult(
        args=tuple(args),
        exit_code=int(completed.returncode),
        stdout=_clean_terminal_output(completed.stdout),
        stderr=_clean_terminal_output(completed.stderr),
    )


def _read_pty_chunk(master_fd: int, timeout: float) -> bytes:
    ready, _, _ = select.select([master_fd], [], [], timeout)
    if not ready:
        return b""
    try:
        return os.read(master_fd, 4096)
    except OSError as exc:
        if exc.errno == errno.EIO:
            return b""
        raise


def _wait_for_output(
    process: subprocess.Popen[bytes],
    master_fd: int,
    buffer: bytearray,
    expected: str | tuple[str, ...],
    *,
    timeout: float,
) -> None:
    def _matches(cleaned: str) -> bool:
        if isinstance(expected, str):
            return expected in cleaned
        return any(sub in cleaned for sub in expected)

    deadline = time.monotonic() + timeout
    while not _matches(_clean_terminal_output(buffer.decode("utf-8", errors="replace"))):
        if time.monotonic() > deadline:
            cleaned = _clean_terminal_output(buffer.decode("utf-8", errors="replace"))
            raise AssertionError(f"Timed out waiting for {expected!r}.\nCurrent output:\n{cleaned}")
        chunk = _read_pty_chunk(master_fd, 0.1)
        if chunk:
            buffer.extend(chunk)
            continue
        if process.poll() is not None:
            break

    cleaned = _clean_terminal_output(buffer.decode("utf-8", errors="replace"))
    if not _matches(cleaned):
        raise AssertionError(
            f"Process exited before showing {expected!r}.\nCurrent output:\n{cleaned}"
        )


def _run_cli_pty(
    sandbox: CliSandbox,
    *args: str,
    actions: list[PtyAction],
    timeout: float = 20.0,
    extra_env: dict[str, str] | None = None,
) -> CliResult:
    executable = _opensre_executable()
    command = [str(executable), *args]
    if executable.suffix != ".exe" and _is_python_script(executable):
        command = [sys.executable, str(executable), *args]

    master_fd, slave_fd = os.openpty()
    env = sandbox.env.copy()
    if extra_env:
        env.update(extra_env)

    process = subprocess.Popen(
        command,
        cwd=REPO_ROOT,
        env=env,
        stdin=slave_fd,
        stdout=slave_fd,
        stderr=slave_fd,
        close_fds=True,
    )
    os.close(slave_fd)

    buffer = bytearray()
    try:
        for action in actions:
            _wait_for_output(process, master_fd, buffer, action.expect, timeout=action.timeout)
            if action.stagger_j:
                for _ in range(action.stagger_j):
                    os.write(master_fd, action.stagger_key)
                    time.sleep(0.05)
            os.write(master_fd, action.send)

        deadline = time.monotonic() + timeout
        while True:
            chunk = _read_pty_chunk(master_fd, 0.1)
            if chunk:
                buffer.extend(chunk)
                continue
            if process.poll() is not None:
                break
            if time.monotonic() > deadline:
                process.kill()
                cleaned = _clean_terminal_output(buffer.decode("utf-8", errors="replace"))
                raise AssertionError(f"Timed out waiting for CLI exit.\nCurrent output:\n{cleaned}")

        for _ in range(5):
            chunk = _read_pty_chunk(master_fd, 0.05)
            if not chunk:
                break
            buffer.extend(chunk)
    finally:
        os.close(master_fd)

    return CliResult(
        args=tuple(args),
        exit_code=int(process.wait(timeout=2.0)),
        stdout=_clean_terminal_output(buffer.decode("utf-8", errors="replace")),
    )


class _ReleaseHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        payload = json.dumps(
            {
                "tag_name": "main-build",
                "body": "- Version: `9999.0.0`\n- Commit: `deadbeef`\n",
            }
        ).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, _format: str, *_args: object) -> None:
        return


@pytest.fixture()
def release_api_url() -> str:
    try:
        server = ThreadingHTTPServer(("127.0.0.1", 0), _ReleaseHandler)
    except OSError as exc:
        if exc.errno in {errno.EPERM, errno.EACCES}:
            pytest.skip("localhost HTTP server binding is not permitted in this environment")
        raise

    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield f"http://127.0.0.1:{server.server_port}/releases/tags/main-build"
    finally:
        server.shutdown()
        thread.join(timeout=5.0)
        server.server_close()


def test_opensre_landing_page_smoke(cli_sandbox: CliSandbox) -> None:
    result = _run_cli(cli_sandbox)

    assert result.exit_code == 0
    assert "Quick start:" in result.stdout
    assert "opensre investigate -i alert.json" in result.stdout


def test_opensre_help_smoke(cli_sandbox: CliSandbox) -> None:
    result = _run_cli(cli_sandbox, "-h")

    assert result.exit_code == 0
    assert "Welcome back" not in result.stdout
    assert "Commands:" in result.stdout
    assert "integrations" in result.stdout
    assert "--interactive / --no-interactive" in result.stdout
    assert "--layout [classic|pinned]" in result.stdout
    assert "update" in result.stdout


def test_opensre_version_smoke(cli_sandbox: CliSandbox) -> None:
    result = _run_cli(cli_sandbox, "--version")

    assert result.exit_code == 0
    assert get_opensre_version() in result.stdout


def test_health_smoke_uses_real_datadog_store_config(cli_sandbox: CliSandbox) -> None:
    cli_sandbox.seed_integrations(
        [
            {
                "id": "datadog-local",
                "service": "datadog",
                "status": "active",
                "credentials": {
                    "api_key": "",
                    "app_key": "",
                    "site": "datadoghq.com",
                },
            }
        ]
    )

    result = _run_cli(cli_sandbox, "health")

    assert result.exit_code == 0
    assert "OpenSRE Health" in result.stdout
    assert "datadog" in result.stdout
    assert "Missing API key or application key." in result.stdout


def test_update_check_smoke_uses_local_stub(cli_sandbox: CliSandbox, release_api_url: str) -> None:
    result = _run_cli(
        cli_sandbox,
        "update",
        "--check",
        extra_env={"OPENSRE_RELEASES_API_URL": release_api_url},
    )

    assert result.exit_code == 1
    assert "current:" in result.stdout
    assert "latest:" in result.stdout
    assert "9999.0.0" in result.stdout


def test_investigate_print_template_smoke(cli_sandbox: CliSandbox) -> None:
    result = _run_cli(cli_sandbox, "investigate", "--print-template", "generic")

    assert result.exit_code == 0
    payload = json.loads(result.stdout)
    assert payload["alert_source"] == "generic"
    assert payload["message"]


def test_investigate_onboard_handoff_smoke(cli_sandbox: CliSandbox, tmp_path: Path) -> None:
    """project.env written by onboard is read by investigate before it reaches the LLM.

    Only the project .env handoff is exercised here — investigate reads
    LLM_PROVIDER and model env vars from that file, not from the wizard store.
    No real API key is required. The test proves the plumbing works by asserting
    the CLI reaches the LLM credential check (exit 1 + ANTHROPIC_API_KEY named
    in the error) rather than crashing in config loading or file parsing.

    LLM_PROVIDER is passed explicitly via extra_env so the assertion holds even
    on CI runners where the variable is set to a different provider in the
    parent environment.
    """
    cli_sandbox.seed_project_env(provider="anthropic", model="claude-opus-4-7")

    alert_path = tmp_path / "alert.json"
    alert_path.write_text(
        json.dumps(
            {
                "alert_name": "High CPU on orders-rds-prod",
                "pipeline_name": "orders",
                "severity": "critical",
                "message": "CPU utilisation exceeded 90% for 5 minutes.",
            }
        ),
        encoding="utf-8",
    )

    result = _run_cli(
        cli_sandbox,
        "investigate",
        "-i",
        str(alert_path),
        extra_env={"LLM_PROVIDER": "anthropic"},
    )

    # Exit 1 is expected — no real API key in CI.
    assert result.exit_code == 1
    # The failure must name the missing credential specifically — not a generic
    # "run opensre onboard" fallback that would also appear when config loading
    # itself is broken (the scenario this test is designed to catch).
    combined = result.stdout + result.stderr
    assert "ANTHROPIC_API_KEY" in combined, (
        f"Expected a missing-credential error naming ANTHROPIC_API_KEY, got:\n{combined}"
    )


def test_integrations_list_and_show_smoke(cli_sandbox: CliSandbox) -> None:
    cli_sandbox.seed_integrations(
        [
            {
                "id": "datadog-local",
                "service": "datadog",
                "status": "active",
                "credentials": {
                    "api_key": "dd-api-key",
                    "app_key": "dd-app-key",
                    "site": "datadoghq.com",
                },
            }
        ]
    )

    list_result = _run_cli(cli_sandbox, "integrations", "list")
    show_result = _run_cli(cli_sandbox, "integrations", "show", "datadog")

    assert list_result.exit_code == 0
    assert "datadog" in list_result.stdout
    assert "datadog-local" in list_result.stdout

    assert show_result.exit_code == 0
    assert '"service": "datadog"' in show_result.stdout
    assert '"api_key": "dd-a****"' in show_result.stdout
    assert '"app_key": "dd-a****"' in show_result.stdout


def test_integrations_verify_datadog_smoke(cli_sandbox: CliSandbox) -> None:
    cli_sandbox.seed_integrations(
        [
            {
                "id": "datadog-local",
                "service": "datadog",
                "status": "active",
                "credentials": {
                    "api_key": "",
                    "app_key": "",
                    "site": "datadoghq.com",
                },
            }
        ]
    )

    result = _run_cli(cli_sandbox, "integrations", "verify", "datadog")

    assert result.exit_code == 1
    assert "datadog" in result.stdout
    assert "Missing API key or application key." in result.stdout


def test_tests_inventory_commands_smoke(cli_sandbox: CliSandbox) -> None:
    list_result = _run_cli(cli_sandbox, "tests", "list", "--category", "ci-safe")
    run_result = _run_cli(cli_sandbox, "tests", "run", "make:test-cov", "--dry-run")

    assert list_result.exit_code == 0
    assert "make:test-cov" in list_result.stdout
    assert "make:test-full" in list_result.stdout

    assert run_result.exit_code == 0
    assert "make test-cov" in run_result.stdout


@pytest.mark.skipif(os.name == "nt", reason="interactive smoke uses POSIX PTYs")
def test_onboard_interactive_smoke(cli_sandbox: CliSandbox) -> None:
    # The select list wraps, and "Skip for now" is always the last option. A single
    # vim-up (`k`) keypress from the first option wraps to it, so this stays correct
    # regardless of how many integrations the picker lists.
    result = _run_cli_pty(
        cli_sandbox,
        "onboard",
        actions=[
            PtyAction(expect="How do you want to get started?", send=b"\r"),
            PtyAction(expect="Choose your LLM provider", send=b"\r"),
            PtyAction(
                expect="Choose Anthropic auth method",
                send=b"\r",
                stagger_j=1,
            ),
            PtyAction(expect="Anthropic API key", send=b"smoke-test-key\r"),
            PtyAction(expect="Choose Anthropic model", send=b"\r"),
            PtyAction(
                expect="Choose an integration to configure",
                send=b"\r",
                stagger_j=1,
                stagger_key=b"k",
            ),
        ],
        timeout=30.0,
        extra_env={"OPENSRE_AUTO_LAUNCH": "0"},
    )

    assert result.exit_code == 0
    assert "Done." in result.stdout
    assert "next" in result.stdout

    store = cli_sandbox.read_wizard_store()
    assert store["targets"]["local"]["provider"] == "anthropic"
    assert "api_key" not in store["targets"]["local"]
    assert "LLM_PROVIDER=anthropic" in cli_sandbox.read_project_env()
    assert "ANTHROPIC_API_KEY=" not in cli_sandbox.read_project_env()
    assert "ANTHROPIC_REASONING_MODEL=" in cli_sandbox.read_project_env()


@pytest.mark.parametrize(
    ("_cli_binary", "provider_key", "provider_label", "uses_oauth", "pty_timeout"),
    [
        pytest.param(
            "codex",
            "openai",
            "OpenAI OAuth",
            True,
            60.0,
            marks=pytest.mark.skipif(
                shutil.which("codex") is None,
                reason="OpenAI Codex CLI not on PATH",
            ),
        ),
        pytest.param(
            "opencode",
            "opencode",
            "OpenCode CLI",
            False,
            120.0,
            marks=pytest.mark.skipif(
                shutil.which("opencode") is None,
                reason="OpenCode CLI not on PATH",
            ),
        ),
    ],
)
@pytest.mark.skipif(os.name == "nt", reason="interactive smoke uses POSIX PTYs")
def test_onboard_interactive_smoke_cli_provider_repick_when_unauthenticated(
    cli_sandbox: CliSandbox,
    _cli_binary: str,
    provider_key: str,
    provider_label: str,
    uses_oauth: bool,
    pty_timeout: float,
) -> None:
    """PTY: quickstart → local CLI LLM → repick when unauthenticated, then finish as Anthropic.

    Navigates from the default provider using the current runtime provider list order.
    Fresh HOME has no CLI auth, so either ``requires login`` or ``Could not verify … login``
    is accepted before choosing repick. Skips when the CLI binary for each parametrized
    case is not on PATH.
    """
    from surfaces.cli.wizard.flow import _onboarding_provider_options

    stagger_j = next(
        (
            i
            for i, provider in enumerate(_onboarding_provider_options())
            if provider.value == provider_key
        ),
        -1,
    )
    assert stagger_j >= 0, f"Provider '{provider_key}' missing from onboarding providers"

    login_prompt: tuple[str, ...] = (
        f"{provider_label} requires login. What next?",
        f"Could not verify {provider_label} login. What next?",
    )
    model_prompt: tuple[str, ...] = (
        f"Choose {provider_label} model",
        "Model",
    )
    actions = [
        PtyAction(expect="How do you want to get started?", send=b"\r"),
        PtyAction(expect="Choose your LLM provider", send=b"\r", stagger_j=stagger_j),
    ]
    if uses_oauth:
        actions.append(PtyAction(expect="Choose OpenAI auth method", send=b"\r"))
    actions.extend(
        [
            PtyAction(expect=model_prompt, send=b"\r", timeout=30.0),
            PtyAction(
                expect=login_prompt,
                send=b"\r",
                stagger_j=2 if uses_oauth else 1,
                timeout=90.0,
            ),
            PtyAction(expect="Choose your LLM provider", send=b"\r"),
            PtyAction(
                expect="Choose Anthropic auth method",
                send=b"\r",
                stagger_j=1,
            ),
            PtyAction(expect="Anthropic API key", send=b"smoke-test-key\r"),
            PtyAction(expect="Choose Anthropic model", send=b"\r"),
            PtyAction(
                expect="Choose an integration to configure",
                send=b"\r",
                stagger_j=1,
                stagger_key=b"k",
            ),
        ]
    )

    try:
        result = _run_cli_pty(
            cli_sandbox,
            "onboard",
            actions=actions,
            timeout=pty_timeout,
            extra_env={
                "OPENSRE_AUTO_LAUNCH": "0",
                "OPENAI_API_KEY": "",
                "OPENAI_ORG_ID": "",
                "OPENAI_PROJECT_ID": "",
                "OPENAI_BASE_URL": "",
            },
        )
    except AssertionError as exc:
        msg = str(exc)
        if (
            _cli_binary == "codex"
            and "Choose OpenAI OAuth model" in msg
            and "requires login" in msg
        ):
            pytest.skip(
                "OpenAI Codex CLI appears already authenticated; unauth repick flow skipped"
            )
        if (
            _cli_binary == "opencode"
            and "environment provider key(s)" in msg
            and "OpenCode:" in msg
        ):
            pytest.skip("OpenCode CLI is already authenticated via env; unauth repick flow skipped")
        raise

    assert result.exit_code == 0
    assert "Done." in result.stdout
    assert "next" in result.stdout

    store = cli_sandbox.read_wizard_store()
    assert store["targets"]["local"]["provider"] == "anthropic"
    assert "api_key" not in store["targets"]["local"]
    env_body = cli_sandbox.read_project_env()
    assert "LLM_PROVIDER=anthropic\n" in env_body
    assert "ANTHROPIC_API_KEY=" not in env_body
    assert "ANTHROPIC_REASONING_MODEL=" in env_body


@pytest.mark.skipif(os.name == "nt", reason="interactive smoke uses POSIX PTYs")
def test_integrations_setup_datadog_interactive_smoke(cli_sandbox: CliSandbox) -> None:
    result = _run_cli_pty(
        cli_sandbox,
        "integrations",
        "setup",
        "datadog",
        actions=[
            PtyAction(expect="API key", send=b"dd-api-key\r"),
            PtyAction(expect="Application key", send=b"dd-app-key\r"),
            PtyAction(expect="Site", send=b"\r"),
        ],
        # Setup runs verify against the Datadog API; CI runners can exceed 20s.
        timeout=45.0,
    )

    assert "Saved" in result.stdout
    # Setup saves credentials then runs verify; placeholder keys fail the Datadog API check.
    assert result.exit_code in (0, 1)

    integrations = cli_sandbox.read_integrations()
    assert len(integrations) == 1
    assert integrations[0]["service"] == "datadog"
    # v2 store shape: credentials live inside the default instance.
    assert integrations[0]["instances"][0]["credentials"]["site"] == "datadoghq.com"


@pytest.mark.skipif(os.name == "nt", reason="interactive smoke uses POSIX PTYs")
def test_integrations_remove_datadog_interactive_smoke(cli_sandbox: CliSandbox) -> None:
    cli_sandbox.seed_integrations(
        [
            {
                "id": "datadog-local",
                "service": "datadog",
                "status": "active",
                "credentials": {
                    "api_key": "dd-api-key",
                    "app_key": "dd-app-key",
                    "site": "datadoghq.com",
                },
            }
        ]
    )

    result = _run_cli_pty(
        cli_sandbox,
        "integrations",
        "remove",
        "datadog",
        actions=[PtyAction(expect="Remove 'datadog'?", send=b"y\r")],
    )

    assert result.exit_code == 0
    assert "Removed 'datadog'." in result.stdout
    assert cli_sandbox.read_integrations() == []


@pytest.mark.skipif(os.name == "nt", reason="interactive smoke uses POSIX PTYs")
def test_tests_interactive_launcher_smoke(cli_sandbox: CliSandbox) -> None:
    # The prompt instruction reads "Esc exit"; Escape is the PTY-safe way to
    # dismiss the prompt in automation (no SIGINT/raw-mode race conditions).
    result = _run_cli_pty(
        cli_sandbox,
        "tests",
        actions=[PtyAction(expect="Choose a test category:", send=b"\x1b")],
    )

    assert result.exit_code == 0
    assert "Choose a test category:" in result.stdout


def test_gateway_help_smoke(cli_sandbox: CliSandbox) -> None:
    result = _run_cli(cli_sandbox, "gateway", "-h")

    assert result.exit_code == 0
    assert "telegram" in result.stdout