company-ai-platform/tests/test_feishu_tenant_routing.py

from collections.abc import Iterator
from typing import Any

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.application.feishu.commands import FeishuCommandService
from app.core.config import get_settings
from app.core.database import Base
from app.modules.audit.models import AuditLog
from app.modules.feishu.service import FeishuService
from app.modules.feishu_users.constants import FeishuUserRole, FeishuUserStatus
from app.modules.feishu_users.principal import FeishuPrincipal


class RecordingFeishuClient:
    def __init__(self) -> None:
        self.calls: list[dict[str, Any]] = []

    def send_text(
        self,
        text: str,
        receive_id: str | None,
        receive_id_type: str,
        uuid: str | None,
        tenant_key: str | None = None,
    ) -> dict[str, Any]:
        self.calls.append(
            {
                "operation": "text",
                "receive_id": receive_id,
                "tenant_key": tenant_key,
                "uuid": uuid,
            }
        )
        return {"code": 0, "data": {"message_id": "om-text"}}

    def send_card(
        self,
        card: dict[str, Any],
        receive_id: str | None,
        receive_id_type: str,
        uuid: str | None,
        tenant_key: str | None = None,
    ) -> dict[str, Any]:
        self.calls.append(
            {
                "operation": "card",
                "receive_id": receive_id,
                "tenant_key": tenant_key,
                "uuid": uuid,
            }
        )
        return {"code": 0, "data": {"message_id": "om-card"}}

    def upload_image(
        self,
        image: bytes,
        filename: str = "lifecycle-report.png",
        tenant_key: str | None = None,
    ) -> dict[str, Any]:
        self.calls.append(
            {
                "operation": "image",
                "filename": filename,
                "tenant_key": tenant_key,
            }
        )
        return {"code": 0, "data": {"image_key": "img-key"}}


@pytest.fixture
def database(
    monkeypatch: pytest.MonkeyPatch,
) -> Iterator[Session]:
    monkeypatch.setenv("FEISHU_APP_ID", "cli-routing")
    monkeypatch.setenv("FEISHU_APP_SECRET", "routing-secret")
    monkeypatch.setenv("FEISHU_DEFAULT_TENANT_KEY", "tenant-default")
    monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "true")
    get_settings.cache_clear()
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine, tables=[AuditLog.__table__])
    try:
        with Session(engine) as db:
            yield db
    finally:
        engine.dispose()
        get_settings.cache_clear()


def test_service_routes_text_card_and_image_through_selected_tenant(
    database: Session,
) -> None:
    service = FeishuService(database)
    client = RecordingFeishuClient()
    service.client = client

    service.send_text("默认消息", receive_id="oc-default")
    service.set_tenant_key("tenant-a")
    service.send_card({"elements": []}, receive_id="oc-a")
    service.upload_image(b"png")
    service.send_text(
        "显式覆盖",
        receive_id="oc-b",
        tenant_key="tenant-b",
    )

    assert [
        (item["operation"], item["tenant_key"])
        for item in client.calls
    ] == [
        ("text", "tenant-default"),
        ("card", "tenant-a"),
        ("image", "tenant-a"),
        ("text", "tenant-b"),
    ]


def test_command_reply_uses_verified_principal_tenant(
    database: Session,
) -> None:
    commands = FeishuCommandService(database)
    client = RecordingFeishuClient()
    commands.feishu.client = client
    principal = FeishuPrincipal(
        owner_id=1,
        user_code="FSU-routing",
        tenant_key="tenant-principal",
        open_id="ou-routing",
        union_id=None,
        feishu_user_id=None,
        role=FeishuUserRole.USER,
        status=FeishuUserStatus.ACTIVE,
        timezone="Asia/Shanghai",
        quiet_hours_start=None,
        quiet_hours_end=None,
        chat_id="oc-routing",
        chat_type="p2p",
    )

    result = commands.handle_text(
        "帮助",
        auto_reply=True,
        principal=principal,
    )

    assert result["command"] == "help"
    assert commands.feishu.tenant_key == "tenant-principal"
    assert client.calls == [
        {
            "operation": "text",
            "receive_id": "oc-routing",
            "tenant_key": "tenant-principal",
            "uuid": None,
        }
    ]