76 lines
3.3 KiB
Python
76 lines
3.3 KiB
Python
"""Current duty save path: SQLite-only missing-identity regressions."""
|
|
from datetime import date, datetime
|
|
from unittest.mock import patch
|
|
|
|
import pymysql
|
|
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 DutyBizError, save_day
|
|
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError('Business database connections forbidden')
|
|
monkeypatch.setattr(pymysql, 'connect', forbidden)
|
|
monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
|
|
engine = create_engine('sqlite://')
|
|
for model in (DutyDay, DutySlot, DutyAssignment):
|
|
model.__table__.create(engine)
|
|
with Session(engine, autoflush=False, expire_on_commit=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='Existing', tenant_no='t', create_by='creator', create_time=datetime(2020, 1, 1)))
|
|
session.add(DutyAssignment(id='grab', duty_slot_id='slot', source='grab', user_id='g',
|
|
person_name='Grab', 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, assignments):
|
|
body = DutyDaySaveBody.model_validate({'dutyDate': '2026-10-27', 'slots': [
|
|
{'shiftType': 'morning', 'capacity': 5, 'assignments': assignments}]})
|
|
return save_day(db, 't', {'id': 'actor'}, body)
|
|
|
|
|
|
def test_name_only_save_preserves_bound_user_assignment_identity(db):
|
|
for _ in range(2):
|
|
save(db, [{'personName': 'Existing'}])
|
|
row = db.get(DutyAssignment, 'old')
|
|
assert row.del_flag == 0 and row.user_id == 'u'
|
|
assert row.create_by == 'creator' and row.create_time == datetime(2020, 1, 1)
|
|
assert db.query(DutyAssignment).count() == 2
|
|
assert db.get(DutyAssignment, 'grab').del_flag == 0
|
|
|
|
|
|
def test_name_only_ambiguous_bound_users_are_rejected_without_replacement(db):
|
|
db.add(DutyAssignment(id='same-name', duty_slot_id='slot', source='assign', user_id='another',
|
|
person_name='Existing', tenant_no='t'))
|
|
db.commit()
|
|
with pytest.raises(DutyBizError):
|
|
save(db, [{'personName': 'Existing'}])
|
|
assert db.get(DutyAssignment, 'old').del_flag == 0
|
|
assert db.get(DutyAssignment, 'same-name').del_flag == 0
|
|
assert not db.new
|
|
|
|
|
|
def test_explicit_user_id_without_child_id_keeps_current_identity(db):
|
|
save(db, [{'userId': 'u', 'personName': 'Existing'}])
|
|
assert db.get(DutyAssignment, 'old').del_flag == 0
|
|
assert db.query(DutyAssignment).count() == 2
|
|
|
|
|
|
def test_invalid_later_assignment_does_not_replace_earlier_assignment(db):
|
|
with pytest.raises(DutyBizError):
|
|
save(db, [{'personName': 'Existing'}, {'id': 'missing', 'personName': 'Missing'}])
|
|
assert db.get(DutyAssignment, 'old').del_flag == 0
|
|
assert not db.new
|