208 lines
8.1 KiB
Python
208 lines
8.1 KiB
Python
"""HTML 车辆卡片的真实汇总口径，使用 SQLite 验证查询与返回模型。"""
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.pl_models import PlVehicle, PlVehicleFuelRegister, PlPersonnel, PlVehiclePersonnel
|
|
from app.services import pl_vehicle_fuel_service as service
|
|
|
|
|
|
@pytest.fixture
|
|
def summary_db(monkeypatch):
|
|
class FixedDate(date):
|
|
@classmethod
|
|
def today(cls):
|
|
return cls(2026, 1, 26)
|
|
|
|
monkeypatch.setattr(service, "date", FixedDate)
|
|
engine = create_engine("sqlite://")
|
|
PlVehicle.__table__.create(engine)
|
|
PlVehicleFuelRegister.__table__.create(engine)
|
|
PlPersonnel.__table__.create(engine)
|
|
PlVehiclePersonnel.__table__.create(engine)
|
|
with Session(engine) as db:
|
|
yield db
|
|
engine.dispose()
|
|
|
|
|
|
def vehicle(db, plate="皖H57837", **kwargs):
|
|
values = dict(id=plate, license_plate=plate, vehicle_type="self", tenant_no="t1",
|
|
vehicle_status="available", create_by="u1")
|
|
values.update(kwargs)
|
|
db.add(PlVehicle(**values))
|
|
|
|
|
|
def fuel(db, day, liters, km, **kwargs):
|
|
key = str(db.query(PlVehicleFuelRegister).count())
|
|
values = dict(id=key, register_no=key, license_plate="皖H57837", tenant_no="t1",
|
|
fuel_category="gas_station", register_date=date.fromisoformat(day),
|
|
registrant_id="u1", fuel_liters=Decimal(str(liters)),
|
|
odometer_km=Decimal(str(km)), create_by="u1")
|
|
values.update(kwargs)
|
|
db.add(PlVehicleFuelRegister(**values))
|
|
db.flush()
|
|
|
|
|
|
def summary(db):
|
|
db.flush()
|
|
user = SimpleNamespace(to_user_info=lambda: {"role": "super_admin"})
|
|
return service.list_plate_month_summary(db, "t1", user)
|
|
|
|
|
|
def test_summary_includes_html_metrics_and_previous_quarter_across_year(summary_db):
|
|
db = summary_db
|
|
vehicle(db)
|
|
fuel(db, "2025-09-30", 10, 1000)
|
|
fuel(db, "2025-10-15", 100, 1500)
|
|
fuel(db, "2025-11-30", 100, 2000)
|
|
fuel(db, "2025-12-31", 200, 3000)
|
|
fuel(db, "2026-01-10", 120, 3400)
|
|
fuel(db, "2026-01-20", 180, 4000)
|
|
result = summary(db)
|
|
item = result.items[0].model_dump(by_alias=True)
|
|
assert (result.month, result.quarter) == ("2026-01", "2026-Q1")
|
|
assert item["liters"] == Decimal("300")
|
|
assert item["prevLiters"] == Decimal("200")
|
|
assert item["runKm"] == Decimal("600")
|
|
assert item["prevRunKm"] == Decimal("400")
|
|
assert item["monthRateL100"] == Decimal("50")
|
|
assert item["quarterRateL100"] == Decimal("50")
|
|
assert item["prevMonthRateL100"] == Decimal("50")
|
|
assert item["prevQuarterRateL100"] == Decimal("21.05")
|
|
|
|
|
|
def test_all_self_vehicles_visible_even_without_calculable_rate(summary_db):
|
|
db = summary_db
|
|
vehicle(db)
|
|
vehicle(db, "无记录车辆", vehicle_status="maintaining")
|
|
vehicle(db, "承运商车辆", vehicle_type="carrier")
|
|
vehicle(db, "其他租户", tenant_no="t2")
|
|
vehicle(db, "删除车辆", del_flag=1)
|
|
vehicle(db, "叉车")
|
|
vehicle(db, " 油桶 ")
|
|
fuel(db, "2026-01-10", 120, 3400)
|
|
items = {item.license_plate: item.model_dump(by_alias=True) for item in summary(db).items}
|
|
assert set(items) == {"皖H57837", "无记录车辆"}
|
|
assert items["皖H57837"]["liters"] == 120
|
|
assert items["皖H57837"]["monthRateL100"] is None
|
|
assert items["无记录车辆"]["liters"] == 0
|
|
assert items["无记录车辆"]["runKm"] is None
|
|
assert items["无记录车辆"]["monthMileageStatus"] == "insufficient"
|
|
|
|
|
|
def test_missing_boundary_month_falls_back_inside_period(summary_db):
|
|
db = summary_db
|
|
vehicle(db)
|
|
fuel(db, "2025-08-30", 10, 100)
|
|
# 缺失 9 月，不得把 8 月的里程当作上季度起点。
|
|
fuel(db, "2025-10-15", 100, 1000)
|
|
fuel(db, "2025-12-31", 100, 2000)
|
|
fuel(db, "2026-01-20", 100, 2500)
|
|
item = summary(db).items[0].model_dump(by_alias=True)
|
|
assert item["prevQuarterRateL100"] == Decimal("13.33")
|
|
# 上月只有一笔，但本月首笔可作为终点。
|
|
assert item["prevMonthRateL100"] == 20
|
|
|
|
|
|
def test_summary_excludes_invalid_records_and_applies_data_scope(summary_db, monkeypatch):
|
|
db = summary_db
|
|
vehicle(db)
|
|
vehicle(db, "权限外车辆", create_by="u2")
|
|
fuel(db, "2025-12-31", 100, 1000)
|
|
fuel(db, "2026-01-10", 100, 1500)
|
|
fuel(db, "2026-01-11", 900, 1600, status="cancelled")
|
|
fuel(db, "2026-01-12", 900, 1700, del_flag=1)
|
|
fuel(db, "2026-01-13", 900, 1800, tenant_no="t2")
|
|
fuel(db, "2026-01-14", 900, 1900, create_by="u2")
|
|
scoped_models = []
|
|
|
|
def scope(query, model, *args):
|
|
scoped_models.append(model)
|
|
return query.filter(model.create_by == "u1")
|
|
|
|
monkeypatch.setattr(service, "apply_pl_scope", scope)
|
|
items = summary(db).items
|
|
assert len(items) == 1
|
|
assert items[0].model_dump(by_alias=True)["liters"] == 100
|
|
assert items[0].month_rate_l100 is None
|
|
assert set(scoped_models) == {PlVehicle, PlVehicleFuelRegister}
|
|
|
|
|
|
def test_next_quarter_reading_is_loaded_without_adding_next_period_fuel(summary_db, monkeypatch):
|
|
class DecemberDate(date):
|
|
@classmethod
|
|
def today(cls):
|
|
return cls(2025, 12, 26)
|
|
monkeypatch.setattr(service, "date", DecemberDate)
|
|
db = summary_db
|
|
vehicle(db)
|
|
fuel(db, "2025-10-05", 100, 1000)
|
|
fuel(db, "2025-11-05", 100, 1500)
|
|
fuel(db, "2025-12-05", 100, 2000)
|
|
fuel(db, "2025-12-20", 200, 2500)
|
|
fuel(db, "2026-01-01", 999, 2700, status="cancelled")
|
|
fuel(db, "2026-01-02", 999, 2800, del_flag=1)
|
|
fuel(db, "2026-01-03", 999, 2900, tenant_no="t2")
|
|
fuel(db, "2026-01-05", 900, 3000)
|
|
fuel(db, "2026-01-10", 900, 3500)
|
|
item = summary(db).items[0].model_dump(by_alias=True)
|
|
assert item["liters"] == 300
|
|
assert item["runKm"] == 1000
|
|
assert item["monthRateL100"] == 30
|
|
assert item["quarterRateL100"] == 25
|
|
assert item["last3MonthsRateL100"] == 25
|
|
assert item["monthMileageStatus"] == "finalized"
|
|
assert item["quarterMileageStatus"] == "finalized"
|
|
|
|
|
|
def test_current_month_and_quarter_are_estimated_until_next_period(summary_db):
|
|
db = summary_db
|
|
vehicle(db)
|
|
fuel(db, "2026-01-01", 100, 1000)
|
|
fuel(db, "2026-01-20", 200, 1500)
|
|
item = summary(db).items[0].model_dump(by_alias=True)
|
|
assert item["runKm"] == 500
|
|
assert item["monthMileageStatus"] == "estimated"
|
|
assert item["quarterMileageStatus"] == "estimated"
|
|
fuel(db, "2026-02-02", 999, 2000)
|
|
item = summary(db).items[0].model_dump(by_alias=True)
|
|
assert item["runKm"] == 1000
|
|
assert item["monthMileageStatus"] == "finalized"
|
|
assert item["quarterMileageStatus"] == "estimated"
|
|
|
|
|
|
def test_rollback_exposes_status_and_no_rate(summary_db):
|
|
db = summary_db
|
|
vehicle(db)
|
|
fuel(db, "2026-01-01", 100, 1000)
|
|
fuel(db, "2026-01-20", 200, 900)
|
|
item = summary(db).items[0].model_dump(by_alias=True)
|
|
assert item["runKm"] is None
|
|
assert item["monthMileageStatus"] == "abnormal"
|
|
assert item["monthRateL100"] is None
|
|
assert item["quarterRateL100"] is None
|
|
|
|
|
|
def test_driver_name_uses_current_vehicle_personnel_association(summary_db):
|
|
db = summary_db
|
|
vehicle(db, default_driver_name="旧司机姓名")
|
|
vehicle(db, "未配置司机")
|
|
db.add_all([
|
|
PlPersonnel(id="driver", personnel_name="当前司机", tenant_no="t1"),
|
|
PlPersonnel(id="escort", personnel_name="押运员", tenant_no="t1"),
|
|
PlPersonnel(id="deleted", personnel_name="已删除司机", tenant_no="t1", del_flag=1),
|
|
PlVehiclePersonnel(id="driver-link", vehicle_id="皖H57837", personnel_id="driver",
|
|
role="driver", is_default=1, tenant_no="t1"),
|
|
PlVehiclePersonnel(id="escort-link", vehicle_id="皖H57837", personnel_id="escort",
|
|
role="escort", is_default=1, tenant_no="t1"),
|
|
PlVehiclePersonnel(id="deleted-link", vehicle_id="皖H57837", personnel_id="deleted",
|
|
role="driver", is_default=1, tenant_no="t1"),
|
|
])
|
|
items = {item.license_plate: item for item in summary(db).items}
|
|
assert items["皖H57837"].default_driver_name == "当前司机"
|
|
assert items["未配置司机"].default_driver_name is None
|