JiuContinent
eb8267ed18
feat(feishu): 添加飞书入站事件inbox和混合数据库协调功能 - 实现飞书入站事件持久化inbox机制，支持状态管理、租约锁定和重试退避 - 添加混合数据库基线协调工具，确保平台PostgreSQL结构安全对齐 - 增加运行组件心跳检测和readiness就绪检查机制 - 实现app_ticket事件的安全轮换和验证处理 - 添加生产环境运行编排和fail-closed安全机制 - 支持webhook快速确认和长连接独立进程处理 - 完善个人数据擦除时的待处理事件清理功能 ```
452 lines
16 KiB
Python
452 lines
16 KiB
Python
import json
|
||
from datetime import datetime
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from sqlalchemy import create_engine, func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.application.feishu.commands import FeishuCommandService
|
||
from app.application.feishu.handlers.subscriptions import handle_subscription_command
|
||
from app.core.config import get_settings
|
||
from app.core.database import Base
|
||
from app.modules.ai_agent.service import AIService
|
||
from app.modules.feishu.constants import FeishuCommandKey, FeishuCommandName
|
||
from app.modules.feishu.service import FeishuService
|
||
from app.modules.feishu_users.constants import FeishuUserRole
|
||
from app.modules.feishu_users.models import FeishuUser
|
||
from app.modules.feishu_users.principal import FeishuPrincipal
|
||
from app.modules.subscriptions.constants import (
|
||
DAILY_DELIVERY_LIMIT_REACHED,
|
||
MAX_ACTIVE_SUBSCRIPTIONS,
|
||
PushDeliveryStatus,
|
||
PushSubscriptionStatus,
|
||
SubscriptionScheduleType,
|
||
SubscriptionTargetType,
|
||
)
|
||
from app.modules.subscriptions.models import PushDelivery, PushSubscription
|
||
|
||
|
||
def _user(
|
||
db: Session,
|
||
*,
|
||
suffix: str,
|
||
role: str = FeishuUserRole.USER,
|
||
) -> FeishuUser:
|
||
record = FeishuUser(
|
||
code=f"FSU-{suffix}",
|
||
tenant_key=f"tenant-{suffix}",
|
||
open_id=f"open-{suffix}",
|
||
role=role,
|
||
timezone="Asia/Shanghai",
|
||
)
|
||
db.add(record)
|
||
db.commit()
|
||
db.refresh(record)
|
||
return record
|
||
|
||
|
||
def _handle(
|
||
db: Session,
|
||
text: str,
|
||
principal: FeishuPrincipal,
|
||
) -> dict:
|
||
result = handle_subscription_command(
|
||
db,
|
||
FeishuService(db),
|
||
text,
|
||
principal,
|
||
False,
|
||
)
|
||
assert result is not None
|
||
return result
|
||
|
||
|
||
def test_create_private_subscription_replies_with_normalized_plan() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="private-command")
|
||
principal = FeishuPrincipal.from_user(
|
||
user,
|
||
chat_id="private-chat",
|
||
chat_type="p2p",
|
||
)
|
||
|
||
result = _handle(
|
||
db,
|
||
"订阅 每天 09:00：请提醒我：喝水",
|
||
principal,
|
||
)
|
||
|
||
subscription = db.scalar(select(PushSubscription))
|
||
assert subscription is not None
|
||
assert result["command"] == FeishuCommandName.SUBSCRIPTION_CREATE
|
||
assert "订阅已启用" in result["content"]
|
||
assert "计划：每天 09:00" in result["content"]
|
||
assert "下次执行：" in result["content"]
|
||
assert f"暂停订阅 {subscription.code}" in result["content"]
|
||
assert subscription.owner_id == user.id
|
||
assert subscription.target_type == SubscriptionTargetType.USER
|
||
assert subscription.target_id == user.open_id
|
||
assert subscription.prompt == "请提醒我：喝水"
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
def test_rich_text_event_routes_subscription_instead_of_fallback_ai(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "true")
|
||
get_settings.cache_clear()
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="rich-text-command")
|
||
principal = FeishuPrincipal.from_user(
|
||
user,
|
||
chat_id="private-chat",
|
||
chat_type="p2p",
|
||
)
|
||
text = "订阅 每隔 15 分钟：给我一句简短的工作提醒"
|
||
payload = {
|
||
"header": {"tenant_key": user.tenant_key},
|
||
"event": {
|
||
"message": {
|
||
"chat_id": "private-chat",
|
||
"chat_type": "p2p",
|
||
"content": json.dumps(
|
||
{
|
||
"content": [
|
||
[
|
||
{
|
||
"tag": "text",
|
||
"text": text,
|
||
"style": [],
|
||
}
|
||
]
|
||
]
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
},
|
||
"sender": {"sender_id": {"open_id": user.open_id}},
|
||
},
|
||
}
|
||
|
||
commands = FeishuCommandService(db)
|
||
extracted = commands.extract_event_command(payload)
|
||
|
||
assert extracted is not None
|
||
assert extracted[FeishuCommandKey.TEXT] == text
|
||
result = commands.handle_text(
|
||
extracted[FeishuCommandKey.TEXT],
|
||
chat_id="private-chat",
|
||
principal=principal,
|
||
auto_reply=False,
|
||
)
|
||
subscription = db.scalar(select(PushSubscription))
|
||
assert result["command"] == FeishuCommandName.SUBSCRIPTION_CREATE
|
||
assert "计划：每隔 15 分钟" in result["content"]
|
||
assert subscription is not None
|
||
assert subscription.prompt == "给我一句简短的工作提醒"
|
||
finally:
|
||
get_settings.cache_clear()
|
||
engine.dispose()
|
||
|
||
|
||
def test_rich_text_event_routes_help_instead_of_fallback_ai(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setenv("FEISHU_USER_FEATURES_ENABLED", "true")
|
||
monkeypatch.setattr(
|
||
AIService,
|
||
"ask_personalized",
|
||
lambda *args, **kwargs: pytest.fail("Rich-text help command reached AI fallback"),
|
||
)
|
||
get_settings.cache_clear()
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="rich-text-help")
|
||
principal = FeishuPrincipal.from_user(
|
||
user,
|
||
chat_id="private-chat",
|
||
chat_type="p2p",
|
||
)
|
||
payload = {
|
||
"header": {"tenant_key": user.tenant_key},
|
||
"event": {
|
||
"message": {
|
||
"chat_id": "private-chat",
|
||
"chat_type": "p2p",
|
||
"content": json.dumps(
|
||
{
|
||
"content": [
|
||
[
|
||
{
|
||
"tag": "text",
|
||
"text": "帮助",
|
||
"style": [],
|
||
}
|
||
]
|
||
]
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
},
|
||
"sender": {"sender_id": {"open_id": user.open_id}},
|
||
},
|
||
}
|
||
|
||
commands = FeishuCommandService(db)
|
||
extracted = commands.extract_event_command(payload)
|
||
|
||
assert extracted is not None
|
||
assert extracted[FeishuCommandKey.TEXT] == "帮助"
|
||
result = commands.handle_text(
|
||
extracted[FeishuCommandKey.TEXT],
|
||
chat_id="private-chat",
|
||
principal=principal,
|
||
auto_reply=False,
|
||
)
|
||
assert result["command"] == FeishuCommandName.HELP
|
||
assert "你可以使用：" in result["content"]
|
||
finally:
|
||
get_settings.cache_clear()
|
||
engine.dispose()
|
||
|
||
|
||
def test_group_subscription_uses_current_verified_chat_and_requires_admin() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
ordinary = _user(db, suffix="ordinary-group")
|
||
denied = _handle(
|
||
db,
|
||
"订阅 每周一 09:00：群提醒",
|
||
FeishuPrincipal.from_user(
|
||
ordinary,
|
||
chat_id="verified-group",
|
||
chat_type="group",
|
||
),
|
||
)
|
||
assert "无权" in denied["content"]
|
||
assert db.scalar(select(func.count()).select_from(PushSubscription)) == 0
|
||
|
||
admin = _user(db, suffix="admin-group", role=FeishuUserRole.ADMIN)
|
||
accepted = _handle(
|
||
db,
|
||
"订阅 每周一 09:00：群提醒",
|
||
FeishuPrincipal.from_user(
|
||
admin,
|
||
chat_id="verified-group",
|
||
chat_type="group",
|
||
),
|
||
)
|
||
subscription = db.scalar(select(PushSubscription))
|
||
assert accepted["command"] == FeishuCommandName.SUBSCRIPTION_CREATE
|
||
assert subscription is not None
|
||
assert subscription.target_type == SubscriptionTargetType.CHAT
|
||
assert subscription.target_id == "verified-group"
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
def test_list_pause_resume_and_cancel_subscription_commands() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="lifecycle-command")
|
||
principal = FeishuPrincipal.from_user(user, chat_type="p2p")
|
||
_handle(db, "订阅 每天 10:00：生命周期测试", principal)
|
||
subscription = db.scalar(select(PushSubscription))
|
||
assert subscription is not None
|
||
|
||
listed = _handle(db, "我的订阅", principal)
|
||
assert listed["command"] == FeishuCommandName.SUBSCRIPTION_LIST
|
||
assert subscription.code in listed["content"]
|
||
assert "每天 10:00" in listed["content"]
|
||
|
||
paused = _handle(db, f"暂停订阅 {subscription.code}", principal)
|
||
db.refresh(subscription)
|
||
assert paused["command"] == FeishuCommandName.SUBSCRIPTION_PAUSE
|
||
assert subscription.status == PushSubscriptionStatus.PAUSED
|
||
|
||
resumed = _handle(db, f"恢复订阅 {subscription.code}", principal)
|
||
db.refresh(subscription)
|
||
assert resumed["command"] == FeishuCommandName.SUBSCRIPTION_RESUME
|
||
assert subscription.status == PushSubscriptionStatus.ACTIVE
|
||
assert "下次执行：" in resumed["content"]
|
||
|
||
cancelled = _handle(db, f"退订 {subscription.code}", principal)
|
||
db.refresh(subscription)
|
||
assert cancelled["command"] == FeishuCommandName.SUBSCRIPTION_CANCEL
|
||
assert subscription.status == PushSubscriptionStatus.CANCELLED
|
||
assert subscription.next_run_at is None
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
def test_timezone_and_quiet_hour_commands_update_current_user() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="settings-command")
|
||
principal = FeishuPrincipal.from_user(user, chat_type="p2p")
|
||
|
||
timezone_result = _handle(db, "设置时区 Asia/Tokyo", principal)
|
||
db.refresh(user)
|
||
assert timezone_result["command"] == FeishuCommandName.SUBSCRIPTION_TIMEZONE
|
||
assert user.timezone == "Asia/Tokyo"
|
||
|
||
quiet_result = _handle(db, "设置安静时段 22:00-07:00", principal)
|
||
db.refresh(user)
|
||
assert quiet_result["command"] == FeishuCommandName.SUBSCRIPTION_QUIET_HOURS
|
||
assert user.quiet_hours_start.strftime("%H:%M") == "22:00"
|
||
assert user.quiet_hours_end.strftime("%H:%M") == "07:00"
|
||
|
||
closed = _handle(db, "关闭安静时段", principal)
|
||
db.refresh(user)
|
||
assert closed["command"] == FeishuCommandName.SUBSCRIPTION_QUIET_HOURS
|
||
assert user.quiet_hours_start is None
|
||
assert user.quiet_hours_end is None
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
def test_invalid_schedule_and_capacity_error_create_no_partial_record() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="invalid-command")
|
||
principal = FeishuPrincipal.from_user(user, chat_type="p2p")
|
||
|
||
invalid = _handle(
|
||
db,
|
||
"订阅 每隔10分钟：过于频繁",
|
||
principal,
|
||
)
|
||
assert "示例" not in invalid["content"]
|
||
assert "订阅 每天 09:00" in invalid["content"]
|
||
assert db.scalar(select(func.count()).select_from(PushSubscription)) == 0
|
||
|
||
for index in range(MAX_ACTIVE_SUBSCRIPTIONS):
|
||
db.add(
|
||
PushSubscription(
|
||
code=f"SUB-CAPACITY-{index}",
|
||
owner_id=user.id,
|
||
target_type=SubscriptionTargetType.USER,
|
||
target_id=user.open_id,
|
||
prompt=f"已有订阅 {index}",
|
||
schedule_type=SubscriptionScheduleType.DAILY,
|
||
schedule_config={"hour": 9, "minute": 0},
|
||
timezone=user.timezone,
|
||
next_run_at=datetime(2026, 7, 27, 1, 0),
|
||
status=PushSubscriptionStatus.ACTIVE,
|
||
consented_at=datetime(2026, 7, 26, 0, 0),
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
limited = _handle(
|
||
db,
|
||
"订阅 每天 11:00：第 51 条",
|
||
principal,
|
||
)
|
||
assert "最多 50 个" in limited["content"]
|
||
assert (
|
||
db.scalar(select(func.count()).select_from(PushSubscription))
|
||
== MAX_ACTIVE_SUBSCRIPTIONS
|
||
)
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
def test_unrelated_text_is_not_claimed_by_subscription_handler() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="unrelated-command")
|
||
|
||
assert (
|
||
handle_subscription_command(
|
||
db,
|
||
FeishuService(db),
|
||
"今天怎么样",
|
||
FeishuPrincipal.from_user(user),
|
||
False,
|
||
)
|
||
is None
|
||
)
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
def test_list_explains_daily_delivery_limit_skip() -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix="delivery-note")
|
||
principal = FeishuPrincipal.from_user(user, chat_type="p2p")
|
||
_handle(db, "订阅 每天 10:00：投递说明", principal)
|
||
subscription = db.scalar(select(PushSubscription))
|
||
assert subscription is not None
|
||
now = datetime(2026, 7, 26, 1, 0)
|
||
db.add(
|
||
PushDelivery(
|
||
code="DEL-DAILY-LIMIT",
|
||
subscription_id=subscription.id,
|
||
scheduled_for=now,
|
||
idempotency_key=uuid4().hex + uuid4().hex,
|
||
message_uuid=str(uuid4()),
|
||
status=PushDeliveryStatus.SKIPPED,
|
||
next_attempt_at=None,
|
||
last_error=DAILY_DELIVERY_LIMIT_REACHED,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
listed = _handle(db, "我的订阅", principal)
|
||
|
||
assert "每日最多 96 条限制" in listed["content"]
|
||
finally:
|
||
engine.dispose()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"command",
|
||
[
|
||
"设置时区 Invalid/Timezone",
|
||
"设置安静时段 25:00-07:00",
|
||
"订阅 每隔999999999999999999小时：不会创建",
|
||
],
|
||
)
|
||
def test_invalid_settings_and_oversized_interval_return_command_error(
|
||
command: str,
|
||
) -> None:
|
||
engine = create_engine("sqlite://")
|
||
Base.metadata.create_all(engine)
|
||
try:
|
||
with Session(engine) as db:
|
||
user = _user(db, suffix=f"invalid-{abs(hash(command))}")
|
||
principal = FeishuPrincipal.from_user(user, chat_type="p2p")
|
||
|
||
result = _handle(db, command, principal)
|
||
|
||
assert "未执行" in result["content"] or "无法识别" in result["content"]
|
||
assert db.scalar(select(func.count()).select_from(PushSubscription)) == 0
|
||
finally:
|
||
engine.dispose()
|