126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app.model.fund import FundRiskAlert
|
|
from app.service.risk_notification_service import RiskNotificationService
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self):
|
|
self.added = []
|
|
|
|
def add(self, value):
|
|
self.added.append(value)
|
|
|
|
|
|
def alert(level: str = "高") -> FundRiskAlert:
|
|
return FundRiskAlert(
|
|
id=1,
|
|
alert_no="ALERT-001",
|
|
customer_id=9,
|
|
alert_type="适当性错配",
|
|
alert_level=level,
|
|
trigger_rule_codes=["RW-007"],
|
|
evidence_summary="C2 客户购买 R5 产品",
|
|
evidence_snapshot={},
|
|
priority_score=90,
|
|
event_status="正在发生",
|
|
status="待处理",
|
|
ack_status="未确认",
|
|
is_escalated=0,
|
|
created_at=datetime(2026, 9, 10),
|
|
updated_at=datetime(2026, 9, 10),
|
|
)
|
|
|
|
|
|
def test_in_app_notification_contains_alert_number() -> None:
|
|
session = FakeSession()
|
|
|
|
notification = RiskNotificationService(session).create_in_app(
|
|
alert(),
|
|
receiver_user_id=990000002,
|
|
title="高风险预警",
|
|
content="C2 客户购买 R5 产品",
|
|
)
|
|
|
|
assert notification.alert_id == 1
|
|
assert notification.channel == "站内提醒"
|
|
assert "预警编号：ALERT-001" in notification.content
|
|
assert notification.id > 0
|
|
assert session.added == [notification]
|
|
|
|
|
|
def test_disabled_mail_only_creates_record() -> None:
|
|
session = FakeSession()
|
|
|
|
notification = RiskNotificationService(session).create_mail_record(
|
|
alert(),
|
|
receiver_email="risk@example.com",
|
|
title="高风险预警",
|
|
content="证据摘要",
|
|
mail_enabled=False,
|
|
)
|
|
|
|
assert notification.channel == "邮件"
|
|
assert notification.send_status == "未启用"
|
|
assert notification.fail_reason == "邮件发送功能未启用"
|
|
|
|
|
|
def test_high_risk_batch_creates_in_app_and_mail_records() -> None:
|
|
session = FakeSession()
|
|
|
|
records = RiskNotificationService(session).create_high_risk_records(
|
|
[alert("高"), alert("低")],
|
|
receiver_user_id=990000002,
|
|
receiver_email="risk@example.com",
|
|
mail_enabled=False,
|
|
)
|
|
|
|
assert len(records) == 2
|
|
assert {record.channel for record in records} == {"站内提醒", "邮件"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notification_list_interprets_time_as_local_time(monkeypatch) -> None:
|
|
from app.api.schemas.risk import RiskNotificationPageQuery
|
|
from app.core.contracts import RequestContext
|
|
from app.repository.fund_query_repository import FundPage
|
|
|
|
captured: dict = {}
|
|
|
|
class StubRepository:
|
|
def __init__(self, _session, *, scope) -> None:
|
|
self.scope = scope
|
|
|
|
async def list_notifications(self, **kwargs) -> FundPage:
|
|
captured.update(kwargs)
|
|
return FundPage(
|
|
entity="risk_notification",
|
|
items=(),
|
|
limit=10,
|
|
offset=0,
|
|
next_offset=None,
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"app.service.risk_notification_service.RiskRepository",
|
|
StubRepository,
|
|
)
|
|
service = RiskNotificationService(object())
|
|
query = RiskNotificationPageQuery(
|
|
start_time=datetime(2026, 9, 10, 12, 0),
|
|
end_time=datetime(2026, 9, 10, 13, 0),
|
|
)
|
|
context = RequestContext(
|
|
user_id="990000002",
|
|
trace_id="trace",
|
|
permissions=("risk:alert:read",),
|
|
data_scope="all",
|
|
)
|
|
|
|
await service.list_notifications(context, query)
|
|
|
|
assert captured["start_time"] == datetime(2026, 9, 10, 4, 0)
|
|
assert captured["end_time"] == datetime(2026, 9, 10, 5, 0)
|