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.models.models import SysDictData
from app.services.duty_schedule import (
    DutyBizError,
    assert_user_is_duty_staff,
    build_dict_value_from_label,
    ensure_dict_data,
    format_time_label,
    grab,
    join_work_plans,
    list_duty_staff_users,
    normalize_person_name,
    parse_hhmm,
    resolve_duty_staff_role_id,
    save_day,
    split_work_plans,
)


def test_parse_hhmm():
    assert parse_hhmm("08:00").hour == 8
    assert parse_hhmm("18:30").minute == 30
    assert parse_hhmm(None) is None
    assert parse_hhmm("") is None


def test_parse_hhmm_invalid_raises():
    with pytest.raises(DutyBizError, match="时间格式"):
        parse_hhmm("08:xx")


def test_parse_time_range_input_day():
    from app.services.duty_schedule import parse_time_range_input

    r = parse_time_range_input("09:00-17:00")
    assert r["startTime"] == "09:00"
    assert r["endTime"] == "17:00"
    assert r["crossDay"] is False
    assert r["value"] == "09:00|17:00|0"
    assert r["label"] == "09:00–17:00"


def test_parse_time_range_input_cross_day():
    from app.services.duty_schedule import parse_time_range_input

    r = parse_time_range_input("18:00-次日07:00")
    assert r["startTime"] == "18:00"
    assert r["endTime"] == "07:00"
    assert r["crossDay"] is True
    assert r["value"] == "18:00|07:00|1"
    assert "次日" in r["label"]


def test_format_time_label_same_day():
    assert format_time_label(parse_hhmm("08:00"), parse_hhmm("18:00"), False) == "08:00–18:00"


def test_format_time_label_cross_day():
    assert format_time_label(parse_hhmm("18:00"), parse_hhmm("07:00"), True) == "18:00–次日07:00"


def test_format_time_label_empty():
    assert format_time_label(None, None, False) is None


def test_work_plans_join_split():
    assert join_work_plans(["peisuan", "zhuangxie"]) == "peisuan,zhuangxie"
    assert split_work_plans("peisuan,zhuangxie") == ["peisuan", "zhuangxie"]
    assert split_work_plans(None) == []


def test_build_dict_value_from_label():
    assert build_dict_value_from_label("配酸") == "peisuan"
    assert build_dict_value_from_label("装卸") == "zhuangxie"
    assert build_dict_value_from_label("  ") == "item"


def test_ensure_dict_data_reuses_existing_label():
    db = MagicMock()
    existing = SimpleNamespace(dict_label="配酸", dict_value="peisuan")
    q = MagicMock()
    q.filter.return_value.first.return_value = existing
    db.query.return_value = q

    result = ensure_dict_data(
        db, tenant_no="t1", dict_type="duty_work_plan", label="配酸", user_id="u1"
    )
    assert result == {"label": "配酸", "value": "peisuan"}
    db.add.assert_not_called()


def test_ensure_dict_data_inserts_new_without_commit():
    db = MagicMock()
    q = MagicMock()
    q.filter.return_value.first.side_effect = [None, None]
    db.query.return_value = q

    result = ensure_dict_data(
        db, tenant_no="t1", dict_type="duty_work_plan", label="巡检", user_id="u1"
    )
    assert result["label"] == "巡检"
    assert result["value"] == "xunjian"
    db.add.assert_called_once()
    db.flush.assert_called_once()
    db.commit.assert_not_called()
    added = db.add.call_args[0][0]
    assert isinstance(added, SysDictData)
    assert added.dict_type == "duty_work_plan"
    assert added.tenant_no == "t1"


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_allows_custom_shift_type_when_slot_exists():
    """自定义 shift_type 只要当日存在 slot 即可，不必在 SHIFT_LABELS。"""
    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="zhongban",
        capacity=2,
        del_flag=0,
    )
    user = SimpleNamespace(
        id="u1",
        tenant_no="t1",
        nick_name="王五",
        user_no="wangwu",
        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
        if name == "SysDictData":
            dq = MagicMock()
            dq.filter.return_value.all.return_value = []
            dq.filter.return_value.first.return_value = None
            return dq
        aq = MagicMock()
        aq.filter.return_value.count.return_value = 0
        aq.filter.return_value.first.return_value = None
        return aq

    db.query.side_effect = query_side_effect
    result = grab(db, "tok", "zhongban", "u1")
    assert result["shiftType"] == "zhongban"
    assert result["status"] == "ok"


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)