company-ai-platform/tests/test_feishu_event_identity.py

import json
from collections.abc import Iterator
from dataclasses import replace
from typing import Any

import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.application.feishu.commands import FeishuCommandService
from app.application.feishu.events import FeishuEventService
from app.core.config import get_settings
from app.core.database import Base, get_db
from app.modules.audit.models import AuditLog
from app.modules.feishu.constants import (
    FeishuCommandName,
    FeishuEventSource,
)
from app.modules.feishu.models import FeishuEventReceipt
from app.modules.feishu.routes import router as feishu_router
from app.modules.feishu_users.constants import (
    FeishuUserAuditAction,
    FeishuUserRole,
    FeishuUserStatus,
)
from app.modules.feishu_users.models import (
    FeishuAdminBootstrapTombstone,
    FeishuUser,
)
from app.modules.feishu_users.principal import FeishuMention
from app.modules.feishu_users.services import (
    FeishuIdentityService,
    FeishuUserManagementService,
)


@pytest.fixture
def session_factory(
    monkeypatch: pytest.MonkeyPatch,
) -> Iterator[sessionmaker[Session]]:
    monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "true")
    monkeypatch.setenv("FEISHU_ADMIN_IDENTITIES", "tenant-a:ou-admin")
    monkeypatch.setenv("FEISHU_APP_ID", "")
    monkeypatch.setenv("FEISHU_APP_SECRET", "")
    monkeypatch.setenv("FEISHU_VERIFICATION_TOKEN", "verified-token")
    get_settings.cache_clear()
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(
        engine,
        tables=[
            AuditLog.__table__,
            FeishuEventReceipt.__table__,
            FeishuAdminBootstrapTombstone.__table__,
            FeishuUser.__table__,
        ],
    )
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    try:
        yield factory
    finally:
        Base.metadata.drop_all(
            engine,
            tables=[
                FeishuAdminBootstrapTombstone.__table__,
                FeishuUser.__table__,
                FeishuEventReceipt.__table__,
                AuditLog.__table__,
            ],
        )
        engine.dispose()
        get_settings.cache_clear()


