shanxia2.0/backend/tests/test_duty_child_diff.py

"""排班差量保存的数据库回归（独立 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