204 lines
8.9 KiB
Python
204 lines
8.9 KiB
Python
"""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'}
|