def _message_event(
    *,
    event_id: str,
    text: str,
    tenant_key: str | None = "tenant-a",
    open_id: str | None = "ou-user",
    chat_id: str = "oc-chat",
    chat_type: str = "p2p",
    mentions: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    header: dict[str, Any] = {
        "event_id": event_id,
        "event_type": "im.message.receive_v1",
        "token": "verified-token",
    }
    if tenant_key is not None:
        header["tenant_key"] = tenant_key
    sender_id: dict[str, str] = {"user_id": "legacy-user-id"}
    if open_id is not None:
        sender_id["open_id"] = open_id
    message: dict[str, Any] = {
        "chat_id": chat_id,
        "chat_type": chat_type,
        "message_id": f"om-{event_id}",
        "message_type": "text",
        "content": json.dumps({"text": text}, ensure_ascii=False),
    }
    if mentions is not None:
        message["mentions"] = mentions
    return {
        "schema": "2.0",
        "header": header,
        "event": {
            "sender": {"sender_id": sender_id},
            "message": message,
        },
    }


def _mention(
    key: str,
    open_id: str,
    *,
    name: str,
    tenant_key: str = "tenant-a",
) -> dict[str, Any]:
    return {
        "key": key,
        "name": name,
        "tenant_key": tenant_key,
        "id": {
            "open_id": open_id,
            "union_id": f"on-{open_id}",
            "user_id": f"u-{open_id}",
        },
    }


def test_verified_event_requires_tenant_and_open_id_without_registering(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        missing_tenant = FeishuEventService(db)._handle_verified_event(
            _message_event(
                event_id="missing-tenant",
                text="risk",
                tenant_key=None,
            ),
            source=FeishuEventSource.WEBHOOK,
            auto_reply=False,
        )
        missing_open_id = FeishuEventService(db)._handle_verified_event(
            _message_event(
                event_id="missing-open",
                text="risk",
                open_id=None,
            ),
            source=FeishuEventSource.WEBHOOK,
            auto_reply=False,
        )

        assert missing_tenant["result"]["command"] == "permission_denied"
        assert missing_open_id["result"]["command"] == "permission_denied"
        assert db.scalar(select(FeishuUser)) is None
        denial_logs = list(
            db.execute(
                select(AuditLog).where(
                    AuditLog.action == FeishuUserAuditAction.PERMISSION_DENIED
                )
            ).scalars()
        )
        assert len(denial_logs) == 2
        assert all(item.actor == "feishu" for item in denial_logs)


def test_unverified_event_cannot_create_identity(
    session_factory: sessionmaker[Session],
) -> None:
    payload = _message_event(
        event_id="invalid-token",
        text="问 你好",
        open_id="ou-unverified",
    )
    payload["header"]["token"] = "invalid-token"
    with session_factory() as db:
        with pytest.raises(HTTPException) as exc_info:
            FeishuEventService(db).handle_event(
                payload,
                source=FeishuEventSource.WEBHOOK,
                auto_reply=False,
            )

        assert exc_info.value.status_code == 401
        assert db.scalar(select(FeishuUser)) is None


def test_group_event_uses_sender_principal_and_structured_context(
    session_factory: sessionmaker[Session],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured: dict[str, Any] = {}

    def fake_handle_text(
        self: FeishuCommandService,
        text: str,
        chat_id: str | None = None,
        actor: str = "feishu",
        auto_reply: bool = True,
        principal: Any = None,
    ) -> dict[str, Any]:
        captured.update(
            {
                "text": text,
                "chat_id": chat_id,
                "actor": actor,
                "principal": principal,
            }
        )
        return {
            "command": "ai_ask",
            "reply_type": "text",
            "title": "test",
            "content": "ok",
            "provider_response": None,
        }

    monkeypatch.setattr(FeishuCommandService, "handle_text", fake_handle_text)
    payload = _message_event(
        event_id="group-principal",
        text="@_bot 问 项目状态",
        open_id="ou-sender",
        chat_id="oc-group",
        chat_type="group",
        mentions=[_mention("@_target", "ou-target", name="目标用户")],
    )
    with session_factory() as db:
        result = FeishuEventService(db)._handle_verified_event(
            payload,
            source=FeishuEventSource.LONG_CONNECTION,
            auto_reply=False,
        )
        principal = captured["principal"]

        assert result["handled"] is True
        assert principal.open_id == "ou-sender"
        assert principal.chat_id == "oc-group"
        assert principal.chat_type == "group"
        assert principal.mentions[0].open_id == "ou-target"
        assert captured["actor"] == principal.user_code
        assert captured["chat_id"] == "oc-group"


@pytest.mark.parametrize(
    "command",
    [
        "日报",
        "项目资金 P-001",
        "风险",
        "学习公司规则：所有人共享",
        "设为管理员 @_target",
    ],
)
def test_ordinary_user_cannot_run_company_commands(
    session_factory: sessionmaker[Session],
    command: str,
) -> None:
    with session_factory() as db:
        principal = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id=f"ou-{abs(hash(command))}",
        )
        result = FeishuCommandService(db).handle_text(
            command,
            principal=principal,
            auto_reply=False,
        )

        assert result["command"] == FeishuCommandName.PERMISSION_DENIED
        assert "公司级功能" in result["content"]


def test_disabled_user_is_rejected_before_command_execution(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        principal = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-disabled",
        )
        FeishuUserManagementService(db).update_user(
            principal.user_code,
            changes={"status": FeishuUserStatus.DISABLED},
            actor="service-admin",
        )
        disabled = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-disabled",
        )
        result = FeishuCommandService(db).handle_text(
            "问 你好",
            principal=disabled,
            auto_reply=False,
        )

        assert result["command"] == FeishuCommandName.PERMISSION_DENIED
        assert "已停用" in result["content"]


def test_admin_command_uses_target_mention_and_ignores_bot_mention(
    session_factory: sessionmaker[Session],
) -> None:
    payload = _message_event(
        event_id="admin-promote",
        text="@_bot 设为管理员 @_target",
        open_id="ou-admin",
        chat_id="oc-group",
        chat_type="group",
        mentions=[
            _mention("@_bot", "ou-bot", name="机器人"),
            _mention("@_target", "ou-target", name="张三"),
        ],
    )
    with session_factory() as db:
        result = FeishuEventService(db)._handle_verified_event(
            payload,
            source=FeishuEventSource.WEBHOOK,
            auto_reply=False,
        )
        target = FeishuIdentityService(db).get_by_identity(
            tenant_key="tenant-a",
            open_id="ou-target",
        )

        assert result["result"]["command"] == FeishuCommandName.USER_SET_ADMIN
        assert "张三 已设为管理员" in result["result"]["content"]
        assert target is not None
        assert target.role == FeishuUserRole.ADMIN
        assert target.union_id == "on-ou-target"


def test_admin_command_rejects_text_identity_cross_tenant_and_last_admin(
    session_factory: sessionmaker[Session],
) -> None:
    with session_factory() as db:
        admin = FeishuIdentityService(db).resolve_or_register(
            tenant_key="tenant-a",
            open_id="ou-admin",
        )
        admin = replace(admin, chat_id="oc-group", chat_type="group")
        service = FeishuCommandService(db)

        forged = service.handle_text(
            "设为管理员 @ou-forged",
            principal=admin,
            auto_reply=False,
        )
        assert forged["command"] == FeishuCommandName.USER_SET_ADMIN
        assert "真实的飞书 @用户" in forged["content"]
        assert (
            FeishuIdentityService(db).get_by_identity(
                tenant_key="tenant-a",
                open_id="ou-forged",
            )
            is None
        )

        cross_tenant = service.handle_text(
            "设为管理员 @_target",
            principal=replace(
                admin,
                mentions=(
                    FeishuMention(
                        key="@_target",
                        name="其他租户用户",
                        tenant_key="tenant-b",
                        open_id="ou-other",
                    ),
                ),
            ),
            auto_reply=False,
        )
        assert "当前租户" in cross_tenant["content"]
        assert (
            FeishuIdentityService(db).get_by_identity(
                tenant_key="tenant-b",
                open_id="ou-other",
            )
            is None
        )

        self_demote = service.handle_text(
            "设为普通用户 @_self",
            principal=replace(
                admin,
                mentions=(
                    FeishuMention(
                        key="@_self",
                        name="管理员",
                        tenant_key="tenant-a",
                        open_id="ou-admin",
                    ),
                ),
            ),
            auto_reply=False,
        )
        assert "最后一个有效管理员" in self_demote["content"]
        stored_admin = FeishuIdentityService(db).get_by_identity(
            tenant_key="tenant-a",
            open_id="ou-admin",
        )
        assert stored_admin is not None
        assert stored_admin.role == FeishuUserRole.ADMIN


def test_preview_cannot_forge_feishu_principal_when_features_enabled(
    session_factory: sessionmaker[Session],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("API_KEY", "service-key")
    monkeypatch.setenv("API_ACTOR", "service-principal")
    get_settings.cache_clear()
    app = FastAPI()
    app.include_router(feishu_router, prefix="/api/v1/integrations/feishu")

    def override_db() -> Iterator[Session]:
        with session_factory() as db:
            yield db

    app.dependency_overrides[get_db] = override_db
    response = TestClient(app).post(
        "/api/v1/integrations/feishu/commands/preview",
        headers={"X-API-Key": "service-key"},
        json={
            "text": "日报",
            "actor": "ou-admin",
            "auto_reply": False,
        },
    )

    assert response.status_code == 200
    assert response.json()["command"] == "permission_denied"
    with session_factory() as db:
        assert db.scalar(select(FeishuUser)) is None


def test_disabled_feature_flag_preserves_legacy_actor_flow(
    session_factory: sessionmaker[Session],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "false")
    get_settings.cache_clear()
    captured: dict[str, Any] = {}

    def fake_handle_text(
        self: FeishuCommandService,
        text: str,
        chat_id: str | None = None,
        actor: str = "feishu",
        auto_reply: bool = True,
        principal: Any = None,
    ) -> dict[str, Any]:
        captured["actor"] = actor
        captured["principal"] = principal
        return {
            "command": "fallback_ai",
            "reply_type": "text",
            "title": "test",
            "content": "ok",
            "provider_response": None,
        }

    monkeypatch.setattr(FeishuCommandService, "handle_text", fake_handle_text)
    with session_factory() as db:
        FeishuEventService(db)._handle_verified_event(
            _message_event(
                event_id="legacy-flow",
                text="hello",
                tenant_key=None,
                open_id=None,
            ),
            source=FeishuEventSource.WEBHOOK,
            auto_reply=False,
        )

        assert captured["actor"] == "legacy-user-id"
        assert captured["principal"] is None
        assert db.scalar(select(FeishuUser)) is None