shanxia2.0/backend/tests/test_duty_schedule.py

"""值班排班服务单测"""
from datetime import date
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

from app.models.duty_models import (
    LINK_CLOSED,
    LINK_OPEN,
    SHIFT_EVENING,
    SHIFT_LABELS,
    SHIFT_MORNING,
    SOURCE_GRAB,
)
from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn
from app.services.duty_schedule import (
    DutyBizError,
    grab,
    normalize_person_name,
    save_day,
)


def test_normalize_person_name_ok():
    assert normalize_person_name("  张三  ") == "张三"


def test_normalize_person_name_empty():
    with pytest.raises(DutyBizError, match="姓名"):
        normalize_person_name("   ")


def test_normalize_person_name_too_long():
    with pytest.raises(DutyBizError, match="姓名"):
        normalize_person_name("啊" * 21)


def test_shift_labels():
    assert SHIFT_LABELS[SHIFT_MORNING] == "早班"


def test_grab_rejects_when_full():
    db = MagicMock()
    day = SimpleNamespace(
        id="day1",
        tenant_no="t1",
        link_status=LINK_OPEN,
        duty_date=date(2026, 8, 20),
        access_token="tok",
        del_flag=0,
    )
    slot = SimpleNamespace(
        id="slot1",
        duty_day_id="day1",
        shift_type=SHIFT_EVENING,
        capacity=1,
        del_flag=0,
    )
    user = SimpleNamespace(
        id="u1",
        tenant_no="t1",
        nick_name="李四",
        user_no="lisi",
        status="1",
        del_flag=0,
    )

    day_q = MagicMock()
    day_q.filter.return_value.with_for_update.return_value.first.return_value = day
    slot_q = MagicMock()
    slot_q.filter.return_value.with_for_update.return_value.first.return_value = slot
    user_q = MagicMock()
    user_q.filter.return_value.first.return_value = user

    def query_side_effect(model):
        name = getattr(model, "__name__", str(model))
        if name == "DutyDay":
            return day_q
        if name == "DutySlot":
            return slot_q
        if name == "SysUser":
            return user_q
        # DutyAssignment count / duplicate
        aq = MagicMock()
        aq.filter.return_value.count.return_value = 1
        aq.filter.return_value.first.return_value = None
        return aq

    db.query.side_effect = query_side_effect

    with pytest.raises(DutyBizError, match="名额已满"):
        grab(db, "tok", SHIFT_EVENING, "u1")


def test_grab_rejects_when_closed():
    db = MagicMock()
    day = SimpleNamespace(
        id="day1",
        tenant_no="t1",
        link_status=LINK_CLOSED,
        duty_date=date(2026, 8, 20),
        access_token="tok",
        del_flag=0,
    )
    day_q = MagicMock()
    day_q.filter.return_value.with_for_update.return_value.first.return_value = day
    db.query.return_value = day_q

    with pytest.raises(DutyBizError, match="关闭"):
        grab(db, "tok", SHIFT_EVENING, "u1")


def test_grab_rejects_invalid_user():
    db = MagicMock()
    day = SimpleNamespace(
        id="day1",
        tenant_no="t1",
        link_status=LINK_OPEN,
        duty_date=date(2026, 8, 20),
        access_token="tok",
        del_flag=0,
    )
    day_q = MagicMock()
    day_q.filter.return_value.with_for_update.return_value.first.return_value = day
    user_q = MagicMock()
    user_q.filter.return_value.first.return_value = None

    def query_side_effect(model):
        name = getattr(model, "__name__", str(model))
        if name == "DutyDay":
            return day_q
        if name == "SysUser":
            return user_q
        return MagicMock()

    db.query.side_effect = query_side_effect

    with pytest.raises(DutyBizError, match="用户|无效|不存在"):
        grab(db, "tok", SHIFT_EVENING, "bad-user")


def test_grab_rejects_duplicate_same_shift():
    db = MagicMock()
    day = SimpleNamespace(
        id="day1",
        tenant_no="t1",
        link_status=LINK_OPEN,
        duty_date=date(2026, 8, 20),
        access_token="tok",
        del_flag=0,
    )
    slot = SimpleNamespace(
        id="slot1",
        duty_day_id="day1",
        shift_type=SHIFT_EVENING,
        capacity=2,
        del_flag=0,
    )
    user = SimpleNamespace(
        id="u1",
        tenant_no="t1",
        nick_name="李四",
        user_no="lisi",
        status="1",
        del_flag=0,
    )
    existing = SimpleNamespace(id="a1", user_id="u1", del_flag=0)

    day_q = MagicMock()
    day_q.filter.return_value.with_for_update.return_value.first.return_value = day
    slot_q = MagicMock()
    slot_q.filter.return_value.with_for_update.return_value.first.return_value = slot
    user_q = MagicMock()
    user_q.filter.return_value.first.return_value = user

    assign_calls = {"n": 0}

    def query_side_effect(model):
        name = getattr(model, "__name__", str(model))
        if name == "DutyDay":
            return day_q
        if name == "DutySlot":
            return slot_q
        if name == "SysUser":
            return user_q
        aq = MagicMock()
        # first: duplicate check .first(); later: count
        assign_calls["n"] += 1
        aq.filter.return_value.first.return_value = existing
        aq.filter.return_value.count.return_value = 1
        return aq

    db.query.side_effect = query_side_effect

    with pytest.raises(DutyBizError, match="已报名"):
        grab(db, "tok", SHIFT_EVENING, "u1")


def test_save_day_rejects_shrink_below_grab_count():
    """缩容：已有 1 人抢班时 capacity=1 且再指定 1 人应失败。"""
    db = MagicMock()
    day = SimpleNamespace(
        id="day1",
        duty_date=date(2026, 8, 20),
        access_token="tok",
        link_status=LINK_OPEN,
        tenant_no="t1",
        del_flag=0,
        update_by=None,
        update_time=None,
    )
    slot = SimpleNamespace(
        id="slot1",
        duty_day_id="day1",
        shift_type=SHIFT_EVENING,
        capacity=2,
        del_flag=0,
        order_num=1,
        update_by=None,
        update_time=None,
    )
    grab_row = SimpleNamespace(
        id="a1",
        duty_slot_id="slot1",
        source=SOURCE_GRAB,
        del_flag=0,
        update_by=None,
        update_time=None,
    )

    day_q = MagicMock()
    day_q.filter.return_value.first.return_value = day
    slot_list_q = MagicMock()
    slot_list_q.filter.return_value.all.return_value = [slot]
    assign_q = MagicMock()
    assign_q.filter.return_value.all.return_value = [grab_row]
    count_q = MagicMock()
    count_q.filter.return_value.count.return_value = 1

    def query_side_effect(model):
        name = getattr(model, "__name__", "")
        if name == "DutyDay":
            return day_q
        if name == "DutySlot":
            return slot_list_q
        if name == "DutyAssignment":
            # first call in loop may be count, then all
            m = MagicMock()
            m.filter.return_value.count.return_value = 1
            m.filter.return_value.all.return_value = [grab_row]
            return m
        return MagicMock()

    db.query.side_effect = query_side_effect

    body = DutyDaySaveBody(
        dutyDate=date(2026, 8, 20),
        slots=[
            DutySlotSaveIn(
                shiftType=SHIFT_EVENING,
                capacity=1,
                assignments=[
                    DutyAssignmentIn(source="assign", userId="u1", personName="张三")
                ],
            )
        ],
    )

    with pytest.raises(DutyBizError, match="超过名额|缩容|报名"):
        save_day(db, "t1", {"user_id": "op1"}, body)