272 lines
7.4 KiB
Python
272 lines
7.4 KiB
Python
"""值班排班服务单测"""
|
|
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,
|
|
)
|
|
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
|
|
|
|
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
|
|
# 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
|
|
|
|
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
|
|
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)
|