JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine, select, update
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.application.events import EventDispatchService
|
|
from app.core.utils.time import utc_now
|
|
from app.modules.audit.constants import AuditAction
|
|
from app.modules.audit.models import AuditLog
|
|
from app.modules.events.constants import EventStatus
|
|
from app.modules.events.models import DomainEvent
|
|
from app.modules.events.services import EventService
|
|
|
|
|
|
def test_stale_worker_is_fenced_after_expired_lease_is_reclaimed(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'event-fencing.db'}")
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql("PRAGMA journal_mode=WAL")
|
|
DomainEvent.__table__.create(engine)
|
|
AuditLog.__table__.create(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
|
|
try:
|
|
with factory() as db:
|
|
event = EventService(db).emit(
|
|
event_type="test.fencing",
|
|
source="pytest",
|
|
aggregate_type="test",
|
|
aggregate_id="fencing",
|
|
idempotency_key="test-event-fencing",
|
|
)
|
|
event_id = event.event_id
|
|
|
|
second_result: dict[str, DomainEvent] = {}
|
|
with factory() as first_db:
|
|
first_worker = EventDispatchService(first_db)
|
|
|
|
def fail_after_lease_is_reclaimed(record: DomainEvent) -> None:
|
|
first_db.add(
|
|
AuditLog(
|
|
actor="worker-one",
|
|
action="stale-worker-side-effect",
|
|
target_id=record.event_id,
|
|
)
|
|
)
|
|
with factory() as second_db:
|
|
second_db.execute(
|
|
update(DomainEvent)
|
|
.where(DomainEvent.event_id == record.event_id)
|
|
.values(locked_until=utc_now() - timedelta(seconds=1))
|
|
)
|
|
second_db.commit()
|
|
|
|
second_worker = EventDispatchService(second_db)
|
|
monkeypatch.setattr(
|
|
second_worker,
|
|
"_handle_event",
|
|
lambda claimed: None,
|
|
)
|
|
second_result["event"] = second_worker.dispatch_event(
|
|
record.event_id,
|
|
worker_id="worker-two",
|
|
)
|
|
raise RuntimeError("worker one failed after losing its lease")
|
|
|
|
monkeypatch.setattr(
|
|
first_worker,
|
|
"_handle_event",
|
|
fail_after_lease_is_reclaimed,
|
|
)
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
first_worker.dispatch_event(event_id, worker_id="worker-one")
|
|
|
|
assert exc_info.value.status_code == 409
|
|
|
|
assert second_result["event"].status == EventStatus.PROCESSED
|
|
with factory() as db:
|
|
stored = db.execute(
|
|
select(DomainEvent).where(DomainEvent.event_id == event_id)
|
|
).scalar_one()
|
|
audit_actions = list(
|
|
db.execute(
|
|
select(AuditLog.action).order_by(AuditLog.id.asc())
|
|
).scalars()
|
|
)
|
|
|
|
assert stored.status == EventStatus.PROCESSED
|
|
assert stored.attempts == 2
|
|
assert stored.last_error is None
|
|
assert stored.locked_by is None
|
|
assert stored.locked_until is None
|
|
assert audit_actions == [AuditAction.EVENT_DISPATCH]
|
|
assert "stale-worker-side-effect" not in audit_actions
|
|
finally:
|
|
engine.dispose()
|