253 lines
8.0 KiB
Python
253 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.api.feishu_api import router as feishu_router
|
|
from app.api.feishu_api import card_action_routes
|
|
from app.core.database import get_db
|
|
from app.services.integrations.feishu_integration_service import FeishuIntegrationService
|
|
|
|
|
|
class _FakeResult:
|
|
def scalar_one_or_none(self):
|
|
return None
|
|
|
|
|
|
class _FakeDB:
|
|
async def execute(self, query): # noqa: ARG002
|
|
return _FakeResult()
|
|
|
|
async def commit(self) -> None:
|
|
return None
|
|
|
|
async def rollback(self) -> None:
|
|
return None
|
|
|
|
|
|
def _build_app(db_session):
|
|
app = FastAPI()
|
|
app.include_router(feishu_router)
|
|
|
|
async def _fake_get_db():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_db] = _fake_get_db
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_card_action_callback_returns_success_and_dispatches_payload(monkeypatch) -> None:
|
|
called: dict[str, object] = {}
|
|
|
|
async def _fake_handle_card_action_trigger(self, event): # noqa: ANN001
|
|
called["event"] = event
|
|
return {
|
|
"code": 0,
|
|
"msg": "success",
|
|
"data": {
|
|
"handled": True,
|
|
"status": "succeeded",
|
|
"followup_reply": "已确认并入库",
|
|
"project_id": 1001,
|
|
"signal_id": 2002,
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
FeishuIntegrationService,
|
|
"_handle_card_action_trigger",
|
|
_fake_handle_card_action_trigger,
|
|
)
|
|
|
|
app = _build_app(_FakeDB())
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/feishu/card-action",
|
|
json={
|
|
"type": "card.action.trigger",
|
|
"open_message_id": "om_123",
|
|
"open_chat_id": "oc_456",
|
|
"open_id": "ou_abc",
|
|
"operator": {"open_id": "ou_abc", "name": "测试用户"},
|
|
"action": {
|
|
"value": {
|
|
"action": "confirm",
|
|
"work_item_id": 99,
|
|
}
|
|
},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["toast"]["content"] == "已确认并入库"
|
|
assert called["event"]["action"]["value"]["action"] == "confirm"
|
|
assert called["event"]["work_item_id"] == 99
|
|
assert called["event"]["open_message_id"] == "om_123"
|
|
assert called["event"]["open_chat_id"] == "oc_456"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_card_action_callback_maps_ignored_to_refresh_message(monkeypatch) -> None:
|
|
async def _fake_handle_card_action_trigger(self, event): # noqa: ANN001
|
|
_ = event
|
|
return {
|
|
"code": 0,
|
|
"msg": "card_action_ignored",
|
|
"data": {
|
|
"reason": "missing_work_item_id",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
FeishuIntegrationService,
|
|
"_handle_card_action_trigger",
|
|
_fake_handle_card_action_trigger,
|
|
)
|
|
|
|
app = _build_app(_FakeDB())
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/feishu/card-action",
|
|
json={
|
|
"type": "card.action.trigger",
|
|
"open_message_id": "om_ignored_1",
|
|
"open_chat_id": "oc_ignored_1",
|
|
"operator": {"open_id": "ou_abc", "name": "测试用户"},
|
|
"action": {
|
|
"value": {
|
|
"action": "confirm",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["toast"]["content"] == "这张卡片已失效，请刷新后再试"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_card_action_callback_drops_legacy_queue_item_id(monkeypatch) -> None:
|
|
called: dict[str, object] = {}
|
|
|
|
async def _fake_handle_card_action_trigger(self, event): # noqa: ANN001
|
|
called["event"] = event
|
|
return {
|
|
"code": 0,
|
|
"msg": "card_action_ignored",
|
|
"data": {
|
|
"reason": "missing_work_item_id",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
FeishuIntegrationService,
|
|
"_handle_card_action_trigger",
|
|
_fake_handle_card_action_trigger,
|
|
)
|
|
|
|
app = _build_app(_FakeDB())
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/feishu/card-action",
|
|
json={
|
|
"type": "card.action.trigger",
|
|
"open_message_id": "om_queue_legacy_1",
|
|
"open_chat_id": "oc_queue_legacy_1",
|
|
"operator": {"open_id": "ou_abc", "name": "测试用户"},
|
|
"action": {
|
|
"value": {
|
|
"action": "continue",
|
|
"queue_item_id": 201,
|
|
}
|
|
},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["toast"]["content"] == "这张卡片已失效，请刷新后再试"
|
|
assert "queue_item_id" not in called["event"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_card_action_callback_falls_back_when_signature_mismatch(monkeypatch) -> None:
|
|
called: dict[str, object] = {}
|
|
|
|
class _FallbackService:
|
|
tenant_id = 1
|
|
encrypt_key = ""
|
|
verification_token = "server-token"
|
|
|
|
async def load_config(self, tenant_id): # noqa: ANN001
|
|
self.tenant_id = tenant_id
|
|
|
|
async def _handle_card_action_trigger(self, event): # noqa: ANN001
|
|
called["event"] = event
|
|
return {
|
|
"code": 0,
|
|
"msg": "success",
|
|
"data": {
|
|
"handled": True,
|
|
"status": "succeeded",
|
|
"followup_reply": "已确认并入库",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(card_action_routes, "get_feishu_service", lambda db: _FallbackService())
|
|
|
|
body = (
|
|
'{"type":"card.action.trigger","open_message_id":"om_123","open_chat_id":"oc_456",'
|
|
'"operator":{"open_id":"ou_abc"},"action":{"value":{"action":"confirm","work_item_id":99}}}'
|
|
)
|
|
timestamp = str(int(time.time()))
|
|
nonce = "nonce-123"
|
|
wrong_token = "client-token"
|
|
signature = hashlib.sha1((timestamp + nonce + wrong_token).encode() + body.encode()).hexdigest()
|
|
|
|
app = _build_app(_FakeDB())
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/feishu/card-action",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"X-Lark-Request-Timestamp": timestamp,
|
|
"X-Lark-Request-Nonce": nonce,
|
|
"X-Lark-Signature": signature,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["toast"]["content"] == "已确认并入库"
|
|
assert called["event"]["action"]["value"]["action"] == "confirm"
|
|
assert called["event"]["work_item_id"] == 99
|
|
assert called["event"]["open_message_id"] == "om_123"
|
|
assert called["event"]["open_chat_id"] == "oc_456"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_card_action_callback_handles_url_verification() -> None:
|
|
app = _build_app(_FakeDB())
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/feishu/card-action",
|
|
json={
|
|
"type": "url_verification",
|
|
"token": "",
|
|
"challenge": "challenge-123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"challenge": "challenge-123"}
|