shanxia2.0/backend/tests/test_pl_personnel_missing_identity.py

"""SQLite-only vehicle personnel identity and partial-field regressions."""
from datetime import datetime

import pymysql
import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session

from app.models.pl_models import PlPersonnel, PlVehiclePersonnel
from app.schemas.pl.base_data import VehiclePersonnelRow
from app.services.pl_personnel_service import save_vehicle_personnel, upsert_personnel


@pytest.fixture
def db(monkeypatch):
    def forbidden(*a, **k):
        raise AssertionError('Business database access forbidden')
    monkeypatch.setattr(pymysql, 'connect', forbidden)
    monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
    engine = create_engine('sqlite://')
    PlPersonnel.__table__.create(engine)
    PlVehiclePersonnel.__table__.create(engine)
    with Session(engine, autoflush=False, expire_on_commit=False) as session:
        yield session
    engine.dispose()


def person(db, identity='p', name='Person', phone='100', **extra):
    row = PlPersonnel(id=identity, tenant_no='t', personnel_name=name, phone_no=phone,
        create_by='creator', create_time=datetime(2020,1,1), **extra)
    db.add(row)
    return row


def link(db, identity='link', pid='p', role='driver', **extra):
    row = PlVehiclePersonnel(id=identity, tenant_no='t', vehicle_id='v', personnel_id=pid,
        role=role, create_by='creator', create_time=datetime(2020,1,1), **extra)
    db.add(row)
    return row


def body(**values):
    return VehiclePersonnelRow.model_validate({'personnelName':'Person', 'role':'driver', **values})


def save(db, values):
    save_vehicle_personnel(db,'v',values,'t',user_id='editor',dept_id='dept')
    db.flush()


@pytest.mark.parametrize('key', ['name','phone'])
@pytest.mark.parametrize('standalone', [False,True])
def test_ambiguous_legacy_identity_rejects_instead_of_replacing_link(db,key,standalone):
    person(db,'p1',phone=None if key=='name' else '100')
    person(db,'p2',phone=None if key=='name' else '100')
    old=link(db,pid='p2');db.commit()
    payload=body(**({'phoneNo':'100'} if key=='phone' else {}))
    with pytest.raises(HTTPException):
        if standalone:upsert_personnel(db,'t',payload,user_id='u',dept_id='d')
        else:save(db,[payload])
    assert not db.dirty and not db.new
    assert old.del_flag==0 and db.query(PlVehiclePersonnel).count()==1


def test_unique_no_id_matches_existing_person_and_preserves_relation_identity(db):
    p=person(db);old=link(db);db.commit()
    save(db,[body(phoneNo='100')]);db.commit();save(db,[body(phoneNo='100')])
    assert db.query(PlPersonnel).count()==db.query(PlVehiclePersonnel).count()==1
    assert old.del_flag==0 and old.personnel_id==p.id
    assert old.create_by==p.create_by=='creator'
    assert old.create_time==p.create_time==datetime(2020,1,1)


def test_omitted_fields_preserve_person_and_link_fields(db):
    p=person(db,age=42);old=link(db,remark='keep',is_default=1);db.commit()
    save(db,[body(personnelId='p')])
    assert p.phone_no=='100' and p.age==42
    assert old.remark=='keep' and old.is_default==1


def test_explicit_null_and_zero_clear_fields(db):
    p=person(db,age=42);old=link(db,remark='keep',is_default=1);db.commit()
    save(db,[body(personnelId='p',phoneNo=None,age=None,remark=None,isDefault=0)])
    assert p.phone_no is None and p.age is None
    assert old.remark is None and old.is_default==0


def test_new_person_can_have_two_roles_without_duplicate_person_creation(db):
    save(db,[body(phoneNo='100'),body(phoneNo='100',role='escort')]);db.commit()
    people=db.query(PlPersonnel).all();links=db.query(PlVehiclePersonnel).all()
    assert len(people)==1 and len(links)==2
    assert {row.personnel_id for row in links}=={people[0].id}
    assert {row.role for row in links}=={'driver','escort'}


@pytest.mark.parametrize('reverse',[False,True])
def test_explicit_identity_reserved_and_other_role_resolves_same_person(db,reverse):
    person(db);old=link(db);db.commit()
    rows=[body(personnelId='p'),body(phoneNo='100',role='escort')]
    save(db,rows[::-1] if reverse else rows)
    assert old.del_flag==0 and db.query(PlPersonnel).count()==1
    assert {row.personnel_id for row in db.query(PlVehiclePersonnel)}=={'p'}


def test_explicit_reservation_does_not_guess_remaining_duplicate_name(db):
    person(db,'p1',phone=None);person(db,'p2',phone=None);link(db,pid='p1');db.commit()
    with pytest.raises(HTTPException):save(db,[body(personnelId='p1'),body()])
    assert not db.dirty and not db.new


@pytest.mark.parametrize('field,values',[('phoneNo',('100','200')),('age',(20,30)),
                                        ('personnelName',('Person','Different'))])
