feat: add feishu question preview services

This commit is contained in:
2026-06-07 22:16:36 +08:00
parent be7fbab0a0
commit bd9b2e872e
7 changed files with 288 additions and 0 deletions

View File

@@ -0,0 +1,92 @@
from io import StringIO
from django.core.management import call_command
import pytest
from review_agent.feishu_questions.intent import parse_question_intent
from review_agent.feishu_questions.query import query_batch_summary
from review_agent.feishu_questions.service import answer_question
from review_agent.models import Conversation, FeishuQuestionLog, FileSummaryBatch, RegulatoryReviewBatch
pytestmark = pytest.mark.django_db
def test_query_latest_regulatory_batch_for_owner(django_user_model):
user = django_user_model.objects.create_user(username="owner", password="pass")
conversation = Conversation.objects.create(user=user, title="会话")
summary = FileSummaryBatch.objects.create(conversation=conversation, user=user, batch_no="FS-001")
RegulatoryReviewBatch.objects.create(
conversation=conversation,
user=user,
source_summary_batch=summary,
batch_no="RR-001",
status=RegulatoryReviewBatch.Status.SUCCESS,
risk_summary={"blocking": 0, "high": 1},
)
result = query_batch_summary(user, workflow_type="regulatory_review", latest=True)
assert result["ok"]
assert result["batch_no"] == "RR-001"
assert "高风险 1" in result["answer_summary"]
def test_query_denies_other_users_batch(django_user_model):
owner = django_user_model.objects.create_user(username="owner", password="pass")
other = django_user_model.objects.create_user(username="other", password="pass")
conversation = Conversation.objects.create(user=owner, title="会话")
batch = FileSummaryBatch.objects.create(conversation=conversation, user=owner, batch_no="FS-PRIVATE")
result = query_batch_summary(other, batch_no=batch.batch_no)
assert not result["ok"]
assert result["permission_result"] == "denied"
def test_query_admin_can_access_other_users_batch(django_user_model):
owner = django_user_model.objects.create_user(username="owner", password="pass")
admin = django_user_model.objects.create_user(username="admin", password="pass", is_staff=True)
conversation = Conversation.objects.create(user=owner, title="会话")
FileSummaryBatch.objects.create(conversation=conversation, user=owner, batch_no="FS-ADMIN")
result = query_batch_summary(admin, batch_no="FS-ADMIN")
assert result["ok"]
assert result["permission_result"] == "allowed"
def test_parse_question_intent_recognizes_batch_latest_and_workflow():
parsed = parse_question_intent("查最新法规核查")
assert parsed["workflow_type"] == "regulatory_review"
assert parsed["latest"] is True
parsed = parse_question_intent("AFF-20260607-001 的 Word 在哪里")
assert parsed["workflow_type"] == "application_form_fill"
assert parsed["batch_no"] == "AFF-20260607-001"
assert parsed["intent"] == "export_summary"
def test_answer_question_records_log_without_full_answer(django_user_model):
user = django_user_model.objects.create_user(username="owner", password="pass")
conversation = Conversation.objects.create(user=user, title="会话")
FileSummaryBatch.objects.create(conversation=conversation, user=user, batch_no="FS-LOG")
result = answer_question(user, "查最新自动汇总")
log = FeishuQuestionLog.objects.get(pk=result["log_id"])
assert log.intent == "batch_status"
assert log.query_object["workflow_type"] == "file_summary"
assert log.answer_summary
assert len(log.answer_summary) <= 500
def test_feishu_question_simulate_command_outputs_summary(django_user_model):
user = django_user_model.objects.create_user(username="owner", password="pass")
conversation = Conversation.objects.create(user=user, title="会话")
FileSummaryBatch.objects.create(conversation=conversation, user=user, batch_no="FS-CMD")
output = StringIO()
call_command("feishu_question_simulate", "--username", user.username, "查最新自动汇总", stdout=output)
assert "FS-CMD" in output.getvalue()