860 lines
29 KiB
Python
860 lines
29 KiB
Python
"""End-to-end service tests for BOXIM takeover timing and human priority."""
|
|
|
|
import asyncio
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from threading import Barrier
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from database import Base
|
|
from models import Avatar, ChatAttachment, TakeoverCursor, TakeoverMessage, TakeoverReplyTask, User
|
|
from services.boxim_client import BoxIMError
|
|
from services.boxim_image_service import DownloadedBoxIMImage
|
|
from services.takeover_service import (
|
|
AVATAR_LOCAL_ID_PREFIX,
|
|
TakeoverService,
|
|
_avatar_local_id,
|
|
_plain_text_reply,
|
|
)
|
|
|
|
|
|
class Clock:
|
|
def __init__(self):
|
|
self.value = datetime(2026, 8, 19, 10, 0, 0)
|
|
|
|
def now(self):
|
|
return self.value
|
|
|
|
def advance(self, seconds: int):
|
|
self.value += timedelta(seconds=seconds)
|
|
|
|
def millis(self):
|
|
return int(self.value.replace(tzinfo=timezone.utc).timestamp() * 1000)
|
|
|
|
|
|
class FakeBoxIM:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.sent = []
|
|
self.read_receipts = []
|
|
|
|
async def exchange_access_token(self, huihui_token):
|
|
assert huihui_token == "prod-huihui-token"
|
|
return {"accessToken": "box-token", "accessTokenExpiresIn": 3600}
|
|
|
|
async def get_self(self, access_token):
|
|
assert access_token == "box-token"
|
|
return {"id": 100}
|
|
|
|
async def fetch_private_messages(self, access_token, min_id="0"):
|
|
assert access_token == "box-token"
|
|
return [item.copy() for item in self.messages if int(item["id"]) > int(min_id)]
|
|
|
|
async def mark_private_messages_read(self, access_token, friend_id, message_id):
|
|
assert access_token == "box-token"
|
|
self.read_receipts.append(
|
|
{"friendId": str(friend_id), "messageId": str(message_id)}
|
|
)
|
|
|
|
async def send_private_message(self, access_token, peer_id, content, *, local_id=None):
|
|
self.sent.append({"peerId": str(peer_id), "content": content, "localId": str(local_id)})
|
|
return {"id": 900 + len(self.sent), "localId": int(local_id)}
|
|
|
|
|
|
class ConcurrentPollingBoxIM(FakeBoxIM):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.active_polls = 0
|
|
self.peak_active_polls = 0
|
|
|
|
async def exchange_access_token(self, huihui_token):
|
|
return {"accessToken": huihui_token, "accessTokenExpiresIn": 3600}
|
|
|
|
async def get_self(self, access_token):
|
|
return {"id": 100 if access_token == "prod-huihui-token" else 101}
|
|
|
|
async def fetch_private_messages(self, access_token, min_id="0"):
|
|
self.active_polls += 1
|
|
self.peak_active_polls = max(self.peak_active_polls, self.active_polls)
|
|
await asyncio.sleep(0.05)
|
|
self.active_polls -= 1
|
|
return []
|
|
|
|
|
|
class ConcurrentMessagePollingBoxIM(ConcurrentPollingBoxIM):
|
|
async def fetch_private_messages(self, access_token, min_id="0"):
|
|
await super().fetch_private_messages(access_token, min_id)
|
|
owner_id = 100 if access_token == "prod-huihui-token" else 101
|
|
return [
|
|
{
|
|
"id": owner_id,
|
|
"localId": owner_id,
|
|
"sendId": owner_id + 100,
|
|
"recvId": owner_id,
|
|
"sendTime": 1_700_000_000_000,
|
|
"type": 0,
|
|
"content": "并发写入测试",
|
|
}
|
|
]
|
|
|
|
async def mark_private_messages_read(self, access_token, friend_id, message_id):
|
|
await asyncio.sleep(0.05)
|
|
self.read_receipts.append(
|
|
{"friendId": str(friend_id), "messageId": str(message_id)}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def service_context(tmp_path):
|
|
engine = create_engine(
|
|
f"sqlite:///{tmp_path / 'takeover.db'}",
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
session_factory = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
|
|
Base.metadata.create_all(engine)
|
|
db = session_factory()
|
|
user = User(
|
|
id="owner-local",
|
|
huihui_user_id="owner-huihui",
|
|
huihui_token="prod-huihui-token",
|
|
app_token="app-token",
|
|
)
|
|
avatar = Avatar(
|
|
id="avatar-1",
|
|
owner_id=user.huihui_user_id,
|
|
name="分身",
|
|
status="active",
|
|
config={
|
|
"authorizationPermissions": ["chat", "takeover"],
|
|
"takeoverReplyDelaySeconds": 3,
|
|
},
|
|
)
|
|
db.add_all([user, avatar])
|
|
db.commit()
|
|
db.close()
|
|
|
|
clock = Clock()
|
|
boxim = FakeBoxIM()
|
|
service = TakeoverService(session_factory, boxim, now=clock.now)
|
|
return session_factory, service, boxim, clock
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_sync_primes_cursor_without_replying_to_history(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
boxim.messages = [
|
|
{"id": 10, "localId": 1, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "旧消息"}
|
|
]
|
|
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "不应发送"}):
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
cursor = db.query(TakeoverCursor).one()
|
|
assert cursor.initialized is True
|
|
assert cursor.last_message_id == "10"
|
|
assert db.query(TakeoverMessage).count() == 1
|
|
assert db.query(TakeoverReplyTask).count() == 0
|
|
assert boxim.sent == []
|
|
assert boxim.read_receipts == [{"friendId": "200", "messageId": "10"}]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incoming_message_is_prepared_then_sent_at_three_seconds(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 11, "localId": 2, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "你好"}
|
|
)
|
|
|
|
await service.poll_and_process_messages()
|
|
assert boxim.sent == []
|
|
assert boxim.read_receipts == [{"friendId": "200", "messageId": "11"}]
|
|
|
|
clock.advance(2)
|
|
await service.poll_and_process_messages()
|
|
assert boxim.sent == []
|
|
|
|
clock.advance(1)
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "**你好**\n\n很高兴见到你"}):
|
|
await service.poll_and_process_messages()
|
|
assert boxim.sent == [{"peerId": "200", "content": "你好\n很高兴见到你", "localId": boxim.sent[0]["localId"]}]
|
|
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).one()
|
|
assert task.status == "sent"
|
|
assert task.sent_at == clock.now()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incoming_image_is_analyzed_and_used_in_takeover_reply(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{
|
|
"id": 111,
|
|
"localId": 111,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 1,
|
|
"content": json.dumps(
|
|
{
|
|
"originUrl": "https://cdn.example/case.png",
|
|
"thumbUrl": "https://cdn.example/case-thumb.png",
|
|
}
|
|
),
|
|
}
|
|
)
|
|
|
|
await service.poll_and_process_messages()
|
|
db = session_factory()
|
|
try:
|
|
scheduled = db.query(TakeoverReplyTask).filter_by(trigger_message_id="111").one()
|
|
assert scheduled.status == "pending"
|
|
assert scheduled.prompt == "请看看这张图片。"
|
|
finally:
|
|
db.close()
|
|
|
|
clock.advance(3)
|
|
|
|
def analyze(db, avatar, content, **kwargs):
|
|
assert content == b"image-content"
|
|
attachment = ChatAttachment(
|
|
avatar_id=avatar.id,
|
|
uploader_kind=kwargs["uploader_kind"],
|
|
filename=kwargs["filename"],
|
|
mime_type="image/jpeg",
|
|
file_size=len(content),
|
|
status="ready",
|
|
category="medical_document",
|
|
summary="一张门诊病例",
|
|
extracted_text="主诉：咳嗽三天",
|
|
structured_data={"medical": {"chief_complaint": "咳嗽三天"}},
|
|
warning="请核对原始资料",
|
|
expires_at=clock.now() + timedelta(hours=24),
|
|
)
|
|
db.add(attachment)
|
|
db.commit()
|
|
db.refresh(attachment)
|
|
return attachment
|
|
|
|
downloaded = DownloadedBoxIMImage(
|
|
content=b"image-content",
|
|
filename="case.png",
|
|
mime_type="image/png",
|
|
source_url="https://cdn.example/case.png",
|
|
)
|
|
with (
|
|
patch("services.takeover_service.download_boxim_image", return_value=downloaded),
|
|
patch("routers.chat._analyze_image_bytes", side_effect=analyze) as analyzer,
|
|
patch("routers.chat._resolve_reply", return_value={"answer": "这份资料里写的是咳嗽三天。"}) as resolver,
|
|
):
|
|
await service.poll_and_process_messages()
|
|
|
|
analyzer.assert_called_once()
|
|
assert resolver.call_args.args[2] == "请看看这张图片。"
|
|
image_contexts = resolver.call_args.kwargs["image_contexts"]
|
|
assert image_contexts[0]["summary"] == "一张门诊病例"
|
|
assert image_contexts[0]["extractedText"] == "主诉：咳嗽三天"
|
|
assert [item["content"] for item in boxim.sent] == ["这份资料里写的是咳嗽三天。"]
|
|
|
|
db = session_factory()
|
|
try:
|
|
event = db.query(TakeoverMessage).filter_by(boxim_message_id="111").one()
|
|
task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="111").one()
|
|
assert event.attachment_id
|
|
assert db.get(ChatAttachment, event.attachment_id).uploader_kind == "boxim"
|
|
assert task.status == "sent"
|
|
with patch(
|
|
"services.takeover_service.download_boxim_image",
|
|
side_effect=AssertionError("cached image must not be downloaded again"),
|
|
):
|
|
cached = service._takeover_image_attachment(db, db.get(Avatar, "avatar-1"), event)
|
|
assert cached.id == event.attachment_id
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_followup_text_recovers_recent_image_recorded_without_task(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
image_message = {
|
|
"id": 113,
|
|
"localId": 113,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 1,
|
|
"content": json.dumps(
|
|
{
|
|
"originUrl": "https://cdn.example/case.png",
|
|
"thumbUrl": "https://cdn.example/case-thumb.png",
|
|
}
|
|
),
|
|
}
|
|
|
|
db = session_factory()
|
|
try:
|
|
avatar = db.get(Avatar, "avatar-1")
|
|
service._record_message(db, avatar, "100", image_message, schedule_reply=False)
|
|
cursor = db.query(TakeoverCursor).one()
|
|
cursor.last_message_id = "113"
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
clock.advance(60)
|
|
boxim.messages.extend(
|
|
[
|
|
image_message,
|
|
{
|
|
"id": 114,
|
|
"localId": 114,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 0,
|
|
"content": "请帮我看看这张图",
|
|
},
|
|
]
|
|
)
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="114").one()
|
|
assert task.source_message_ids == ["113", "114"]
|
|
assert task.prompt == "请看看这张图片。\n请帮我看看这张图"
|
|
finally:
|
|
db.close()
|
|
|
|
clock.advance(1)
|
|
boxim.messages.append(
|
|
{
|
|
"id": 115,
|
|
"localId": 115,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 0,
|
|
"content": "图里写了什么",
|
|
}
|
|
)
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
latest = db.query(TakeoverReplyTask).filter_by(trigger_message_id="115").one()
|
|
assert latest.source_message_ids == ["113", "114", "115"]
|
|
assert latest.source_message_ids.count("113") == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_explicit_followup_reuses_handled_image_within_two_days(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
image_message = {
|
|
"id": 116,
|
|
"localId": 116,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 1,
|
|
"content": json.dumps({"originUrl": "https://cdn.example/handled-case.png"}),
|
|
}
|
|
boxim.messages.append(image_message)
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
image_task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="116").one()
|
|
image_task.status = "sent"
|
|
image_task.sent_at = clock.now()
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
clock.advance(47 * 60 * 60)
|
|
boxim.messages.append(
|
|
{
|
|
"id": 117,
|
|
"localId": 117,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 0,
|
|
"content": "重新看一下刚才那张病例图片",
|
|
}
|
|
)
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="117").one()
|
|
assert task.source_message_ids == ["116", "117"]
|
|
assert task.prompt == "请看看这张图片。\n重新看一下刚才那张病例图片"
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_image_message_is_recorded_but_not_scheduled(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{
|
|
"id": 112,
|
|
"localId": 112,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 1,
|
|
"content": json.dumps({"width": 100, "height": 100}),
|
|
}
|
|
)
|
|
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
assert db.query(TakeoverMessage).filter_by(boxim_message_id="112").one()
|
|
assert db.query(TakeoverReplyTask).count() == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_reply_delay_is_three_minutes(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
db = session_factory()
|
|
try:
|
|
avatar = db.query(Avatar).one()
|
|
avatar.config = {"authorizationPermissions": ["chat", "takeover"]}
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 12, "localId": 12, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "三分钟后回复"}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).one()
|
|
assert task.scheduled_at == clock.now() + timedelta(seconds=180)
|
|
finally:
|
|
db.close()
|
|
|
|
clock.advance(179)
|
|
await service.process_reply_tasks()
|
|
assert boxim.sent == []
|
|
clock.advance(1)
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "好的"}):
|
|
await service.poll_and_process_messages()
|
|
assert [item["content"] for item in boxim.sent] == ["好的"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_avatar_owners_are_polled_concurrently(service_context):
|
|
session_factory, _service, _boxim, clock = service_context
|
|
db = session_factory()
|
|
try:
|
|
db.add_all(
|
|
[
|
|
User(
|
|
id="owner-local-2",
|
|
huihui_user_id="owner-huihui-2",
|
|
huihui_token="prod-huihui-token-2",
|
|
app_token="app-token-2",
|
|
),
|
|
Avatar(
|
|
id="avatar-2",
|
|
owner_id="owner-huihui-2",
|
|
name="分身二",
|
|
status="active",
|
|
config={"authorizationPermissions": ["chat", "takeover"]},
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
boxim = ConcurrentMessagePollingBoxIM()
|
|
service = TakeoverService(
|
|
session_factory,
|
|
boxim,
|
|
poll_concurrency=2,
|
|
now=clock.now,
|
|
)
|
|
|
|
await service.poll_messages()
|
|
|
|
assert boxim.peak_active_polls == 2
|
|
db = session_factory()
|
|
try:
|
|
assert db.query(TakeoverCursor).filter(TakeoverCursor.initialized.is_(True)).count() == 2
|
|
assert db.query(TakeoverMessage).count() == 2
|
|
assert len(boxim.read_receipts) == 2
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delayed_poll_still_schedules_recent_message(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_messages()
|
|
delayed_send_time = int(
|
|
(clock.value - timedelta(seconds=150)).replace(tzinfo=timezone.utc).timestamp()
|
|
* 1000
|
|
)
|
|
boxim.messages.append(
|
|
{
|
|
"id": 13,
|
|
"localId": 13,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": delayed_send_time,
|
|
"type": 0,
|
|
"content": "排队后仍需回复",
|
|
}
|
|
)
|
|
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).one()
|
|
assert task.status == "pending"
|
|
assert task.scheduled_at == clock.now()
|
|
finally:
|
|
db.close()
|
|
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "已经收到"}):
|
|
await service.process_reply_tasks()
|
|
assert [item["content"] for item in boxim.sent] == ["已经收到"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_avatar_origin_message_never_schedules_a_reply(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
local_id = _avatar_local_id("peer-owner", "peer-trigger")
|
|
boxim.messages.append(
|
|
{"id": 15, "localId": local_id, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "另一端分身回复"}
|
|
)
|
|
|
|
with patch("routers.chat._resolve_reply") as resolver:
|
|
await service.poll_and_process_messages()
|
|
|
|
resolver.assert_not_called()
|
|
db = session_factory()
|
|
try:
|
|
event = db.query(TakeoverMessage).filter(TakeoverMessage.boxim_message_id == "15").one()
|
|
assert event.is_avatar is True
|
|
assert db.query(TakeoverReplyTask).count() == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_peer_avatar_messages_are_excluded_from_later_human_context(service_context):
|
|
_session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{
|
|
"id": 16,
|
|
"localId": _avatar_local_id("peer-owner", "peer-trigger"),
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 0,
|
|
"content": "分身生成的夸张长文",
|
|
}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(1)
|
|
boxim.messages.append(
|
|
{
|
|
"id": 17,
|
|
"localId": 17,
|
|
"sendId": 200,
|
|
"recvId": 100,
|
|
"sendTime": clock.millis(),
|
|
"type": 0,
|
|
"content": "真人的新问题",
|
|
}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
clock.advance(3)
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "正常回复"}) as resolver:
|
|
await service.poll_and_process_messages()
|
|
|
|
assert resolver.call_args.args[3] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_owner_message_pauses_future_takeover_for_ten_minutes(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 18, "localId": 18, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我先来回复"}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(30)
|
|
boxim.messages.append(
|
|
{"id": 19, "localId": 19, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "收到"}
|
|
)
|
|
with patch("routers.chat._resolve_reply") as resolver:
|
|
await service.poll_and_process_messages()
|
|
|
|
resolver.assert_not_called()
|
|
db = session_factory()
|
|
try:
|
|
assert db.query(TakeoverReplyTask).count() == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_avatar_local_id_is_deterministic_and_self_describing():
|
|
first = _avatar_local_id("owner", "message-1")
|
|
assert first == _avatar_local_id("owner", "message-1")
|
|
assert first != _avatar_local_id("owner", "message-2")
|
|
assert first.startswith(AVATAR_LOCAL_ID_PREFIX)
|
|
assert len(first) == 18
|
|
assert first.isdigit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_contacts_generate_without_blocking_each_other(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.extend(
|
|
[
|
|
{"id": 13, "localId": 31, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "联系人甲"},
|
|
{"id": 14, "localId": 32, "sendId": 300, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "联系人乙"},
|
|
]
|
|
)
|
|
both_generating = Barrier(2, timeout=2)
|
|
|
|
def resolve(_db, _avatar, prompt, _history, **_kwargs):
|
|
both_generating.wait()
|
|
return {"answer": f"回复{prompt[-1]}"}
|
|
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(3)
|
|
with patch("routers.chat._resolve_reply", side_effect=resolve):
|
|
await service.poll_and_process_messages()
|
|
assert {(item["peerId"], item["content"]) for item in boxim.sent} == {
|
|
("200", "回复甲"),
|
|
("300", "回复乙"),
|
|
}
|
|
|
|
db = session_factory()
|
|
try:
|
|
assert {task.status for task in db.query(TakeoverReplyTask).all()} == {"sent"}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_receipt_failure_does_not_advance_cursor(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 12, "localId": 3, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "未读消息"}
|
|
)
|
|
boxim.mark_private_messages_read = AsyncMock(side_effect=BoxIMError("回执失败"))
|
|
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "稍后回复"}):
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
cursor = db.query(TakeoverCursor).one()
|
|
assert cursor.last_message_id == "0"
|
|
assert db.query(TakeoverMessage).count() == 0
|
|
assert db.query(TakeoverReplyTask).count() == 0
|
|
finally:
|
|
db.close()
|
|
|
|
boxim.mark_private_messages_read = AsyncMock(return_value=None)
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "稍后回复"}):
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
assert db.query(TakeoverCursor).one().last_message_id == "12"
|
|
assert db.query(TakeoverMessage).count() == 1
|
|
assert db.query(TakeoverReplyTask).count() == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_owner_message_cancels_pending_reply(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 21, "localId": 3, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "在吗"}
|
|
)
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "在的"}):
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(2)
|
|
boxim.messages.append(
|
|
{"id": 22, "localId": 4, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我来回复"}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
clock.advance(2)
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.trigger_message_id == "21").one()
|
|
assert task.status == "cancelled"
|
|
assert task.cancel_reason == "owner_replied"
|
|
assert boxim.sent == []
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_owner_message_in_final_second_wins_before_generation(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 23, "localId": 23, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "还在吗"}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(2)
|
|
boxim.messages.append(
|
|
{"id": 24, "localId": 24, "sendId": 100, "recvId": 200, "sendTime": clock.millis(), "type": 0, "content": "我来处理"}
|
|
)
|
|
clock.advance(1)
|
|
with patch("routers.chat._resolve_reply") as resolver:
|
|
await service.poll_and_process_messages()
|
|
|
|
resolver.assert_not_called()
|
|
assert boxim.sent == []
|
|
db = session_factory()
|
|
try:
|
|
task = db.query(TakeoverReplyTask).filter(TakeoverReplyTask.trigger_message_id == "23").one()
|
|
assert task.status == "cancelled"
|
|
assert task.cancel_reason == "owner_replied"
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quick_successive_messages_are_coalesced_into_one_reply(service_context):
|
|
session_factory, service, boxim, clock = service_context
|
|
await service.poll_and_process_messages()
|
|
boxim.messages.append(
|
|
{"id": 31, "localId": 5, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第一句"}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(1)
|
|
boxim.messages.append(
|
|
{"id": 32, "localId": 6, "sendId": 200, "recvId": 100, "sendTime": clock.millis(), "type": 0, "content": "第二句"}
|
|
)
|
|
await service.poll_and_process_messages()
|
|
|
|
clock.advance(3)
|
|
with patch("routers.chat._resolve_reply", return_value={"answer": "合并回复"}) as resolver:
|
|
await service.poll_and_process_messages()
|
|
assert resolver.call_args.args[2] == "第一句\n第二句"
|
|
assert [item["content"] for item in boxim.sent] == ["合并回复"]
|
|
|
|
db = session_factory()
|
|
try:
|
|
tasks = db.query(TakeoverReplyTask).order_by(TakeoverReplyTask.created_at).all()
|
|
assert [task.status for task in tasks] == ["cancelled", "sent"]
|
|
assert tasks[0].cancel_reason == "newer_incoming_message"
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_failure_disables_takeover_and_stops_retrying(service_context):
|
|
session_factory, service, boxim, _ = service_context
|
|
boxim.exchange_access_token = AsyncMock(
|
|
side_effect=BoxIMError("无效的访问令牌", code=40101, auth_error=True)
|
|
)
|
|
|
|
await service.poll_and_process_messages()
|
|
await service.poll_and_process_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
avatar = db.query(Avatar).one()
|
|
cursor = db.query(TakeoverCursor).one()
|
|
assert "takeover" not in avatar.config["authorizationPermissions"]
|
|
assert cursor.initialized is False
|
|
assert "重新登录" in cursor.last_error
|
|
assert db.query(TakeoverReplyTask).count() == 0
|
|
finally:
|
|
db.close()
|
|
boxim.exchange_access_token.assert_awaited_once_with("prod-huihui-token")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transient_connection_failure_keeps_takeover_and_recovers(service_context):
|
|
session_factory, service, boxim, _ = service_context
|
|
boxim.exchange_access_token = AsyncMock(
|
|
side_effect=[
|
|
BoxIMError("连接超时"),
|
|
{"accessToken": "box-token", "accessTokenExpiresIn": 3600},
|
|
]
|
|
)
|
|
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
avatar = db.query(Avatar).one()
|
|
cursor = db.query(TakeoverCursor).one()
|
|
assert "takeover" in avatar.config["authorizationPermissions"]
|
|
assert cursor.initialized is False
|
|
assert "暂时连接失败" in cursor.last_error
|
|
finally:
|
|
db.close()
|
|
|
|
await service.poll_messages()
|
|
|
|
db = session_factory()
|
|
try:
|
|
avatar = db.query(Avatar).one()
|
|
cursor = db.query(TakeoverCursor).one()
|
|
assert "takeover" in avatar.config["authorizationPermissions"]
|
|
assert cursor.initialized is True
|
|
assert cursor.last_error == ""
|
|
finally:
|
|
db.close()
|
|
assert boxim.exchange_access_token.await_count == 2
|
|
|
|
|
|
def test_plain_text_reply_removes_markdown_and_empty_lines():
|
|
assert _plain_text_reply("## 建议\n\n**不能自行用药**\n`必要时就医`") == "建议\n不能自行用药\n必要时就医"
|