dealerhub/backend/tests/test_notify.py

"""消息通知与智能预警集成测试。"""

import pytest
from datetime import date, timedelta
from decimal import Decimal
from model_bakery import baker
from rest_framework.test import APIClient
from rest_framework_simplejwt.tokens import RefreshToken

from apps.catalog.models import Product
from apps.inventory.models import Warehouse, Stock
from apps.partner.models import Customer
from apps.finance.models import Receivable
from apps.notify.models import Notification, AlertRule
from apps.notify import services as notify_services


@pytest.fixture
def auth_client(db, user, tenant):
    client = APIClient()
    refresh = RefreshToken.for_user(user)
    client.credentials(
        HTTP_AUTHORIZATION=f"Bearer {refresh.access_token}",
        HTTP_X_TENANT_ID=tenant.code,
    )
    return client


@pytest.mark.django_db
def test_send_notification(tenant, user):
    msg = notify_services.send_notification(
        tenant=tenant,
        title="系统通知",
        content="欢迎使用 dealerhub 经销商中枢",
        recipient=user,
        category="info",
        link="/dashboard/",
    )
    assert msg.title == "系统通知"
    assert msg.is_read is False
    assert msg.recipient == user


@pytest.mark.django_db
def test_inventory_low_alert_trigger(tenant):
    notify_services.init_default_alert_rules(tenant)

    wh = baker.make(Warehouse, tenant=tenant, code="WH_ALERT", name="警戒仓")
    p_low = baker.make(Product, tenant=tenant, code="P_LOW", name="缺货饮料")
    p_ok = baker.make(Product, tenant=tenant, code="P_OK", name="充足大米")

    # 低于阈值 (5 <= 10)
    baker.make(Stock, tenant=tenant, warehouse=wh, product=p_low, on_hand=Decimal("5.00"), locked=Decimal("0"))
    # 充足库存 (100 > 10)
    baker.make(Stock, tenant=tenant, warehouse=wh, product=p_ok, on_hand=Decimal("100.00"), locked=Decimal("0"))

    triggered = notify_services.check_inventory_low_alerts(tenant)
    assert triggered == 1

    note = Notification.objects.filter(tenant=tenant, category="warning").first()
    assert note is not None
    assert "缺货饮料" in note.title
    assert note.is_read is False


@pytest.mark.django_db
def test_receivable_overdue_alert_trigger(tenant):
    notify_services.init_default_alert_rules(tenant)
    cust = baker.make(Customer, tenant=tenant, code="C_DEBT", name="欠款客户")

    # 30天前到期，至今未还
    overdue_date = date.today() - timedelta(days=30)
    baker.make(
        Receivable,
        tenant=tenant,
        customer=cust,
        bill_no="RC-OVERDUE-01",
        bill_date=overdue_date - timedelta(days=30),
        due_date=overdue_date,
        total_amount=Decimal("10000.00"),
        paid_amount=Decimal("0.00"),
        status="open",
    )

    triggered = notify_services.check_receivable_overdue_alerts(tenant)
    assert triggered == 1

    note = Notification.objects.filter(tenant=tenant, category="warning", extra_data__receivable_id__isnull=False).first()
    assert note is not None
    assert "欠款客户" in note.title
    assert "催款提醒" in note.title


@pytest.mark.django_db
def test_notify_api_endpoints(auth_client, tenant):
    # 触发全量预警扫描
    r = auth_client.post("/api/v1/notify/rules/run-checks/")
    assert r.status_code == 200
    assert r.json()["ok"] is True

    # 写入一条通知并标记已读
    note = notify_services.send_notification(tenant, "测试待办", "内容", category="audit")
    r2 = auth_client.post(f"/api/v1/notify/messages/{note.id}/mark-read/")
    assert r2.status_code == 200
    assert r2.json()["is_read"] is True

    # 全部标记已读
    r3 = auth_client.post("/api/v1/notify/messages/mark-all-read/")
    assert r3.status_code == 200
    assert r3.json()["ok"] is True


@pytest.mark.django_db
def test_notification_list_is_scoped_to_broadcast_and_current_recipient(
    auth_client, tenant, other_tenant, user, django_user_model
):
    other_user = django_user_model.objects.create_user(
        username="notification-owner", password="password12345"
    )
    visible_broadcast = notify_services.send_notification(
        tenant, "租户广播", "所有人可见"
    )
    visible_personal = notify_services.send_notification(
        tenant, "我的通知", "当前用户可见", recipient=user
    )
    hidden_personal = notify_services.send_notification(
        tenant, "别人的通知", "其他用户不可见", recipient=other_user
    )
    hidden_tenant = notify_services.send_notification(
        other_tenant, "其他租户广播", "当前租户不可见"
    )

    response = auth_client.get("/api/v1/notify/messages/")

    assert response.status_code == 200
    returned_ids = {item["id"] for item in response.json()["results"]}
    assert returned_ids == {visible_broadcast.id, visible_personal.id}
    assert hidden_personal.id not in returned_ids
    assert hidden_tenant.id not in returned_ids


@pytest.mark.django_db
def test_mark_read_rejects_notification_owned_by_another_user(
    auth_client, tenant, user, django_user_model
):
    other_user = django_user_model.objects.create_user(
        username="different-notification-owner", password="password12345"
    )
    note = notify_services.send_notification(
        tenant, "私有通知", "只属于其他用户", recipient=other_user
    )

    response = auth_client.post(f"/api/v1/notify/messages/{note.id}/mark-read/")

    assert response.status_code == 404
    note.refresh_from_db()
    assert note.is_read is False
    assert note.read_at is None


@pytest.mark.django_db
def test_mark_all_read_only_updates_broadcast_and_current_user(
    auth_client, tenant, user, django_user_model
):
    other_user = django_user_model.objects.create_user(
        username="mark-all-other-owner", password="password12345"
    )
    broadcast = notify_services.send_notification(tenant, "广播", "所有人")
    mine = notify_services.send_notification(tenant, "我的", "当前用户", recipient=user)
    theirs = notify_services.send_notification(tenant, "别人的", "其他用户", recipient=other_user)

    response = auth_client.post("/api/v1/notify/messages/mark-all-read/")

    assert response.status_code == 200
    assert response.json()["updated_count"] == 2
    broadcast.refresh_from_db()
    mine.refresh_from_db()
    theirs.refresh_from_db()
    assert broadcast.is_read is True
    assert mine.is_read is True
    assert theirs.is_read is False