shanxia2.0/backend/tests/test_pl_vehicle_recharge_mine_vehicles.py

"""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.models import SysRole, SysUserRole
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__,
        SysRole.__table__,
        SysUserRole.__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 _seed_admin_role(db, *, user_id="u-admin", role_name="管理员", role_id="role-admin"):
    db.add(
        SysRole(
            id=role_id,
            role_name=role_name,
            status="1",
            tenant_no="t1",
            create_dept_id="d1",
            create_time=datetime.now(),
            update_time=datetime.now(),
            del_flag=0,
        )
    )
    db.add(
        SysUserRole(
            id=f"ur-{role_id}",
            user_id=user_id,
            role_id=role_id,
            tenant_no="t1",
            create_dept_id="d1",
            create_time=datetime.now(),
            update_time=datetime.now(),
            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()


def test_user_is_pl_recharge_admin_by_role_name():
    from app.services.pl_vehicle_mine_vehicles import (
        should_restrict_mine_vehicles,
        user_is_pl_recharge_admin,
    )

    db = _make_session()
    _seed_admin_role(db, user_id="u-admin", role_name="管理员")
    _seed_admin_role(
        db, user_id="u-super", role_name="超级管理员", role_id="role-super"
    )
    db.commit()
    assert user_is_pl_recharge_admin(db, "t1", "u-admin") is True
    assert user_is_pl_recharge_admin(db, "t1", "u-super") is True
    assert user_is_pl_recharge_admin(db, "t1", "u-driver") is False
    assert (
        should_restrict_mine_vehicles(db, "t1", "u-admin", mine_vehicles=True) is False
    )
    assert (
        should_restrict_mine_vehicles(db, "t1", "u-driver", mine_vehicles=True) is True
    )
    assert (
        should_restrict_mine_vehicles(db, "t1", "u-admin", mine_vehicles=False) is False
    )


def _seed_apply(db, *, apply_id, plate, applicant_id, status="draft"):
    db.add(
        PlVehicleRechargeApply(
            id=apply_id,
            apply_no=f"SXSQ{apply_id}",
            license_plate=plate,
            recharge_type="etc",
            apply_date=date.today(),
            applicant_id=applicant_id,
            status=status,
            tenant_no="t1",
            create_by=applicant_id,
            create_dept_id="d1",
            create_time=datetime.now(),
            update_time=datetime.now(),
            del_flag=0,
        )
    )


def test_list_applies_mine_vehicles_includes_others_on_my_plate(monkeypatch):
    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")
    _seed_apply(db, apply_id="a2", plate="鄂A99999", applicant_id="u-office")
    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,
        "apply_to_read",
        lambda row, db, **kw: SimpleNamespace(
            id=row.id, licensePlate=row.license_plate, applicantId=row.applicant_id
        ),
    )

    items, total = svc.list_applies(
        db, "t1", _user(), mine_vehicles=True, page=1, page_size=20
    )
    assert total == 1
    assert items[0].id == "a1"


def test_list_applies_mine_vehicles_admin_sees_all_plates(monkeypatch):
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_admin_role(db, user_id="u-admin")
    _seed_apply(db, apply_id="a1", plate="鄂A11111", applicant_id="u-office")
    _seed_apply(db, apply_id="a2", plate="鄂A99999", applicant_id="u-office")
    db.commit()
    monkeypatch.setattr(svc, "apply_pl_scope", lambda query, *_a, **_k: query)
    monkeypatch.setattr(
        svc,
        "apply_to_read",
        lambda row, db, **kw: SimpleNamespace(id=row.id),
    )

    items, total = svc.list_applies(
        db, "t1", _user("u-admin"), mine_vehicles=True, page=1, page_size=20
    )
    assert total == 2
    assert {i.id for i in items} == {"a1", "a2"}


def test_get_apply_mine_vehicles_allows_others_on_my_plate(monkeypatch):
    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")
    db.commit()
    monkeypatch.setattr(
        svc,
        "assert_pl_row_in_scope",
        lambda *_a, **_k: (_ for _ in ()).throw(
            AssertionError("should not use data scope when mine_vehicles")
        ),
    )
    monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X")
    monkeypatch.setattr(
        svc,
        "apply_to_read",
        lambda row, db, **kw: SimpleNamespace(id=row.id),
    )

    result = svc.get_apply(db, "t1", _user(), "a1", mine_vehicles=True)
    assert result.id == "a1" or getattr(result, "id", None) == "a1"


def test_get_apply_mine_vehicles_rejects_foreign_plate(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="a2", plate="鄂A99999", applicant_id="u-office")
    db.commit()
    monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X")

    with pytest.raises(HTTPException) as exc:
        svc.get_apply(db, "t1", _user(), "a2", mine_vehicles=True)
    assert exc.value.status_code == 403
    assert "无权" in exc.value.detail


def test_get_apply_mine_vehicles_admin_allows_foreign_plate(monkeypatch):
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_admin_role(db, user_id="u-admin")
    _seed_apply(db, apply_id="a2", plate="鄂A99999", applicant_id="u-office")
    db.commit()
    monkeypatch.setattr(
        svc,
        "assert_pl_row_in_scope",
        lambda *_a, **_k: (_ for _ in ()).throw(
            AssertionError("admin mineVehicles should skip data scope")
        ),
    )
    monkeypatch.setattr(
        svc,
        "apply_to_read",
        lambda row, db, **kw: SimpleNamespace(id=row.id),
    )

    result = svc.get_apply(db, "t1", _user("u-admin"), "a2", mine_vehicles=True)
    assert result.id == "a2"


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(),
        odometerKm=1000,
        attachmentUrls="http://example.com/a.jpg",
    )
    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_create_apply_mine_vehicles_admin_allows_foreign_plate(monkeypatch):
    from app.schemas.pl.vehicle_recharge import VehicleRechargeApplyCreate
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_admin_role(db, user_id="u-admin")
    db.commit()
    monkeypatch.setattr(svc, "_assert_vehicle_exists", lambda *_a, **_k: None)
    monkeypatch.setattr(svc, "generate_apply_no", lambda *_a, **_k: "SXSQADMIN1")
    monkeypatch.setattr(
        svc,
        "apply_to_read",
        lambda row, db, **kw: SimpleNamespace(
            id=row.id, licensePlate=row.license_plate, applicantId=row.applicant_id
        ),
    )

    payload = VehicleRechargeApplyCreate(
        licensePlate="鄂A99999",
        rechargeType="etc",
        applyDate=date.today(),
        odometerKm=1000,
        attachmentUrls="http://example.com/a.jpg",
    )
    result = svc.create_apply(
        db, "t1", _user("u-admin"), payload, mine_vehicles=True
    )
    assert result.licensePlate == "鄂A99999"
    assert result.applicantId == "u-admin"


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


def test_admin_cancel_still_requires_own_applicant(monkeypatch):
    from fastapi import HTTPException

    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_admin_role(db, user_id="u-admin")
    _seed_apply(
        db, apply_id="a1", plate="鄂A99999", applicant_id="u-office", status="draft"
    )
    db.commit()
    monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "X")

    with pytest.raises(HTTPException) as ei:
        svc.cancel_apply(db, "t1", _user("u-admin"), "a1", mine_vehicles=True)
    assert ei.value.status_code == 403