gyxx-flow/tests/test_feishu_messages.py

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from gyxx_flow.adapters import feishu_messages


def _completed(payload: dict, *, returncode: int = 0) -> SimpleNamespace:
    return SimpleNamespace(
        returncode=returncode,
        stdout=json.dumps(payload),
        stderr="",
    )


def test_bot_text_send_has_one_fixed_profile_identity_and_stable_key(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    commands: list[list[str]] = []
    monkeypatch.setattr(feishu_messages, "_lark_command", lambda: ["node", "run.js"])
    monkeypatch.setattr(
        feishu_messages,
        "resolve_notification_recipients",
        lambda recipients, *, app_profile: tuple(recipients),
    )

    def fake_run(command: list[str], **_kwargs: object) -> SimpleNamespace:
        commands.append(command)
        return _completed({"message_id": "om_receipt123", "chat_id": "oc_chat123"})

    monkeypatch.setattr(feishu_messages.subprocess, "run", fake_run)

    first = feishu_messages.send_lark_bot_message(
        user_id="ou_recipient123",
        text="告警正文",
        idempotency_key="product.alert:2026-08-10",
    )
    second = feishu_messages.send_lark_bot_message(
        user_id="ou_recipient123",
        text="告警正文",
        idempotency_key="product.alert:2026-08-10",
    )

    assert first["message_id"] == second["message_id"] == "om_receipt123"
    assert len(commands) == 2
    assert commands[0][:4] == ["node", "run.js", "--profile", "hermes-analyzer"]
    assert commands[0][commands[0].index("--as") + 1] == "bot"
    assert commands[0][commands[0].index("--user-id") + 1] == "ou_recipient123"
    assert commands[0][commands[0].index("--text") + 1] == "告警正文"
    key_index = commands[0].index("--idempotency-key") + 1
    assert commands[0][key_index] == commands[1][key_index]
    assert commands[0][key_index].startswith("gyxx-")
    assert len(commands[0][key_index]) == 45


def test_interactive_bot_send_preserves_card_json_and_nested_receipt(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    observed: dict[str, object] = {}
    monkeypatch.setattr(feishu_messages, "_lark_command", lambda: ["node", "run.js"])
    monkeypatch.setattr(
        feishu_messages,
        "resolve_notification_recipients",
        lambda recipients, *, app_profile: tuple(recipients),
    )

    def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
        observed["command"] = command
        observed["kwargs"] = kwargs
        return _completed({"data": {"message_id": "om_nested-123"}})

    monkeypatch.setattr(feishu_messages.subprocess, "run", fake_run)
    card = {"header": {"title": {"tag": "plain_text", "content": "日报"}}}

    result = feishu_messages.send_lark_bot_message(
        user_id="ou_recipient123",
        content=card,
        msg_type="interactive",
    )

    command = observed["command"]
    assert isinstance(command, list)
    assert json.loads(command[command.index("--content") + 1]) == card
    assert command[command.index("--msg-type") + 1] == "interactive"
    assert command[command.index("--as") + 1] == "bot"
    assert result["message_id"] == "om_nested-123"


def test_chat_and_image_messages_use_bot_identity(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    commands: list[list[str]] = []
    image = tmp_path / "qr.png"
    image.write_bytes(b"png")
    monkeypatch.setattr(feishu_messages, "_lark_command", lambda: ["node", "run.js"])

    def fake_run(command: list[str], **_kwargs: object) -> SimpleNamespace:
        commands.append(command)
        return _completed({"message_id": f"om_image{len(commands)}"})

    monkeypatch.setattr(feishu_messages.subprocess, "run", fake_run)

    feishu_messages.send_lark_bot_message(chat_id="oc_group123", text="群通知")
    feishu_messages.send_lark_bot_message(user_id="ou_owner123", image=image)

    assert commands[0][commands[0].index("--chat-id") + 1] == "oc_group123"
    assert commands[1][commands[1].index("--image") + 1] == "./qr.png"
    assert all(command[command.index("--as") + 1] == "bot" for command in commands)


@pytest.mark.parametrize(
    ("completed", "message"),
    (
        (_completed({}, returncode=1), "exit=1"),
        (_completed({"ok": False, "error": "denied"}), "denied"),
        (_completed({"id": "chatcmpl-not-a-receipt"}), "no valid message_id"),
    ),
)
def test_bot_send_fails_without_a_verified_lark_receipt(
    monkeypatch: pytest.MonkeyPatch,
    completed: SimpleNamespace,
    message: str,
) -> None:
    monkeypatch.setattr(feishu_messages, "_lark_command", lambda: ["node", "run.js"])
    monkeypatch.setattr(
        feishu_messages,
        "resolve_notification_recipients",
        lambda recipients, *, app_profile: tuple(recipients),
    )
    monkeypatch.setattr(feishu_messages.subprocess, "run", lambda *_args, **_kwargs: completed)

    with pytest.raises(RuntimeError, match=message):
        feishu_messages.send_lark_bot_message(
            user_id="ou_recipient123",
            text="告警正文",
        )


def test_sender_accepts_collector_bot_and_rejects_other_profiles(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(feishu_messages, "_lark_command", lambda: ["node", "run.js"])
    monkeypatch.setattr(
        feishu_messages.subprocess,
        "run",
        lambda *_args, **_kwargs: _completed({"message_id": "om_collector123"}),
    )
    result = feishu_messages.send_lark_bot_message(
        user_id="ou_recipient123",
        text="采集心跳",
        profile=feishu_messages.COLLECTOR_NOTIFICATION_APP_PROFILE,
    )
    assert result["message_id"] == "om_collector123"

    with pytest.raises(ValueError, match="Hermes application profile"):
        feishu_messages.send_lark_bot_message(
            user_id="ou_recipient123",
            text="告警正文",
            profile="collector",
        )