def test_same_person_conflicting_fields_across_roles_reject_before_mutation(db,field,values):
    p=person(db,age=42);link(db);db.commit()
    a=body(personnelId='p',**{field:values[0]})
    b=body(personnelId='p',role='escort',**{field:values[1]})
    with pytest.raises(HTTPException):save(db,[a,b])
    assert not db.dirty and not db.new and p.age==42 and p.phone_no=='100'


@pytest.mark.parametrize('bad',['foreign','deleted','missing','duplicate'])
def test_entire_batch_rejects_invalid_explicit_id_before_editing_first_person(db,bad):
    p=person(db);link(db)
    if bad in ('foreign','deleted'):
        other=person(db,'other',name='Other');other.tenant_no='other' if bad=='foreign' else 't'
        other.del_flag=1 if bad=='deleted' else 0
    db.commit()
    rows=[body(personnelId='p',phoneNo='new')]
    rows.append(body(personnelId='p' if bad=='duplicate' else 'other'))
    with pytest.raises(HTTPException):save(db,rows)
    assert not db.dirty and not db.new and p.phone_no=='100'


def test_effective_defaults_cannot_create_two_default_drivers(db):
    person(db);person(db,'second',name='Second');old=link(db,is_default=1);db.commit()
    with pytest.raises(HTTPException):save(db,[body(personnelId='p'),
        body(personnelId='second',personnelName='Second',isDefault=1)])
    assert old.is_default==1 and not db.dirty and not db.new


def test_add_edit_remove_and_omitted_collection_are_differential(db):
    p=person(db);old=link(db);person(db,'removed',name='Removed');removed=link(db,'removed-link','removed')
    db.commit();save_vehicle_personnel(db,'v',None,'t',user_id='u',dept_id='d')
    assert old.del_flag==removed.del_flag==0
    save(db,[body(personnelId='p',phoneNo='new'),body(personnelName='New',phoneNo='300',role='escort')])
    assert old.del_flag==0 and old.personnel_id==p.id and p.phone_no=='new'
    assert removed.del_flag==1 and db.query(PlVehiclePersonnel).filter_by(del_flag=0).count()==2
    save(db,[])
    assert db.query(PlVehiclePersonnel).filter_by(del_flag=0).count()==0
    assert db.query(PlPersonnel).count()==3


def test_cross_role_partial_person_update_merges_without_clearing_phone(db):
    p=person(db,age=42);link(db);db.commit()
    save(db,[body(personnelId='p',phoneNo='200'),body(personnelId='p',role='escort')])
    assert p.phone_no=='200' and p.age==42


def test_standalone_upsert_preserves_omitted_optional_fields(db):
    p=person(db,age=42);db.commit()
    identity=upsert_personnel(db,'t',body(personnelId='p'),user_id='u',dept_id='d')
    assert identity==p.id and p.phone_no=='100' and p.age==42


def test_explicit_current_client_id_is_safe_with_duplicate_names(db):
    person(db,'p1',phone=None);p=person(db,'p2',phone=None);old=link(db,pid='p2');db.commit()
    save(db,[body(personnelId='p2',remark='')])
    assert old.del_flag==0 and old.personnel_id==p.id and old.remark==''
    assert db.query(PlVehiclePersonnel).count()==1


def test_same_name_in_other_tenant_does_not_participate_in_matching(db):
    p=person(db);old=link(db);other=person(db,'other');other.tenant_no='other';db.commit()
    save(db,[body()])
    assert old.personnel_id==p.id and old.del_flag==0 and p.phone_no=='100'


def test_empty_collection_only_removes_current_vehicle_links(db):
    person(db);old=link(db);other=link(db,'other-link');other.vehicle_id='other-vehicle';db.commit()
    save(db,[])
    assert old.del_flag==1 and other.del_flag==0


def test_duplicate_historical_links_reject_before_person_update(db):
    p=person(db);link(db);link(db,'duplicate-link');db.commit()
    with pytest.raises(HTTPException):save(db,[body(personnelId='p',phoneNo='200')])
    assert p.phone_no=='100' and not db.dirty and not db.new


@pytest.mark.parametrize('reverse',[False,True])
def test_explicit_and_missing_id_same_role_reject_regardless_of_order(db,reverse):
    person(db);link(db);db.commit()
    rows=[body(),body(personnelId='p')]
    with pytest.raises(HTTPException):save(db,rows[::-1] if reverse else rows)
    assert not db.dirty and not db.new


def test_explicit_new_name_can_be_reused_by_other_role_in_same_batch(db):
    p=person(db);link(db);db.commit()
    save(db,[body(personnelName='Renamed',role='escort'),body(personnelId='p',personnelName='Renamed')])
    assert p.personnel_name=='Renamed' and db.query(PlPersonnel).count()==1
    assert {row.personnel_id for row in db.query(PlVehiclePersonnel)}=={'p'}