110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
import pytest
|
|
|
|
from review_agent.models import (
|
|
Conversation,
|
|
FileSummaryBatch,
|
|
RegulatoryIssue,
|
|
RegulatoryNotificationRecord,
|
|
RegulatoryReviewBatch,
|
|
)
|
|
from review_agent.regulatory_review.services.export import build_markdown_report, build_result_payload
|
|
from review_agent.regulatory_review.services.feishu_notifier import create_mock_notifications
|
|
from review_agent.regulatory_review.workflow import RegulatoryWorkflowExecutor
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_create_mock_notifications_for_medium_and_above(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-NOTIFY",
|
|
status=FileSummaryBatch.Status.SUCCESS,
|
|
)
|
|
batch = RegulatoryReviewBatch.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
source_summary_batch=summary,
|
|
batch_no="RR-NOTIFY",
|
|
)
|
|
high = RegulatoryIssue.objects.create(
|
|
batch=batch,
|
|
rule_code="attachment4_1_2_application_form",
|
|
category=RegulatoryIssue.Category.COMPLETENESS,
|
|
severity=RegulatoryIssue.Severity.HIGH,
|
|
title="缺少申请表",
|
|
)
|
|
RegulatoryIssue.objects.create(
|
|
batch=batch,
|
|
rule_code="info",
|
|
category=RegulatoryIssue.Category.RAG,
|
|
severity=RegulatoryIssue.Severity.INFO,
|
|
title="提示项",
|
|
)
|
|
|
|
records = create_mock_notifications(batch)
|
|
|
|
assert len(records) == 1
|
|
assert records[0].channel == RegulatoryNotificationRecord.Channel.MOCK
|
|
assert records[0].status == RegulatoryNotificationRecord.Status.SENT
|
|
assert records[0].payload["issue_id"] == high.pk
|
|
|
|
|
|
def test_notification_records_enter_reports(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-NOTIFY",
|
|
status=FileSummaryBatch.Status.SUCCESS,
|
|
)
|
|
batch = RegulatoryReviewBatch.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
source_summary_batch=summary,
|
|
batch_no="RR-NOTIFY",
|
|
)
|
|
RegulatoryNotificationRecord.objects.create(
|
|
batch=batch,
|
|
channel=RegulatoryNotificationRecord.Channel.MOCK,
|
|
target="法规整改负责人",
|
|
status=RegulatoryNotificationRecord.Status.SENT,
|
|
payload={"title": "缺少申请表", "severity": "high"},
|
|
)
|
|
|
|
assert "通知记录" in build_markdown_report(batch)
|
|
assert build_result_payload(batch)["notifications"][0]["channel"] == "mock"
|
|
|
|
|
|
def test_regulatory_completion_notification_uses_dispatcher(monkeypatch, 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-NOTIFY-DISPATCH",
|
|
status=FileSummaryBatch.Status.SUCCESS,
|
|
)
|
|
batch = RegulatoryReviewBatch.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
source_summary_batch=summary,
|
|
batch_no="RR-NOTIFY-DISPATCH",
|
|
status=RegulatoryReviewBatch.Status.SUCCESS,
|
|
)
|
|
calls = []
|
|
|
|
monkeypatch.setattr(
|
|
"review_agent.regulatory_review.workflow.dispatch_workflow_notification",
|
|
lambda context: calls.append(context),
|
|
)
|
|
|
|
RegulatoryWorkflowExecutor(batch)._dispatch_completion_notification()
|
|
|
|
assert calls
|
|
assert calls[0].workflow_type == "regulatory_review"
|