beaver_project/external-connector/tests/test_vendor_cli_provider.py

from __future__ import annotations

import json
from pathlib import Path

from external_connector.providers.vendor_cli import VendorCliProvider
from external_connector.state import SidecarStateStore


class FakeRunner:
    def __init__(self) -> None:
        self.commands: list[list[str]] = []
        self.cwd: str | None = None
        self.timeout: float | None = None

    def __call__(self, command: list[str], cwd: str, timeout: float) -> tuple[int, str, str]:
        self.commands.append(command)
        self.cwd = cwd
        self.timeout = timeout
        return 0, "connected account=weixin:me", ""


def test_vendor_cli_provider_uses_env_command_templates(tmp_path) -> None:
    runner = FakeRunner()
    provider = VendorCliProvider(
        store=SidecarStateStore(tmp_path / "state.json"),
        env={
            "WEIXIN_CONNECT_COMMAND": "vendor-weixin install --state {state_dir}",
            "CONNECTOR_COMMAND_TIMEOUT_SECONDS": "30",
        },
        runner=runner,
    )

    session = provider.start_session(
        {
            "kind": "weixin",
            "connectionId": "conn_1",
            "channelId": "weixin-main",
            "displayName": "Weixin Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {},
        }
    )

    assert session["status"] in {"waiting_for_user", "connected"}
    assert runner.commands[0][0] == "vendor-weixin"
    assert runner.cwd == str(tmp_path)
    assert runner.timeout == 30.0


def test_vendor_cli_provider_redacts_sensitive_error(tmp_path) -> None:
    def runner(command: list[str], cwd: str, timeout: float) -> tuple[int, str, str]:
        return 1, "", "failed secret-token appSecret=abc"

    provider = VendorCliProvider(
        store=SidecarStateStore(tmp_path / "state.json"),
        env={"FEISHU_CONNECT_COMMAND": "vendor-feishu install --secret abc"},
        runner=runner,
    )

    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {},
        }
    )

    assert session["status"] == "error"
    assert "secret-token" not in (session["error"] or "")
    assert "appSecret=abc" not in (session["error"] or "")


def test_vendor_cli_provider_refreshes_session_from_status_command_json(tmp_path) -> None:
    calls: list[list[str]] = []

    def runner(command: list[str], cwd: str, timeout: float) -> tuple[int, str, str]:
        calls.append(command)
        if command[0] == "vendor-weixin-status":
            return (
                0,
                '{"status":"qr_ready","qrImage":"data:image/png;base64,abc","qrCode":"weixin://scan","metadata":{"phase":"scan"}}',
                "",
            )
        return 0, "waiting", ""

    provider = VendorCliProvider(
        store=SidecarStateStore(tmp_path / "state.json"),
        env={
            "WEIXIN_CONNECT_COMMAND": "vendor-weixin install --state {state_dir}",
            "WEIXIN_STATUS_COMMAND": "vendor-weixin-status --state {state_dir} --session {session_id}",
        },
        runner=runner,
    )
    session = provider.start_session(
        {
            "kind": "weixin",
            "connectionId": "conn_1",
            "channelId": "weixin-main",
            "displayName": "Weixin Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {},
        }
    )

    refreshed = provider.get_session(session["sessionId"])

    assert calls[1][0] == "vendor-weixin-status"
    assert refreshed["status"] == "qr_ready"
    assert refreshed["qrImage"] == "data:image/png;base64,abc"
    assert refreshed["qrCode"] == "weixin://scan"
    assert refreshed["metadata"]["phase"] == "scan"


def test_vendor_cli_provider_refreshes_connected_session_from_key_value_status(tmp_path) -> None:
    def runner(command: list[str], cwd: str, timeout: float) -> tuple[int, str, str]:
        if command[0] == "vendor-feishu-status":
            return 0, "status=connected accountId=feishu:tenant-bot displayName=FeishuBot", ""
        return 0, "waiting", ""

    provider = VendorCliProvider(
        store=SidecarStateStore(tmp_path / "state.json"),
        env={
            "FEISHU_CONNECT_COMMAND": "vendor-feishu install --state {state_dir}",
            "FEISHU_STATUS_COMMAND": "vendor-feishu-status --state {state_dir}",
        },
        runner=runner,
    )
    session = provider.start_session(
        {
            "kind": "feishu",
            "connectionId": "conn_1",
            "channelId": "feishu-main",
            "displayName": "Feishu Main",
            "callbackBaseUrl": "http://beaver:8080",
            "options": {},
        }
    )

    refreshed = provider.get_session(session["sessionId"])

    assert refreshed["status"] == "connected"
    assert refreshed["accountId"] == "feishu:tenant-bot"
    assert refreshed["displayName"] == "FeishuBot"


def test_vendor_cli_provider_send_uses_payload_file_and_dedupes_result(tmp_path) -> None:
    payloads: list[dict[str, object]] = []
    commands: list[list[str]] = []

    def runner(command: list[str], cwd: str, timeout: float) -> tuple[int, str, str]:
        commands.append(command)
        payload_path = Path(command[command.index("--payload") + 1])
        payloads.append(json.loads(payload_path.read_text(encoding="utf-8")))
        return 0, '{"providerMessageId":"wx-msg-1"}', ""

    provider = VendorCliProvider(
        store=SidecarStateStore(tmp_path / "state.json"),
        env={"WEIXIN_SEND_COMMAND": "vendor-weixin-send --payload {payload_path}"},
        runner=runner,
    )
    payload = {
        "requestId": "out_1",
        "connectionId": "conn_1",
        "channelId": "weixin-main",
        "kind": "weixin",
        "target": {"peerId": "peer-1", "peerType": "dm", "threadId": None},
        "content": "hello world",
        "metadata": {"source": "test"},
    }

    first = provider.send(payload)
    duplicate = provider.send(payload)

    assert first == {"ok": True, "providerMessageId": "wx-msg-1"}
    assert duplicate == first
    assert len(commands) == 1
    assert payloads[0]["content"] == "hello world"
    assert payloads[0]["target"] == {"peerId": "peer-1", "peerType": "dm", "threadId": None}