From 9345cf4ad8040ed5f29072345698b63a97dd9cee Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Mon, 17 Aug 2026 18:29:20 +0800 Subject: [PATCH] fix(duty): allow re-saving historical assigns without duty staff role Co-authored-by: Cursor --- backend/app/services/duty_schedule.py | 29 ++++- backend/tests/test_duty_schedule.py | 175 ++++++++++++++++++++++++++ 2 files changed, 201 insertions(+), 3 deletions(-) diff --git a/backend/app/services/duty_schedule.py b/backend/app/services/duty_schedule.py index 8163f988a..0a7f5a9bc 100644 --- a/backend/app/services/duty_schedule.py +++ b/backend/app/services/duty_schedule.py @@ -290,6 +290,8 @@ def save_day( for slot_in in body.slots: order += 1 slot = existing_by_type.get(slot_in.shiftType) + # 已有 slot 上历史 assign 用户可再保存（D4）；新建 slot 必须具备值班人员角色（D7） + grandfathered_user_ids: set = set() if slot is None: slot = DutySlot( id=generate_id(), @@ -317,6 +319,11 @@ def save_day( .filter(DutyAssignment.duty_slot_id == slot.id, DutyAssignment.del_flag == 0) .all() ) + grandfathered_user_ids = { + a.user_id + for a in existing_assigns + if a.source == SOURCE_ASSIGN and a.user_id + } # body.assignments 视为该班次最终「指定」名单；grab 全部保留，除非超容时拒绝缩容 grab_rows = [a for a in existing_assigns if a.source == SOURCE_GRAB] new_assign_count = len(slot_in.assignments or []) @@ -352,7 +359,8 @@ def save_day( if not person_name: raise DutyBizError("指定员工姓名不能为空") if item.source == SOURCE_ASSIGN and uid: - assert_user_is_duty_staff(db, tenant_no, uid) + if uid not in grandfathered_user_ids: + assert_user_is_duty_staff(db, tenant_no, uid) db.add( DutyAssignment( id=generate_id(), @@ -444,13 +452,28 @@ def list_duty_staff_users(db: Session, tenant_no: str) -> List[Dict[str, str]]: def assert_user_is_duty_staff(db: Session, tenant_no: str, user_id: str) -> None: + uid = (user_id or "").strip() + if not uid: + raise DutyBizError("用户无效或不具备值班人员角色") + user = ( + db.query(SysUser) + .filter( + SysUser.id == uid, + SysUser.tenant_no == tenant_no, + SysUser.del_flag == 0, + SysUser.status == "1", + ) + .first() + ) + if not user: + raise DutyBizError("用户无效或不具备值班人员角色") role_id = resolve_duty_staff_role_id(db, tenant_no) if not role_id: raise DutyBizError("未配置「值班人员」角色，请联系管理员") link = ( db.query(SysUserRole) .filter( - SysUserRole.user_id == user_id, + SysUserRole.user_id == uid, SysUserRole.role_id == role_id, SysUserRole.tenant_no == tenant_no, SysUserRole.del_flag == 0, @@ -458,7 +481,7 @@ def assert_user_is_duty_staff(db: Session, tenant_no: str, user_id: str) -> None .first() ) if not link: - raise DutyBizError("仅可选择「值班人员」角色的用户") + raise DutyBizError("用户无效或不具备值班人员角色") def _list_public_users(db: Session, tenant_no: str) -> List[Dict[str, str]]: diff --git a/backend/tests/test_duty_schedule.py b/backend/tests/test_duty_schedule.py index abc675c90..dd2e6857e 100644 --- a/backend/tests/test_duty_schedule.py +++ b/backend/tests/test_duty_schedule.py @@ -13,6 +13,7 @@ from app.models.duty_models import ( SHIFT_EVENING, SHIFT_LABELS, SHIFT_MORNING, + SOURCE_ASSIGN, SOURCE_GRAB, ) from app.schemas.duty import DutyAssignmentIn, DutyDaySaveBody, DutySlotSaveIn @@ -375,3 +376,177 @@ def test_grab_rejects_non_duty_staff(): 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)