zkxw/backend/tests/test_basic_experiment_delivery_remind.py

"""基础实验交付日前一天服务号提醒单测。"""
from datetime import date, datetime
from unittest.mock import MagicMock, patch

from app.services.basic_experiment_delivery_remind import (
    format_delivery_product_name,
    run_delivery_remind_for_date,
)


def test_format_delivery_product_name_full():
    assert (
        format_delivery_product_name(
            project_name="NK扩增",
            spec_text="1e8",
            package_text="冻存管",
        )
        == "NK扩增（1e8、冻存管）"
    )


def test_format_delivery_product_name_name_and_spec_only():
    assert (
        format_delivery_product_name(
            project_name="NK扩增",
            spec_text="1e8",
            package_text="",
        )
        == "NK扩增（1e8）"
    )


def test_format_delivery_product_name_name_and_package_only():
    assert (
        format_delivery_product_name(
            project_name="NK扩增",
            spec_text="",
            package_text="冻存管",
        )
        == "NK扩增（冻存管）"
    )


def test_format_delivery_product_name_name_only():
    assert (
        format_delivery_product_name(
            project_name="NK扩增",
            spec_text="",
            package_text="",
        )
        == "NK扩增"
    )


def test_format_delivery_product_name_empty():
    assert (
        format_delivery_product_name(
            project_name="",
            spec_text="",
            package_text="",
        )
        == ""
    )


def test_send_basic_experiment_delivery_remind_builds_data():
    with patch(
        "app.services.wechat_mp_message._send_template"
    ) as send_tpl, patch("app.services.wechat_mp_message.settings") as st:
        st.WX_MP_TEMPLATE_BASIC_EXPERIMENT_DELIVERY = "tpl-be"
        st.WX_MP_TPL_BE_DELIVERY_FIELD_CUSTOMER = "thing6"
        st.WX_MP_TPL_BE_DELIVERY_FIELD_PRODUCT = "thing10"
        st.WX_MP_TPL_BE_DELIVERY_FIELD_QTY = "character_string8"
        st.WX_MP_TPL_BE_DELIVERY_FIELD_CONTACT = "thing7"
        st.WX_MP_TPL_BE_DELIVERY_FIELD_TIME = "time14"
        send_tpl.return_value = {"success": True, "msgid": "1"}

        from app.services.wechat_mp_message import send_basic_experiment_delivery_remind

        long_name = "这是一个超过二十个汉字的产品名称规格包装拼接"
        send_basic_experiment_delivery_remind(
            mp_openid="oid",
            customer_name="张三客户",
            product_name=long_name,
            quantity="3",
            contact_name="李联系人",
            time_text="2026-10-23 14:00",
        )
        assert send_tpl.called
        kwargs = send_tpl.call_args.kwargs
        assert kwargs["template_id"] == "tpl-be"
        assert kwargs["mp_openid"] == "oid"
        data = kwargs["data"]
        assert data["thing6"]["value"] == "张三客户"
        assert len(data["thing10"]["value"]) <= 20
        assert data["character_string8"]["value"] == "3"
        assert data["thing7"]["value"] == "李联系人"
        assert data["time14"]["value"] == "2026-10-23 14:00"


def _make_db_for_remind(
    *,
    schedules,
    already_success=False,
    user_openid="mp-oid",
):
    """构造轻量 MagicMock Session，覆盖 remind 扫描路径。"""
    db = MagicMock()

    candidate_q = MagicMock()
    candidate_q.join.return_value = candidate_q
    candidate_q.outerjoin.return_value = candidate_q
    candidate_q.filter.return_value = candidate_q
    candidate_q.order_by.return_value = candidate_q
    candidate_q.all.return_value = schedules

    success_q = MagicMock()
    success_q.filter.return_value = success_q
    success_q.first.return_value = MagicMock() if already_success else None

    user = MagicMock()
    user.wx_mp_openid = user_openid
    user_q = MagicMock()
    user_q.filter.return_value = user_q
    user_q.first.return_value = user

    def query_side_effect(*models):
        from app.models.models import BizWxNotifyLog, SysUser

        if len(models) > 1:
            return candidate_q
        model = models[0]
        if model is BizWxNotifyLog:
            return success_q
        if model is SysUser:
            return user_q
        return candidate_q

    db.query.side_effect = query_side_effect
    return db


