feat(duty): add schemas and duty_schedule service with capacity tests
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
f5ad2d7217
commit
49cad98c84
3 files changed
+683
No files matched your search
@@ -0,0 +1,38 @@
|
||||
"""值班排班 Pydantic schemas"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import List, Literal, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
ShiftType = Literal["morning", "evening", "duty"]
|
||||
LinkStatus = Literal["open", "closed"]
|
||||
SourceType = Literal["assign", "grab"]
|
||||
|
||||
|
||||
class DutyAssignmentIn(BaseModel):
|
||||
source: SourceType = "assign"
|
||||
userId: Optional[str] = None
|
||||
personName: str = ""
|
||||
|
||||
|
||||
class DutySlotSaveIn(BaseModel):
|
||||
shiftType: ShiftType
|
||||
capacity: int = Field(ge=1, le=99)
|
||||
assignments: List[DutyAssignmentIn] = []
|
||||
|
||||
|
||||
class DutyDaySaveBody(BaseModel):
|
||||
dutyDate: date
|
||||
slots: List[DutySlotSaveIn]
|
||||
linkStatus: Optional[LinkStatus] = None
|
||||
|
||||
|
||||
class DutyLinkBody(BaseModel):
|
||||
linkStatus: LinkStatus
|
||||
|
||||
|
||||
class DutyGrabBody(BaseModel):
|
||||
shiftType: ShiftType
|
||||
personName: str
|
||||
@@ -0,0 +1,473 @@
|
||||
"""值班排班领域服务 — docs/superpowers/specs/2026-08-14-duty-schedule-design.md"""
|
||||
from __future__ import annotations
|
||||
|
||||
import calendar
|
||||
import secrets
|
||||
from datetime import date, datetime
|
||||
from typing import Any, Dict, List, Optional, Sequence
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.utils import generate_id
|
||||
from app.models.duty_models import (
|
||||
LINK_CLOSED,
|
||||
LINK_OPEN,
|
||||
SHIFT_LABELS,
|
||||
SOURCE_ASSIGN,
|
||||
SOURCE_GRAB,
|
||||
DutyAssignment,
|
||||
DutyDay,
|
||||
DutySlot,
|
||||
)
|
||||
from app.schemas.duty import DutyDaySaveBody, DutySlotSaveIn
|
||||
|
||||
PERSON_NAME_MAX_LEN = 20
|
||||
|
||||
|
||||
class DutyBizError(ValueError):
|
||||
"""业务错误（可映射为 HTTP 400）"""
|
||||
|
||||
|
||||
def normalize_person_name(name: str) -> str:
|
||||
raw = (name or "").strip()
|
||||
if not raw:
|
||||
raise DutyBizError("姓名不能为空")
|
||||
if len(raw) > PERSON_NAME_MAX_LEN:
|
||||
raise DutyBizError(f"姓名不能超过{PERSON_NAME_MAX_LEN}个字符")
|
||||
return raw
|
||||
|
||||
|
||||
def new_access_token() -> str:
|
||||
return secrets.token_urlsafe(24)[:64]
|
||||
|
||||
|
||||
def build_public_url(token: str) -> str:
|
||||
origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/")
|
||||
return f"{origin}/#/public/duty/{token}"
|
||||
|
||||
|
||||
def _assignment_dict(row: DutyAssignment) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": row.id,
|
||||
"source": row.source,
|
||||
"userId": row.user_id,
|
||||
"personName": row.person_name,
|
||||
}
|
||||
|
||||
|
||||
def _slot_dict(slot: DutySlot, assignments: Sequence[DutyAssignment]) -> Dict[str, Any]:
|
||||
active = [a for a in assignments if a.del_flag == 0]
|
||||
return {
|
||||
"id": slot.id,
|
||||
"shiftType": slot.shift_type,
|
||||
"shiftLabel": SHIFT_LABELS.get(slot.shift_type, slot.shift_type),
|
||||
"capacity": slot.capacity,
|
||||
"taken": len(active),
|
||||
"assignments": [_assignment_dict(a) for a in active],
|
||||
}
|
||||
|
||||
|
||||
def _day_dict(
|
||||
day: DutyDay,
|
||||
slots: Sequence[DutySlot],
|
||||
assignments_by_slot: Dict[str, List[DutyAssignment]],
|
||||
) -> Dict[str, Any]:
|
||||
slot_payload = [
|
||||
_slot_dict(s, assignments_by_slot.get(s.id, []))
|
||||
for s in sorted(slots, key=lambda x: (x.order_num or 0, x.shift_type or ""))
|
||||
if s.del_flag == 0
|
||||
]
|
||||
return {
|
||||
"id": day.id,
|
||||
"dutyDate": day.duty_date.isoformat() if day.duty_date else None,
|
||||
"linkStatus": day.link_status,
|
||||
"accessToken": day.access_token,
|
||||
"publicUrl": build_public_url(day.access_token),
|
||||
"slots": slot_payload,
|
||||
}
|
||||
|
||||
|
||||
def _load_day_by_date(db: Session, tenant_no: str, duty_date: date) -> Optional[DutyDay]:
|
||||
return (
|
||||
db.query(DutyDay)
|
||||
.filter(
|
||||
DutyDay.tenant_no == tenant_no,
|
||||
DutyDay.duty_date == duty_date,
|
||||
DutyDay.del_flag == 0,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
|
||||
def _load_day_by_id(db: Session, tenant_no: str, day_id: str) -> Optional[DutyDay]:
|
||||
return (
|
||||
db.query(DutyDay)
|
||||
.filter(DutyDay.id == day_id, DutyDay.tenant_no == tenant_no, DutyDay.del_flag == 0)
|
||||
.first()
|
||||
)
|
||||
|
||||
|
||||
def _load_slots(db: Session, day_id: str) -> List[DutySlot]:
|
||||
return (
|
||||
db.query(DutySlot)
|
||||
.filter(DutySlot.duty_day_id == day_id, DutySlot.del_flag == 0)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
def _load_assignments_for_slots(
|
||||
db: Session, slot_ids: Sequence[str]
|
||||
) -> Dict[str, List[DutyAssignment]]:
|
||||
if not slot_ids:
|
||||
return {}
|
||||
rows = (
|
||||
db.query(DutyAssignment)
|
||||
.filter(DutyAssignment.duty_slot_id.in_(list(slot_ids)), DutyAssignment.del_flag == 0)
|
||||
.all()
|
||||
)
|
||||
out: Dict[str, List[DutyAssignment]] = {sid: [] for sid in slot_ids}
|
||||
for row in rows:
|
||||
out.setdefault(row.duty_slot_id, []).append(row)
|
||||
return out
|
||||
|
||||
|
||||
def _serialize_day(db: Session, day: DutyDay) -> Dict[str, Any]:
|
||||
slots = _load_slots(db, day.id)
|
||||
assigns = _load_assignments_for_slots(db, [s.id for s in slots])
|
||||
return _day_dict(day, slots, assigns)
|
||||
|
||||
|
||||
def empty_day_skeleton(duty_date: date) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": None,
|
||||
"dutyDate": duty_date.isoformat(),
|
||||
"linkStatus": LINK_OPEN,
|
||||
"accessToken": None,
|
||||
"publicUrl": None,
|
||||
"slots": [],
|
||||
}
|
||||
|
||||
|
||||
def get_month_summary(db: Session, tenant_no: str, year: int, month: int) -> List[Dict[str, Any]]:
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
start = date(year, month, 1)
|
||||
end = date(year, month, last_day)
|
||||
days = (
|
||||
db.query(DutyDay)
|
||||
.filter(
|
||||
DutyDay.tenant_no == tenant_no,
|
||||
DutyDay.del_flag == 0,
|
||||
DutyDay.duty_date >= start,
|
||||
DutyDay.duty_date <= end,
|
||||
)
|
||||
.order_by(DutyDay.duty_date.asc())
|
||||
.all()
|
||||
)
|
||||
result = []
|
||||
for day in days:
|
||||
payload = _serialize_day(db, day)
|
||||
# 月历摘要可精简 assignments 人名
|
||||
result.append(payload)
|
||||
return result
|
||||
|
||||
|
||||
def get_day_detail(db: Session, tenant_no: str, duty_date: date) -> Dict[str, Any]:
|
||||
day = _load_day_by_date(db, tenant_no, duty_date)
|
||||
if not day:
|
||||
return empty_day_skeleton(duty_date)
|
||||
return _serialize_day(db, day)
|
||||
|
||||
|
||||
def soft_delete_assignment(
|
||||
db: Session, tenant_no: str, assignment_id: str, operator_id: str = ""
|
||||
) -> None:
|
||||
row = (
|
||||
db.query(DutyAssignment)
|
||||
.filter(
|
||||
DutyAssignment.id == assignment_id,
|
||||
DutyAssignment.tenant_no == tenant_no,
|
||||
DutyAssignment.del_flag == 0,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not row:
|
||||
raise DutyBizError("报名记录不存在")
|
||||
row.del_flag = 1
|
||||
row.update_by = operator_id or None
|
||||
row.update_time = datetime.now()
|
||||
db.commit()
|
||||
|
||||
|
||||
def set_link_status(
|
||||
db: Session, tenant_no: str, day_id: str, status: str, operator_id: str = ""
|
||||
) -> Dict[str, Any]:
|
||||
if status not in (LINK_OPEN, LINK_CLOSED):
|
||||
raise DutyBizError("链接状态无效")
|
||||
day = _load_day_by_id(db, tenant_no, day_id)
|
||||
if not day:
|
||||
raise DutyBizError("值班日不存在")
|
||||
day.link_status = status
|
||||
day.update_by = operator_id or None
|
||||
day.update_time = datetime.now()
|
||||
db.commit()
|
||||
db.refresh(day)
|
||||
return _serialize_day(db, day)
|
||||
|
||||
|
||||
def _count_active_assignments(db: Session, slot_id: str) -> int:
|
||||
return (
|
||||
db.query(DutyAssignment)
|
||||
.filter(DutyAssignment.duty_slot_id == slot_id, DutyAssignment.del_flag == 0)
|
||||
.count()
|
||||
)
|
||||
|
||||
|
||||
def save_day(
|
||||
db: Session,
|
||||
tenant_no: str,
|
||||
user: Dict[str, Any],
|
||||
body: DutyDaySaveBody,
|
||||
) -> Dict[str, Any]:
|
||||
user_id = (user.get("user_id") or user.get("id") or "") or ""
|
||||
dept_id = (user.get("region_id") or user.get("create_dept_id") or "") or ""
|
||||
now = datetime.now()
|
||||
|
||||
# 校验：每个 slot 提交的指定人数不超过 capacity
|
||||
seen_types = set()
|
||||
for slot_in in body.slots:
|
||||
if slot_in.shiftType in seen_types:
|
||||
raise DutyBizError("同一天不能重复配置相同班次")
|
||||
seen_types.add(slot_in.shiftType)
|
||||
assign_count = len(slot_in.assignments or [])
|
||||
if assign_count > slot_in.capacity:
|
||||
label = SHIFT_LABELS.get(slot_in.shiftType, slot_in.shiftType)
|
||||
raise DutyBizError(f"{label}报名人数不能超过名额")
|
||||
|
||||
day = _load_day_by_date(db, tenant_no, body.dutyDate)
|
||||
if not day:
|
||||
day = DutyDay(
|
||||
id=generate_id(),
|
||||
duty_date=body.dutyDate,
|
||||
access_token=new_access_token(),
|
||||
link_status=body.linkStatus or LINK_OPEN,
|
||||
tenant_no=tenant_no,
|
||||
create_by=user_id,
|
||||
create_dept_id=dept_id,
|
||||
create_time=now,
|
||||
update_by=user_id,
|
||||
update_time=now,
|
||||
del_flag=0,
|
||||
)
|
||||
db.add(day)
|
||||
db.flush()
|
||||
else:
|
||||
if body.linkStatus in (LINK_OPEN, LINK_CLOSED):
|
||||
day.link_status = body.linkStatus
|
||||
day.update_by = user_id
|
||||
day.update_time = now
|
||||
|
||||
existing_slots = _load_slots(db, day.id)
|
||||
existing_by_type = {s.shift_type: s for s in existing_slots}
|
||||
incoming_types = {s.shiftType for s in body.slots}
|
||||
|
||||
# 停用未再提交且已有报名的班次：禁止
|
||||
for shift_type, slot in existing_by_type.items():
|
||||
if shift_type not in incoming_types:
|
||||
taken = _count_active_assignments(db, slot.id)
|
||||
if taken > 0:
|
||||
label = SHIFT_LABELS.get(shift_type, shift_type)
|
||||
raise DutyBizError(f"{label}仍有报名，请先清空后再移除该班次")
|
||||
slot.del_flag = 1
|
||||
slot.update_by = user_id
|
||||
slot.update_time = now
|
||||
|
||||
order = 0
|
||||
for slot_in in body.slots:
|
||||
order += 1
|
||||
slot = existing_by_type.get(slot_in.shiftType)
|
||||
if slot is None:
|
||||
slot = DutySlot(
|
||||
id=generate_id(),
|
||||
duty_day_id=day.id,
|
||||
shift_type=slot_in.shiftType,
|
||||
capacity=slot_in.capacity,
|
||||
tenant_no=tenant_no,
|
||||
order_num=order,
|
||||
create_by=user_id,
|
||||
create_dept_id=dept_id,
|
||||
create_time=now,
|
||||
update_by=user_id,
|
||||
update_time=now,
|
||||
del_flag=0,
|
||||
)
|
||||
db.add(slot)
|
||||
db.flush()
|
||||
else:
|
||||
taken = _count_active_assignments(db, slot.id)
|
||||
# 缩容：不能小于「保留后的报名」——保存时以 body.assignments 为准重建指定项，
|
||||
# 但抢班记录不在 body 里时需计入 taken 中不在本次提交删除范围的 grab。
|
||||
# 简化策略：先统计将保留的 grab + 新 assign 总数，再与 capacity 比。
|
||||
existing_assigns = (
|
||||
db.query(DutyAssignment)
|
||||
.filter(DutyAssignment.duty_slot_id == slot.id, DutyAssignment.del_flag == 0)
|
||||
.all()
|
||||
)
|
||||
# body.assignments 视为该班次最终「指定」名单；grab 全部保留，除非超容时拒绝缩容
|
||||
grab_rows = [a for a in existing_assigns if a.source == SOURCE_GRAB]
|
||||
new_assign_count = len(slot_in.assignments or [])
|
||||
final_count = len(grab_rows) + new_assign_count
|
||||
if final_count > slot_in.capacity:
|
||||
label = SHIFT_LABELS.get(slot_in.shiftType, slot_in.shiftType)
|
||||
raise DutyBizError(
|
||||
f"{label}当前已有 {len(grab_rows)} 人抢班 + {new_assign_count} 人指定，"
|
||||
f"超过名额 {slot_in.capacity}，请先删人再缩容"
|
||||
)
|
||||
slot.capacity = slot_in.capacity
|
||||
slot.order_num = order
|
||||
slot.update_by = user_id
|
||||
slot.update_time = now
|
||||
|
||||
# 软删旧的 assign 指定，再写入新的；grab 保留
|
||||
for a in existing_assigns:
|
||||
if a.source == SOURCE_ASSIGN:
|
||||
a.del_flag = 1
|
||||
a.update_by = user_id
|
||||
a.update_time = now
|
||||
|
||||
# 新 slot：写入 assignments；已存在 slot：写入新的 assign
|
||||
for item in slot_in.assignments or []:
|
||||
person_name = (item.personName or "").strip()
|
||||
uid = (item.userId or "").strip() or None
|
||||
if item.source == SOURCE_ASSIGN and not person_name and not uid:
|
||||
continue
|
||||
if not person_name:
|
||||
raise DutyBizError("指定员工须有姓名")
|
||||
person_name = normalize_person_name(person_name) if person_name else person_name
|
||||
# 若只有 userId，允许 personName 已由上层填好
|
||||
if not person_name:
|
||||
raise DutyBizError("指定员工姓名不能为空")
|
||||
db.add(
|
||||
DutyAssignment(
|
||||
id=generate_id(),
|
||||
duty_slot_id=slot.id,
|
||||
source=SOURCE_ASSIGN if item.source != SOURCE_GRAB else SOURCE_GRAB,
|
||||
user_id=uid,
|
||||
person_name=person_name,
|
||||
tenant_no=tenant_no,
|
||||
create_by=user_id,
|
||||
create_dept_id=dept_id,
|
||||
create_time=now,
|
||||
update_by=user_id,
|
||||
update_time=now,
|
||||
del_flag=0,
|
||||
)
|
||||
)
|
||||
|
||||
# 新建 slot 时也要校验人数
|
||||
if slot.id and _count_active_assignments(db, slot.id) > slot.capacity:
|
||||
label = SHIFT_LABELS.get(slot.shift_type, slot.shift_type)
|
||||
raise DutyBizError(f"{label}报名人数不能超过名额")
|
||||
|
||||
db.commit()
|
||||
db.refresh(day)
|
||||
return _serialize_day(db, day)
|
||||
|
||||
|
||||
def get_public_page(db: Session, token: str) -> Dict[str, Any]:
|
||||
token = (token or "").strip()
|
||||
if not token:
|
||||
raise DutyBizError("链接无效或已失效")
|
||||
day = (
|
||||
db.query(DutyDay)
|
||||
.filter(DutyDay.access_token == token, DutyDay.del_flag == 0)
|
||||
.first()
|
||||
)
|
||||
if not day:
|
||||
raise DutyBizError("链接无效或已失效")
|
||||
payload = _serialize_day(db, day)
|
||||
# 公开页不强制返回 accessToken 也可，保留便于调试；可去掉敏感则仅留必要字段
|
||||
return {
|
||||
"dutyDate": payload["dutyDate"],
|
||||
"linkStatus": payload["linkStatus"],
|
||||
"slots": payload["slots"],
|
||||
"closed": day.link_status == LINK_CLOSED,
|
||||
}
|
||||
|
||||
|
||||
def grab(db: Session, token: str, shift_type: str, person_name: str) -> Dict[str, Any]:
|
||||
name = normalize_person_name(person_name)
|
||||
shift_type = (shift_type or "").strip()
|
||||
if shift_type not in SHIFT_LABELS:
|
||||
raise DutyBizError("班次类型无效")
|
||||
|
||||
day = (
|
||||
db.query(DutyDay)
|
||||
.filter(DutyDay.access_token == token, DutyDay.del_flag == 0)
|
||||
.with_for_update()
|
||||
.first()
|
||||
)
|
||||
if not day:
|
||||
raise DutyBizError("链接无效或已失效")
|
||||
if day.link_status == LINK_CLOSED:
|
||||
raise DutyBizError("报名已关闭")
|
||||
|
||||
slot = (
|
||||
db.query(DutySlot)
|
||||
.filter(
|
||||
DutySlot.duty_day_id == day.id,
|
||||
DutySlot.shift_type == shift_type,
|
||||
DutySlot.del_flag == 0,
|
||||
)
|
||||
.with_for_update()
|
||||
.first()
|
||||
)
|
||||
if not slot:
|
||||
raise DutyBizError("该班次不存在")
|
||||
|
||||
taken = _count_active_assignments(db, slot.id)
|
||||
if taken >= slot.capacity:
|
||||
raise DutyBizError("名额已满")
|
||||
|
||||
now = datetime.now()
|
||||
row = DutyAssignment(
|
||||
id=generate_id(),
|
||||
duty_slot_id=slot.id,
|
||||
source=SOURCE_GRAB,
|
||||
user_id=None,
|
||||
person_name=name,
|
||||
tenant_no=day.tenant_no,
|
||||
create_by="",
|
||||
create_dept_id="",
|
||||
create_time=now,
|
||||
update_by="",
|
||||
update_time=now,
|
||||
del_flag=0,
|
||||
)
|
||||
db.add(row)
|
||||
db.commit()
|
||||
return {
|
||||
"dutyDate": day.duty_date.isoformat(),
|
||||
"shiftType": shift_type,
|
||||
"shiftLabel": SHIFT_LABELS.get(shift_type, shift_type),
|
||||
"personName": name,
|
||||
"status": "ok",
|
||||
}
|
||||
|
||||
|
||||
# re-export for tests
|
||||
__all__ = [
|
||||
"SHIFT_LABELS",
|
||||
"DutyBizError",
|
||||
"normalize_person_name",
|
||||
"new_access_token",
|
||||
"build_public_url",
|
||||
"get_month_summary",
|
||||
"get_day_detail",
|
||||
"empty_day_skeleton",
|
||||
"save_day",
|
||||
"set_link_status",
|
||||
"soft_delete_assignment",
|
||||
"get_public_page",
|
||||
"grab",
|
||||
]
|
||||
@@ -0,0 +1,172 @@
|
||||
"""值班排班服务单测"""
|
||||
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,
|
||||
)
|
||||
|
||||
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
|
||||
|
||||
def query_side_effect(model):
|
||||
name = getattr(model, "__name__", str(model))
|
||||
if name == "DutyDay":
|
||||
return day_q
|
||||
if name == "DutySlot":
|
||||
return slot_q
|
||||
# DutyAssignment count
|
||||
aq = MagicMock()
|
||||
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, "李四")
|
||||
|
||||
|
||||
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, "李四")
|
||||
|
||||
|
||||
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)
|
||||
Reference in new issue
Block a user