"""值班排班服务单测""" 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)