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 <cursoragent@cursor.com>

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)

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