diff --git a/backend/app/services/pl_vehicle_mine_vehicles.py b/backend/app/services/pl_vehicle_mine_vehicles.py new file mode 100644 index 000000000..c638e6408 --- /dev/null +++ b/backend/app/services/pl_vehicle_mine_vehicles.py @@ -0,0 +1,40 @@ +"""当前用户关联驾押人员所挂靠车辆的车牌集合。""" +from __future__ import annotations + +from typing import Optional, Set + +from sqlalchemy.orm import Session + +from app.models.pl_models import PlVehicle, PlVehiclePersonnel +from app.services.pl_driver_plan_service import resolve_personnel_for_user + + +def license_plates_for_user_personnel( + db: Session, tenant_no: str, user_id: str +) -> Set[str]: + personnel = resolve_personnel_for_user(db, tenant_no=tenant_no, user_id=user_id) + if not personnel: + return set() + rows = ( + db.query(PlVehicle.license_plate) + .join( + PlVehiclePersonnel, + PlVehiclePersonnel.vehicle_id == PlVehicle.id, + ) + .filter( + PlVehiclePersonnel.personnel_id == personnel.id, + PlVehiclePersonnel.tenant_no == tenant_no, + PlVehiclePersonnel.del_flag == 0, + PlVehicle.tenant_no == tenant_no, + PlVehicle.del_flag == 0, + ) + .all() + ) + return {str(r[0]).strip() for r in rows if r and r[0] and str(r[0]).strip()} + + +def resolve_personnel_id_for_user( + db: Session, tenant_no: str, user_id: str +) -> Optional[str]: + p = resolve_personnel_for_user(db, tenant_no=tenant_no, user_id=user_id) + return p.id if p else None diff --git a/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py b/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py new file mode 100644 index 000000000..f0222388a --- /dev/null +++ b/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py @@ -0,0 +1,98 @@ +"""mineVehicles:车辆人员归属过滤与写校验。""" +from __future__ import annotations + +from datetime import date, datetime +from types import SimpleNamespace + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.models.pl_models import ( + PlPersonnel, + PlVehicle, + PlVehiclePersonnel, + PlVehicleRechargeApply, +) + + +def _make_session(): + engine = create_engine("sqlite://") + for table in ( + PlVehicle.__table__, + PlPersonnel.__table__, + PlVehiclePersonnel.__table__, + PlVehicleRechargeApply.__table__, + ): + table.create(engine) + return sessionmaker(bind=engine)() + + +def _user(uid: str = "u-driver") -> SimpleNamespace: + return SimpleNamespace(id=uid, name="司机甲", dept_id="d1", role="driver", tenant_no="t1") + + +def _seed_person_vehicle(db, *, user_id="u-driver", plate="鄂A11111", vehicle_id="veh-a"): + db.add( + PlVehicle( + id=vehicle_id, + vehicle_type="self", + license_plate=plate, + tenant_no="t1", + create_dept_id="d1", + del_flag=0, + ) + ) + db.add( + PlPersonnel( + id="p-1", + personnel_name="司机甲", + personnel_type="driver", + personnel_affiliation="self", + status="1", + sys_user_id=user_id, + tenant_no="t1", + create_dept_id="d1", + del_flag=0, + ) + ) + db.add( + PlVehiclePersonnel( + id="vp-1", + vehicle_id=vehicle_id, + personnel_id="p-1", + role="driver", + is_default=1, + tenant_no="t1", + create_dept_id="d1", + del_flag=0, + ) + ) + + +def test_license_plates_for_user_personnel(): + from app.services.pl_vehicle_mine_vehicles import license_plates_for_user_personnel + + db = _make_session() + _seed_person_vehicle(db) + db.add( + PlVehicle( + id="veh-b", + vehicle_type="self", + license_plate="鄂A22222", + tenant_no="t1", + create_dept_id="d1", + del_flag=0, + ) + ) + db.commit() + plates = license_plates_for_user_personnel(db, "t1", "u-driver") + assert plates == {"鄂A11111"} + + +def test_license_plates_empty_when_no_personnel(): + from app.services.pl_vehicle_mine_vehicles import license_plates_for_user_personnel + + db = _make_session() + db.commit() + assert license_plates_for_user_personnel(db, "t1", "u-nobody") == set()