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

from __future__ import annotations

from unittest.mock import patch

from click.testing import CliRunner

from integrations.cli import _HANDLERS, _setup_openclaw, _setup_smtp, _setup_vercel
from surfaces.cli.__main__ import cli
from surfaces.cli.constants import SETUP_SERVICES, VERIFY_SERVICES


def test_integrations_show_redacts_api_token() -> None:
    runner = CliRunner()

    with patch(
        "integrations.cli.get_integration",
        return_value={
            "id": "vercel-1234",
            "service": "vercel",
            "status": "active",
            "credentials": {
                "api_token": "vcp_sensitive_token_value",
                "team_id": "team_123",
            },
        },
    ):
        result = runner.invoke(cli, ["integrations", "show", "vercel"])

    assert result.exit_code == 0
    assert "vcp_****" in result.output
    assert "vcp_sensitive_token_value" not in result.output


def test_integrations_setup_accepts_github() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified"),
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        mock_setup.return_value = "github"
        result = runner.invoke(cli, ["integrations", "setup", "github"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("github")
    mock_verify.assert_called_once_with("github")


def test_integrations_setup_accepts_vercel() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=1) as mock_verify,
    ):
        mock_setup.return_value = "vercel"
        result = runner.invoke(cli, ["integrations", "setup", "vercel"])

    assert result.exit_code == 1
    mock_setup.assert_called_once_with("vercel")
    mock_verify.assert_called_once_with("vercel")
    mock_capture.assert_not_called()


def test_integrations_setup_accepts_openclaw() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=1) as mock_verify,
    ):
        mock_setup.return_value = "openclaw"
        result = runner.invoke(cli, ["integrations", "setup", "openclaw"])

    assert result.exit_code == 1
    mock_setup.assert_called_once_with("openclaw")
    mock_verify.assert_called_once_with("openclaw")
    mock_capture.assert_not_called()


def test_setup_vercel_saves_credentials(monkeypatch) -> None:
    answers = iter(["vcp_test_token", "team_123"])

    def fake_p(_label: str, default: str = "", secret: bool = False) -> str:
        return next(answers)

    saved: list[tuple[str, dict[str, object]]] = []
    monkeypatch.setattr("integrations.cli._p", fake_p)
    monkeypatch.setattr(
        "integrations.cli.upsert_integration",
        lambda service, entry: saved.append((service, entry)),
    )

    _setup_vercel()

    assert _HANDLERS["vercel"] is _setup_vercel
    assert saved == [
        (
            "vercel",
            {"credentials": {"api_token": "vcp_test_token", "team_id": "team_123"}},
        )
    ]


def test_setup_openclaw_saves_credentials(monkeypatch) -> None:
    answers = iter(["openclaw", "mcp serve"])

    def fake_p(_label: str, default: str = "", secret: bool = False) -> str:
        return next(answers)

    saved: list[tuple[str, dict[str, object]]] = []
    monkeypatch.setattr("integrations.cli._p", fake_p)
    monkeypatch.setattr(
        "integrations.cli.upsert_integration",
        lambda service, entry: saved.append((service, entry)),
    )
    monkeypatch.setattr(
        "integrations.cli.validate_openclaw_config",
        lambda _config: type("Result", (), {"ok": True, "detail": "ok"})(),
    )

    _setup_openclaw()

    assert _HANDLERS["openclaw"] is _setup_openclaw
    assert saved == [
        (
            "openclaw",
            {
                "credentials": {
                    "mode": "stdio",
                    "command": "openclaw",
                    "args": ["mcp", "serve"],
                    "url": "",
                    "auth_token": "",
                }
            },
        )
    ]


