shanxia2.0/backend/tests/test_duty_missing_identity_patch.py

"""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