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 (
    DUTY_STAFF_ROLE_ID,
    DUTY_STAFF_ROLE_NAME,
    LINK_CLOSED,
    LINK_OPEN,
    SHIFT_EVENING,
    SHIFT_LABELS,
    SHIFT_MORNING,
    SOURCE_ASSIGN,
    SOURCE_GRAB,
)
from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn
from app.services.duty_schedule import (
    DutyBizError,
    assert_user_is_duty_staff,
    grab,
    list_duty_staff_users,
    normalize_person_name,
    resolve_duty_staff_role_id,
    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
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
    role_q = MagicMock()
    role_q.filter.return_value.first.return_value = role
    ur_q = MagicMock()
    ur_q.filter.return_value.first.return_value = SimpleNamespace(user_id="u1", role_id=DUTY_STAFF_ROLE_ID)

    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
        if name == "SysRole":
            return role_q
        if name == "SysUserRole":
            return ur_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
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
    role_q = MagicMock()
    role_q.filter.return_value.first.return_value = role
    ur_q = MagicMock()
    ur_q.filter.return_value.first.return_value = SimpleNamespace(user_id="u1", role_id=DUTY_STAFF_ROLE_ID)

    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
        if name == "SysRole":
            return role_q
        if name == "SysUserRole":
            return ur_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)


def test_resolve_duty_staff_role_id_by_fixed_id():
    db = MagicMock()
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID, role_name=DUTY_STAFF_ROLE_NAME, del_flag=0)
    q = MagicMock()
    q.filter.return_value.first.return_value = role
    db.query.return_value = q
    assert resolve_duty_staff_role_id(db, "shanxia2.0") == DUTY_STAFF_ROLE_ID


def test_list_duty_staff_users_empty_when_no_role():
    db = MagicMock()
    q = MagicMock()
    q.filter.return_value.first.return_value = None  # resolve role → None twice
    db.query.return_value = q
    assert list_duty_staff_users(db, "t1") == []


def test_assert_user_is_duty_staff_rejects_without_role():
    db = MagicMock()
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
    role_q = MagicMock()
    role_q.filter.return_value.first.return_value = role
    ur_q = MagicMock()
    ur_q.filter.return_value.first.return_value = None

    def query_side_effect(model):
        name = getattr(model, "__name__", "")
        if name == "SysRole":
            return role_q
        if name == "SysUserRole":
            return ur_q
        return MagicMock()

    db.query.side_effect = query_side_effect
    with pytest.raises(DutyBizError, match="值班人员"):
        assert_user_is_duty_staff(db, "t1", "u1")


def test_grab_rejects_non_duty_staff():
    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,
    )
    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
    user_q = MagicMock()
    user_q.filter.return_value.first.return_value = user
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
    role_q = MagicMock()
    role_q.filter.return_value.first.return_value = role
    ur_q = MagicMock()
    ur_q.filter.return_value.first.return_value = None

    def query_side_effect(model):
        name = getattr(model, "__name__", "")
        if name == "DutyDay":
            return day_q
        if name == "SysUser":
            return user_q
        if name == "SysRole":
            return role_q
        if name == "SysUserRole":
            return ur_q
        return MagicMock()

    db.query.side_effect = query_side_effect
    with pytest.raises(DutyBizError, match="值班人员"):
        grab(db, "tok", SHIFT_EVENING, "u1")


def test_save_day_grandfathers_existing_assign_without_role():
    """历史指定用户无值班人员角色时，原样再保存不得因角色校验失败。"""
    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,
    )
    old_assign = SimpleNamespace(
        id="a_old",
        duty_slot_id="slot1",
        source=SOURCE_ASSIGN,
        user_id="u_old",
        person_name="老员工",
        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]
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
    role_q = MagicMock()
    role_q.filter.return_value.first.return_value = role
    ur_q = MagicMock()
    ur_q.filter.return_value.first.return_value = None  # 无角色关联
    user_q = MagicMock()
    user_q.filter.return_value.first.return_value = SimpleNamespace(
        id="u_old", tenant_no="t1", del_flag=0, status="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":
            m = MagicMock()
            m.filter.return_value.count.return_value = 1
            m.filter.return_value.all.return_value = [old_assign]
            return m
        if name == "SysRole":
            return role_q
        if name == "SysUserRole":
            return ur_q
        if name == "SysUser":
            return user_q
        return MagicMock()

    db.query.side_effect = query_side_effect

    body = DutyDaySaveBody(
        dutyDate=date(2026, 8, 20),
        slots=[
            DutySlotSaveIn(
                shiftType=SHIFT_EVENING,
                capacity=2,
                assignments=[
                    DutyAssignmentIn(source="assign", userId="u_old", personName="老员工")
                ],
            )
        ],
    )

    # 不得因「无值班人员角色」抛错；序列化侧 mock 足够跑完即可
    result = save_day(db, "t1", {"user_id": "op1"}, body)
    assert result is not None
    db.commit.assert_called()


def test_save_day_rejects_new_assign_without_role():
    """新指定用户无值班人员角色时必须拒绝。"""
    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,
    )
    old_assign = SimpleNamespace(
        id="a_old",
        duty_slot_id="slot1",
        source=SOURCE_ASSIGN,
        user_id="u_old",
        person_name="老员工",
        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]
    role = SimpleNamespace(id=DUTY_STAFF_ROLE_ID)
    role_q = MagicMock()
    role_q.filter.return_value.first.return_value = role
    ur_q = MagicMock()
    ur_q.filter.return_value.first.return_value = None
    user_q = MagicMock()
    user_q.filter.return_value.first.return_value = SimpleNamespace(
        id="u_new", tenant_no="t1", del_flag=0, status="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":
            m = MagicMock()
            m.filter.return_value.count.return_value = 1
            m.filter.return_value.all.return_value = [old_assign]
            return m
        if name == "SysRole":
            return role_q
        if name == "SysUserRole":
            return ur_q
        if name == "SysUser":
            return user_q
        return MagicMock()

    db.query.side_effect = query_side_effect

    body = DutyDaySaveBody(
        dutyDate=date(2026, 8, 20),
        slots=[
            DutySlotSaveIn(
                shiftType=SHIFT_EVENING,
                capacity=2,
                assignments=[
                    DutyAssignmentIn(source="assign", userId="u_new", personName="新员工")
                ],
            )
        ],
    )

    with pytest.raises(DutyBizError, match="值班人员"):
        save_day(db, "t1", {"user_id": "op1"}, body)