def test_integrations_setup_accepts_telegram() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified"),
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        mock_setup.return_value = "telegram"
        result = runner.invoke(cli, ["integrations", "setup", "telegram"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("telegram")
    mock_verify.assert_called_once_with("telegram")


def test_integrations_setup_accepts_whatsapp() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified"),
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        mock_setup.return_value = "whatsapp"
        result = runner.invoke(cli, ["integrations", "setup", "whatsapp"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("whatsapp")
    mock_verify.assert_called_once_with("whatsapp")


def test_integrations_setup_accepts_twilio() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified"),
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        mock_setup.return_value = "twilio"
        result = runner.invoke(cli, ["integrations", "setup", "twilio"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("twilio")
    mock_verify.assert_called_once_with("twilio")


def test_integrations_setup_accepts_smtp() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified"),
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        mock_setup.return_value = "smtp"
        result = runner.invoke(cli, ["integrations", "setup", "smtp"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("smtp")
    mock_verify.assert_called_once_with("smtp")


def test_setup_smtp_saves_credentials(monkeypatch) -> None:
    answers = iter(
        [
            "smtp.example.com",
            "opensre@example.com",
            "587",
            "starttls",
            "mailer",
            "secret",
            "team@example.com",
        ]
    )

    def fake_p(_label: str, default: str = "", secret: bool = False) -> str:
        return next(answers)

    saved: list[tuple[str, dict[str, object]]] = []
    monkeypatch.setattr("integrations.cli._p", fake_p)
    monkeypatch.setattr(
        "integrations.cli.upsert_integration",
        lambda service, entry: saved.append((service, entry)),
    )

    _setup_smtp()

    assert _HANDLERS["smtp"] is _setup_smtp
    assert saved == [
        (
            "smtp",
            {
                "credentials": {
                    "host": "smtp.example.com",
                    "port": 587,
                    "security": "starttls",
                    "username": "mailer",
                    "password": "secret",
                    "from_address": "opensre@example.com",
                    "default_to": "team@example.com",
                }
            },
        )
    ]


def test_integrations_setup_skips_auto_verify_for_unverifiable_service() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified"),
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify") as mock_verify,
    ):
        # rds is registered in SETUP_SERVICES but intentionally absent from
        # VERIFY_SERVICES, so it exercises the auto-verify-skip path.
        # (opensearch was used here previously but moved into VERIFY_SERVICES
        # by PR #1143, which is why this assertion was updated.)
        mock_setup.return_value = "rds"
        result = runner.invoke(cli, ["integrations", "setup", "rds"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("rds")
    mock_verify.assert_not_called()


def test_integrations_verify_accepts_github() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        result = runner.invoke(cli, ["integrations", "verify", "github"])

    assert result.exit_code == 0
    mock_verify.assert_called_once_with(
        "github",
        send_slack_test=False,
    )
    mock_capture.assert_called_once_with("github")


def test_integrations_verify_accepts_openclaw() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_verify", return_value=1) as mock_verify,
    ):
        result = runner.invoke(cli, ["integrations", "verify", "openclaw"])

    assert result.exit_code == 1
    mock_verify.assert_called_once_with(
        "openclaw",
        send_slack_test=False,
    )
    mock_capture.assert_not_called()


def test_integrations_verify_accepts_argocd() -> None:
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        result = runner.invoke(cli, ["integrations", "verify", "argocd"])

    assert result.exit_code == 0
    mock_verify.assert_called_once_with(
        "argocd",
        send_slack_test=False,
    )
    mock_capture.assert_called_once_with("argocd")


def test_integrations_setup_accepts_helm() -> None:
    # Regression test for #1973: helm had verify wired in the registry but no
    # setup_order / _setup_helm handler, so Click rejected the positional arg.
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_setup_started"),
        patch("surfaces.cli.commands.integrations.capture_integration_setup_completed"),
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_setup") as mock_setup,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        mock_setup.return_value = "helm"
        result = runner.invoke(cli, ["integrations", "setup", "helm"])

    assert result.exit_code == 0
    mock_setup.assert_called_once_with("helm")
    mock_verify.assert_called_once_with("helm")
    mock_capture.assert_called_once_with("helm")


def test_integrations_verify_accepts_helm() -> None:
    # Regression test for #1973: helm was registered in the runtime registry
    # but rejected by Click because the CLI's hardcoded VERIFY_SERVICES tuple
    # had drifted out of sync.
    runner = CliRunner()

    with (
        patch("surfaces.cli.commands.integrations.capture_integration_verified") as mock_capture,
        patch("integrations.cli.cmd_verify", return_value=0) as mock_verify,
    ):
        result = runner.invoke(cli, ["integrations", "verify", "helm"])

    assert result.exit_code == 0
    mock_verify.assert_called_once_with(
        "helm",
        send_slack_test=False,
    )
    mock_capture.assert_called_once_with("helm")


def test_verify_services_includes_previously_missing_integrations() -> None:
    # #1973 surfaced these names as registered in the runtime registry but
    # rejected by Click's positional-arg validator (the CLI's hardcoded
    # VERIFY_SERVICES tuple had drifted). Anchor them here so a revert to a
    # hardcoded tuple — or accidental removal from the registry — fails this
    # test loudly.
    previously_missing = {
        "azure",
        "azure_sql",
        "helm",
        "openobserve",
        "snowflake",
        "splunk",
        "supabase",
    }
    assert previously_missing <= set(VERIFY_SERVICES)


def test_setup_services_includes_previously_missing_integrations() -> None:
    # #2537: telegram, whatsapp and twilio had handlers and registry entries
    # but were rejected by Click because the CLI's hardcoded SETUP_SERVICES
    # tuple had drifted. Anchor them here so a revert to a hardcoded tuple —
    # or accidental removal from the registry — fails this test loudly.
    previously_missing = {"telegram", "twilio", "whatsapp"}
    assert previously_missing <= set(SETUP_SERVICES)