73 lines
3.1 KiB
Python
73 lines
3.1 KiB
Python
"""排班差量保存的数据库回归（独立 SQLite，无业务库连接）。"""
|
|
from datetime import date, datetime
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.duty_models import DutyAssignment, DutyDay, DutySlot
|
|
from app.schemas.duty import DutyDaySaveBody
|
|
from app.services.duty_schedule import save_day, DutyBizError
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
engine = create_engine('sqlite:///:memory:')
|
|
for model in (DutyDay, DutySlot, DutyAssignment):
|
|
model.__table__.create(engine)
|
|
with Session(engine, autoflush=False) as session:
|
|
session.add(DutyDay(id='day', duty_date=date(2026, 10, 27), access_token='token', tenant_no='t'))
|
|
session.add(DutySlot(id='slot', duty_day_id='day', shift_type='morning', capacity=5, tenant_no='t'))
|
|
session.add(DutyAssignment(id='old', duty_slot_id='slot', source='assign', user_id='u', person_name='张三', tenant_no='t', create_time=datetime(2020, 1, 1)))
|
|
session.add(DutyAssignment(id='grab', duty_slot_id='slot', source='grab', user_id='g', person_name='李四', tenant_no='t'))
|
|
session.commit()
|
|
with patch('app.services.duty_schedule._serialize_day', return_value={}), patch('app.services.duty_schedule.assert_user_is_duty_staff'):
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def save(db, **fields):
|
|
body = DutyDaySaveBody.model_validate({'dutyDate': '2026-10-27', 'slots': [{'shiftType':'morning', 'capacity':5, **fields}]})
|
|
return save_day(db, 't', {'id':'actor'}, body)
|
|
|
|
|
|
def test_assignment_edit_preserves_id_creation_and_grab(db):
|
|
save(db, assignments=[{'id':'old', 'userId':'u', 'personName':'张三新'}])
|
|
old = db.get(DutyAssignment, 'old')
|
|
assert old.del_flag == 0
|
|
assert old.person_name == '张三新'
|
|
assert old.create_time == datetime(2020, 1, 1)
|
|
assert db.get(DutyAssignment, 'grab').del_flag == 0
|
|
assert db.query(DutyAssignment).count() == 2
|
|
|
|
|
|
def test_legacy_payload_uniquely_matches_user_and_omission_preserves(db):
|
|
save(db, assignments=[{'userId':'u', 'personName':'张三'}])
|
|
assert db.get(DutyAssignment, 'old').del_flag == 0
|
|
save(db)
|
|
assert db.get(DutyAssignment, 'old').del_flag == 0
|
|
save(db, assignments=[])
|
|
assert db.get(DutyAssignment, 'old').del_flag == 1
|
|
assert db.get(DutyAssignment, 'grab').del_flag == 0
|
|
|
|
|
|
@pytest.mark.parametrize('items', [
|
|
[{'id':'foreign', 'userId':'u','personName':'张三'}],
|
|
[{'id':'grab', 'userId':'u','personName':'张三'}],
|
|
[{'id':'old', 'userId':'u','personName':'张三'}, {'id':'old','userId':'u','personName':'张三'}],
|
|
[{'source':'grab','userId':'x','personName':'张三'}],
|
|
])
|
|
def test_invalid_assignment_identity_is_rejected(db, items):
|
|
with pytest.raises(DutyBizError):
|
|
save(db, assignments=items)
|
|
db.rollback()
|
|
assert db.get(DutyAssignment, 'old').del_flag == 0
|
|
assert db.query(DutyAssignment).count() == 2
|
|
|
|
|
|
def test_removal_and_capacity_shrink_counts_persisted_difference(db):
|
|
save(db, assignments=[], capacity=1)
|
|
assert db.get(DutyAssignment, 'old').del_flag == 1
|
|
assert db.get(DutySlot, 'slot').capacity == 1
|