def test_run_delivery_remind_sends_for_tomorrow_schedule():
    sched = MagicMock()
    sched.id = "sch-1"
    sched.tenant_no = "zkxw"
    sched.infusion_time = datetime(2026, 10, 23, 14, 0, 0)
    sched.spec_text = "1e8"
    sched.package_text = "冻存管"
    sched.schedule_status = "dsc"

    app = MagicMock()
    app.id = "app-1"
    app.tenant_no = "zkxw"
    app.create_by = "user-1"
    app.customer_name = "客户甲"
    app.customer_contact = "联系人乙"

    prod = MagicMock()
    prod.project_name = "NK扩增"
    prod.spec_text = "1e8"
    prod.package_text = "冻存管"
    prod.quantity = 2

    db = _make_db_for_remind(schedules=[(sched, app, prod)])

    with patch(
        "app.services.basic_experiment_delivery_remind._notify_enabled",
        return_value=True,
    ), patch(
        "app.services.wechat_mp_message.send_basic_experiment_delivery_remind"
    ) as send_fn:
        send_fn.return_value = {"success": True, "msgid": "m1"}
        stats = run_delivery_remind_for_date(db, today=date(2026, 10, 22))

    assert stats["candidates"] == 1
    assert stats["sent"] == 1
    assert send_fn.called
    kwargs = send_fn.call_args.kwargs
    assert kwargs["customer_name"] == "客户甲"
    assert kwargs["product_name"] == "NK扩增（1e8、冻存管）"
    assert kwargs["quantity"] == "2"
    assert kwargs["contact_name"] == "联系人乙"
    assert kwargs["time_text"] == "2026-10-23 14:00"
    assert db.add.called
    assert db.commit.called


def test_run_delivery_remind_skips_when_already_success():
    sched = MagicMock()
    sched.id = "sch-1"
    sched.tenant_no = "zkxw"
    sched.infusion_time = datetime(2026, 10, 23, 14, 0, 0)
    sched.spec_text = ""
    sched.package_text = ""

    app = MagicMock()
    app.id = "app-1"
    app.tenant_no = "zkxw"
    app.create_by = "user-1"
    app.customer_name = "客户甲"
    app.customer_contact = "联系人乙"

    prod = MagicMock()
    prod.project_name = "NK"
    prod.spec_text = ""
    prod.package_text = ""
    prod.quantity = 1

    db = _make_db_for_remind(
        schedules=[(sched, app, prod)], already_success=True
    )

    with patch(
        "app.services.basic_experiment_delivery_remind._notify_enabled",
        return_value=True,
    ), patch(
        "app.services.wechat_mp_message.send_basic_experiment_delivery_remind"
    ) as send_fn:
        send_fn.return_value = {"success": True}
        stats = run_delivery_remind_for_date(db, today=date(2026, 10, 22))

    assert stats["skipped"] == 1
    assert stats["sent"] == 0
    assert not send_fn.called


def test_run_delivery_remind_skips_without_openid():
    sched = MagicMock()
    sched.id = "sch-1"
    sched.tenant_no = "zkxw"
    sched.infusion_time = datetime(2026, 10, 23, 9, 0, 0)
    sched.spec_text = ""
    sched.package_text = ""

    app = MagicMock()
    app.id = "app-1"
    app.tenant_no = "zkxw"
    app.create_by = "user-1"
    app.customer_name = "客户甲"
    app.customer_contact = "联系人乙"

    prod = MagicMock()
    prod.project_name = "NK"
    prod.spec_text = ""
    prod.package_text = ""
    prod.quantity = 1

    db = _make_db_for_remind(
        schedules=[(sched, app, prod)], user_openid=""
    )

    with patch(
        "app.services.basic_experiment_delivery_remind._notify_enabled",
        return_value=True,
    ), patch(
        "app.services.wechat_mp_message.send_basic_experiment_delivery_remind"
    ) as send_fn:
        stats = run_delivery_remind_for_date(db, today=date(2026, 10, 22))

    assert stats["skipped"] == 1
    assert not send_fn.called


def test_run_delivery_remind_noop_when_disabled():
    db = MagicMock()
    with patch(
        "app.services.basic_experiment_delivery_remind._notify_enabled",
        return_value=False,
    ):
        stats = run_delivery_remind_for_date(db, today=date(2026, 10, 22))
    assert stats == {"candidates": 0, "sent": 0, "skipped": 0, "failed": 0}
    assert not db.query.called