From 0d2c5a20483b9774908cfd23b7b8164886c80fcc Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Sun, 27 Sep 2026 18:33:25 +0800 Subject: [PATCH] feat(pl): enforce mineVehicles write rules on recharge apply Reject foreign plates on create and require own applicant for update/submit/cancel when mineVehicles is enabled. Co-authored-by: Cursor --- .../services/pl_vehicle_recharge_service.py | 65 +++++++++++++++++-- .../test_pl_vehicle_recharge_mine_vehicles.py | 43 ++++++++++++ 2 files changed, 101 insertions(+), 7 deletions(-) diff --git a/backend/app/services/pl_vehicle_recharge_service.py b/backend/app/services/pl_vehicle_recharge_service.py index a3e7246a7..797f09784 100644 --- a/backend/app/services/pl_vehicle_recharge_service.py +++ b/backend/app/services/pl_vehicle_recharge_service.py @@ -564,15 +564,35 @@ def get_apply( return apply_to_read(row, db, include_detail=True) +def _assert_mine_vehicle_plate( + db: Session, tenant_no: str, user: DbUserContext, plate: str +) -> None: + plates = license_plates_for_user_personnel(db, tenant_no, user.id) + if (plate or "").strip() not in plates: + raise HTTPException(status_code=403, detail="只能选择本人车辆") + + +def _assert_own_apply_for_write(row: PlVehicleRechargeApply, user: DbUserContext) -> None: + if (row.applicant_id or "").strip() != (user.id or "").strip(): + raise HTTPException(status_code=403, detail="只能操作本人申请") + + def create_apply( db: Session, tenant_no: str, user: DbUserContext, payload: VehicleRechargeApplyCreate, + *, + mine_vehicles: bool = False, ) -> VehicleRechargeApplyRead: license_plate = payload.license_plate.strip() + if mine_vehicles: + _assert_mine_vehicle_plate(db, tenant_no, user, license_plate) _assert_vehicle_exists(db, tenant_no, license_plate) - applicant_id = _resolve_applicant_id(db, tenant_no, payload.applicant_id, user) + if mine_vehicles: + applicant_id = user.id + else: + applicant_id = _resolve_applicant_id(db, tenant_no, payload.applicant_id, user) now = datetime.now() row = PlVehicleRechargeApply( id=generate_uuid(), @@ -608,18 +628,31 @@ def update_apply( user: DbUserContext, apply_id: str, payload: VehicleRechargeApplyUpdate, + *, + mine_vehicles: bool = False, ) -> VehicleRechargeApplyRead: - row = _get_apply_row(db, tenant_no, apply_id, user) + row = _get_apply_row( + db, tenant_no, apply_id, user, mine_vehicles=mine_vehicles + ) + if mine_vehicles: + _assert_own_apply_for_write(row, user) if row.status not in _EDITABLE_STATUSES: raise HTTPException(status_code=400, detail="仅草稿或已驳回申请可编辑") license_plate = payload.license_plate.strip() + if mine_vehicles: + _assert_mine_vehicle_plate(db, tenant_no, user, license_plate) _assert_vehicle_exists(db, tenant_no, license_plate) row.license_plate = license_plate row.recharge_type = payload.recharge_type row.apply_date = payload.apply_date row.odometer_km = payload.odometer_km row.attachment_urls = payload.attachment_urls - row.applicant_id = _resolve_applicant_id(db, tenant_no, payload.applicant_id, user) + if mine_vehicles: + row.applicant_id = user.id + else: + row.applicant_id = _resolve_applicant_id( + db, tenant_no, payload.applicant_id, user + ) row.remark = payload.remark row.update_by = user.id row.update_time = datetime.now() @@ -629,9 +662,18 @@ def update_apply( def cancel_apply( - db: Session, tenant_no: str, user: DbUserContext, apply_id: str + db: Session, + tenant_no: str, + user: DbUserContext, + apply_id: str, + *, + mine_vehicles: bool = False, ) -> VehicleRechargeApplyRead: - row = _get_apply_row(db, tenant_no, apply_id, user) + row = _get_apply_row( + db, tenant_no, apply_id, user, mine_vehicles=mine_vehicles + ) + if mine_vehicles: + _assert_own_apply_for_write(row, user) if row.status not in _EDITABLE_STATUSES: raise HTTPException(status_code=400, detail="仅草稿或已驳回申请可作废") row.status = "cancelled" @@ -643,9 +685,18 @@ def cancel_apply( def submit_apply( - db: Session, tenant_no: str, user: DbUserContext, apply_id: str + db: Session, + tenant_no: str, + user: DbUserContext, + apply_id: str, + *, + mine_vehicles: bool = False, ) -> VehicleRechargeApplyRead: - row = _get_apply_row(db, tenant_no, apply_id, user) + row = _get_apply_row( + db, tenant_no, apply_id, user, mine_vehicles=mine_vehicles + ) + if mine_vehicles: + _assert_own_apply_for_write(row, user) if row.status not in _EDITABLE_STATUSES: raise HTTPException(status_code=400, detail="仅草稿或已驳回申请可提交") workflow = _get_vehicle_recharge_workflow(db, tenant_no) diff --git a/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py b/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py index c75059275..ca0ba5c32 100644 --- a/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py +++ b/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py @@ -184,3 +184,46 @@ def test_get_apply_mine_vehicles_rejects_foreign_plate(monkeypatch): svc.get_apply(db, "t1", _user(), "a2", mine_vehicles=True) assert exc.value.status_code == 403 assert "无权" in exc.value.detail + + +def test_create_apply_mine_vehicles_rejects_foreign_plate(monkeypatch): + from fastapi import HTTPException + + from app.schemas.pl.vehicle_recharge import VehicleRechargeApplyCreate + from app.services import pl_vehicle_recharge_service as svc + + db = _make_session() + _seed_person_vehicle(db) + db.commit() + monkeypatch.setattr(svc, "apply_pl_scope", lambda query, *_a, **_k: query) + monkeypatch.setattr(svc, "assert_pl_row_in_scope", lambda *_a, **_k: None) + monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X") + monkeypatch.setattr(svc, "_resolve_applicant_id", lambda *_a, **_k: "u-driver") + monkeypatch.setattr(svc, "_assert_vehicle_exists", lambda *_a, **_k: None) + monkeypatch.setattr(svc, "generate_apply_no", lambda *_a, **_k: "SXSQTEST1") + + payload = VehicleRechargeApplyCreate( + licensePlate="鄂A99999", + rechargeType="etc", + applyDate=date.today(), + ) + with pytest.raises(HTTPException) as ei: + svc.create_apply(db, "t1", _user(), payload, mine_vehicles=True) + assert ei.value.status_code in (400, 403) + + +def test_update_cancel_submit_require_own_applicant(monkeypatch): + from fastapi import HTTPException + + from app.services import pl_vehicle_recharge_service as svc + + db = _make_session() + _seed_person_vehicle(db) + _seed_apply(db, apply_id="a1", plate="鄂A11111", applicant_id="u-office", status="draft") + db.commit() + monkeypatch.setattr(svc, "assert_pl_row_in_scope", lambda *_a, **_k: None) + monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X") + + with pytest.raises(HTTPException) as ei: + svc.cancel_apply(db, "t1", _user(), "a1", mine_vehicles=True) + assert ei.value.status_code == 403