105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
from django.utils import timezone
|
||
import pytest
|
||
|
||
from review_agent.models import (
|
||
Conversation,
|
||
FeishuAccessTokenCache,
|
||
FeishuQuestionLog,
|
||
FeishuUserMapping,
|
||
FileSummaryBatch,
|
||
WorkflowNotificationRecord,
|
||
)
|
||
|
||
|
||
pytestmark = pytest.mark.django_db
|
||
|
||
|
||
def test_feishu_user_mapping_preferred_identifier(django_user_model):
|
||
user = django_user_model.objects.create_user(username="owner", password="pass")
|
||
mapping = FeishuUserMapping.objects.create(
|
||
system_user=user,
|
||
feishu_display_name="负责人",
|
||
feishu_open_id="ou_open",
|
||
feishu_user_id="user_id",
|
||
feishu_mobile="13800000000",
|
||
)
|
||
|
||
assert mapping.preferred_identifier() == ("open_id", "ou_open")
|
||
|
||
mapping.feishu_open_id = ""
|
||
assert mapping.preferred_identifier() == ("user_id", "user_id")
|
||
|
||
mapping.feishu_user_id = ""
|
||
assert mapping.preferred_identifier() == ("mobile", "13800000000")
|
||
|
||
|
||
def test_feishu_access_token_cache_expiry():
|
||
now = timezone.now()
|
||
cache = FeishuAccessTokenCache.objects.create(
|
||
app_id_hash="hash",
|
||
tenant_access_token="token",
|
||
expires_at=now + timezone.timedelta(minutes=5),
|
||
)
|
||
|
||
assert cache.is_valid(now=now)
|
||
|
||
cache.expires_at = now - timezone.timedelta(seconds=1)
|
||
assert not cache.is_valid(now=now)
|
||
|
||
|
||
def test_workflow_notification_success_dedupe_only_success(django_user_model):
|
||
user = django_user_model.objects.create_user(username="owner", password="pass")
|
||
conversation = Conversation.objects.create(user=user, title="飞书")
|
||
batch = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-FEISHU",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
)
|
||
dedupe_key = WorkflowNotificationRecord.build_dedupe_key("file_summary", batch.pk, "success")
|
||
WorkflowNotificationRecord.objects.create(
|
||
workflow_type="file_summary",
|
||
workflow_batch_id=batch.pk,
|
||
workflow_batch_no=batch.batch_no,
|
||
workflow_status="success",
|
||
dedupe_key=dedupe_key,
|
||
trigger_user=user,
|
||
channel=WorkflowNotificationRecord.Channel.FEISHU_API,
|
||
send_status=WorkflowNotificationRecord.SendStatus.FAILED,
|
||
message_title="失败通知",
|
||
)
|
||
|
||
assert not WorkflowNotificationRecord.already_successfully_sent(dedupe_key)
|
||
|
||
WorkflowNotificationRecord.objects.create(
|
||
workflow_type="file_summary",
|
||
workflow_batch_id=batch.pk,
|
||
workflow_batch_no=batch.batch_no,
|
||
workflow_status="success",
|
||
dedupe_key=dedupe_key,
|
||
trigger_user=user,
|
||
channel=WorkflowNotificationRecord.Channel.FEISHU_API,
|
||
send_status=WorkflowNotificationRecord.SendStatus.SUCCESS,
|
||
message_title="成功通知",
|
||
)
|
||
|
||
assert WorkflowNotificationRecord.already_successfully_sent(dedupe_key)
|
||
|
||
|
||
def test_feishu_question_log_records_summary_without_full_answer(django_user_model):
|
||
user = django_user_model.objects.create_user(username="owner", password="pass")
|
||
log = FeishuQuestionLog.objects.create(
|
||
system_user=user,
|
||
source_type=FeishuQuestionLog.SourceType.SIMULATE,
|
||
question_text="查最新法规核查",
|
||
intent="batch_status",
|
||
query_object={"workflow_type": "regulatory_review", "latest": True},
|
||
answer_summary="RR-001 成功，阻断项 0，高风险 1。",
|
||
permission_result="allowed",
|
||
status=FeishuQuestionLog.Status.SUCCESS,
|
||
processed_at=timezone.now(),
|
||
)
|
||
|
||
assert "完整回答" not in log.answer_summary
|
||
assert log.query_object["latest"] is True
|