功能：新增托盘业务报表接口
This commit is contained in:
焦龙言
@@ -0,0 +1,131 @@
|
||||
"""业务报表接口。"""
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Query, Request, Response
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from common.aspect.db_seesion import DBSessionDependency
|
||||
from common.aspect.interface_auth import UserInterfaceAuthDependency
|
||||
from common.aspect.pre_auth import CurrentUserDependency, PreAuthDependency
|
||||
from common.router import APIRouterPro
|
||||
from common.vo import PageResponseModel
|
||||
from module_admin.entity.vo.user_vo import CurrentUserModel
|
||||
from module_business.entity.vo.report_vo import (
|
||||
ConsumptionReportModel,
|
||||
ConsumptionReportQueryModel,
|
||||
FillingReportModel,
|
||||
FillingReportQueryModel,
|
||||
PalletInventoryReportModel,
|
||||
PalletInventoryReportQueryModel,
|
||||
PalletTraceReportModel,
|
||||
PalletTraceReportQueryModel,
|
||||
ReturnReportModel,
|
||||
ReturnReportQueryModel,
|
||||
ShipmentReportModel,
|
||||
ShipmentReportQueryModel,
|
||||
)
|
||||
from module_business.service.report_service import ReportService
|
||||
from utils.response_util import ResponseUtil
|
||||
|
||||
report_controller = APIRouterPro(
|
||||
prefix='/report',
|
||||
tags=['业务管理-只读报表'],
|
||||
dependencies=[PreAuthDependency()],
|
||||
)
|
||||
|
||||
|
||||
@report_controller.get(
|
||||
'/pallet-inventory/list',
|
||||
summary='获取托盘库存报表分页列表接口',
|
||||
response_model=PageResponseModel[PalletInventoryReportModel],
|
||||
dependencies=[UserInterfaceAuthDependency('report:palletinventory:list')],
|
||||
)
|
||||
async def get_pallet_inventory_report_list(
|
||||
request: Request,
|
||||
query_object: Annotated[PalletInventoryReportQueryModel, Query()],
|
||||
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
||||
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
||||
) -> Response:
|
||||
result = await ReportService.get_pallet_inventory_list_services(query_db, query_object, current_user)
|
||||
return ResponseUtil.success(model_content=result)
|
||||
|
||||
|
||||
@report_controller.get(
|
||||
'/filling/list',
|
||||
summary='获取装填记录报表分页列表接口',
|
||||
response_model=PageResponseModel[FillingReportModel],
|
||||
dependencies=[UserInterfaceAuthDependency('report:filling:list')],
|
||||
)
|
||||
async def get_filling_report_list(
|
||||
request: Request,
|
||||
query_object: Annotated[FillingReportQueryModel, Query()],
|
||||
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
||||
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
||||
) -> Response:
|
||||
result = await ReportService.get_filling_report_list_services(query_db, query_object, current_user)
|
||||
return ResponseUtil.success(model_content=result)
|
||||
|
||||
|
||||
@report_controller.get(
|
||||
'/shipment/list',
|
||||
summary='获取发货出库报表分页列表接口',
|
||||
response_model=PageResponseModel[ShipmentReportModel],
|
||||
dependencies=[UserInterfaceAuthDependency('report:shipment:list')],
|
||||
)
|
||||
async def get_shipment_report_list(
|
||||
request: Request,
|
||||
query_object: Annotated[ShipmentReportQueryModel, Query()],
|
||||
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
||||
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
||||
) -> Response:
|
||||
result = await ReportService.get_shipment_report_list_services(query_db, query_object, current_user)
|
||||
return ResponseUtil.success(model_content=result)
|
||||
|
||||
|
||||
@report_controller.get(
|
||||
'/consumption/list',
|
||||
summary='获取产线消耗报表分页列表接口',
|
||||
response_model=PageResponseModel[ConsumptionReportModel],
|
||||
dependencies=[UserInterfaceAuthDependency('report:consumption:list')],
|
||||
)
|
||||
async def get_consumption_report_list(
|
||||
request: Request,
|
||||
query_object: Annotated[ConsumptionReportQueryModel, Query()],
|
||||
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
||||
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
||||
) -> Response:
|
||||
result = await ReportService.get_consumption_report_list_services(query_db, query_object, current_user)
|
||||
return ResponseUtil.success(model_content=result)
|
||||
|
||||
|
||||
@report_controller.get(
|
||||
'/return/list',
|
||||
summary='获取退回仓库报表分页列表接口',
|
||||
response_model=PageResponseModel[ReturnReportModel],
|
||||
dependencies=[UserInterfaceAuthDependency('report:return:list')],
|
||||
)
|
||||
async def get_return_report_list(
|
||||
request: Request,
|
||||
query_object: Annotated[ReturnReportQueryModel, Query()],
|
||||
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
||||
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
||||
) -> Response:
|
||||
result = await ReportService.get_return_report_list_services(query_db, query_object, current_user)
|
||||
return ResponseUtil.success(model_content=result)
|
||||
|
||||
|
||||
@report_controller.get(
|
||||
'/pallet-trace/list',
|
||||
summary='获取托盘追溯报表分页列表接口',
|
||||
response_model=PageResponseModel[PalletTraceReportModel],
|
||||
dependencies=[UserInterfaceAuthDependency('report:pallettrace:list')],
|
||||
)
|
||||
async def get_pallet_trace_report_list(
|
||||
request: Request,
|
||||
query_object: Annotated[PalletTraceReportQueryModel, Query()],
|
||||
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
||||
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
||||
) -> Response:
|
||||
result = await ReportService.get_pallet_trace_list_services(query_db, query_object, current_user)
|
||||
return ResponseUtil.success(model_content=result)
|
||||
@@ -0,0 +1,494 @@
|
||||
from datetime import datetime, time, timedelta
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import ColumnElement, Select, false, or_, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import aliased
|
||||
from sqlalchemy.orm.attributes import InstrumentedAttribute
|
||||
|
||||
from common.vo import PageModel
|
||||
from module_admin.entity.do.user_do import SysUser
|
||||
from module_business.entity.do.master_do import (
|
||||
BizCustomer,
|
||||
BizCustomerSite,
|
||||
BizPallet,
|
||||
BizProductionLine,
|
||||
BizWarehouse,
|
||||
)
|
||||
from module_business.entity.do.operation_do import (
|
||||
BizFillingRecord,
|
||||
BizPalletEvent,
|
||||
BizPalletInventory,
|
||||
BizReturnItem,
|
||||
BizReturnOrder,
|
||||
BizShipmentItem,
|
||||
BizShipmentOrder,
|
||||
BizUsageRecord,
|
||||
)
|
||||
from module_business.entity.vo.report_vo import (
|
||||
ConsumptionReportModel,
|
||||
ConsumptionReportQueryModel,
|
||||
FillingReportModel,
|
||||
FillingReportQueryModel,
|
||||
PalletInventoryReportModel,
|
||||
PalletInventoryReportQueryModel,
|
||||
PalletTraceReportModel,
|
||||
PalletTraceReportQueryModel,
|
||||
ReturnReportModel,
|
||||
ReturnReportQueryModel,
|
||||
ShipmentReportModel,
|
||||
ShipmentReportQueryModel,
|
||||
)
|
||||
from utils.page_util import PageUtil
|
||||
|
||||
|
||||
class ReportDao:
|
||||
"""六类只读业务报表数据库查询。"""
|
||||
|
||||
@staticmethod
|
||||
def _time_conditions(
|
||||
column: InstrumentedAttribute,
|
||||
begin_time: str | None,
|
||||
end_time: str | None,
|
||||
) -> list[ColumnElement[bool]]:
|
||||
conditions = []
|
||||
if begin_time:
|
||||
begin_at = datetime.combine(datetime.strptime(begin_time, '%Y-%m-%d').date(), time.min)
|
||||
conditions.append(column >= begin_at)
|
||||
if end_time:
|
||||
end_at = datetime.combine(
|
||||
datetime.strptime(end_time, '%Y-%m-%d').date() + timedelta(days=1),
|
||||
time.min,
|
||||
)
|
||||
conditions.append(column < end_at)
|
||||
return conditions
|
||||
|
||||
@staticmethod
|
||||
def _scope_condition(
|
||||
column: InstrumentedAttribute,
|
||||
scope_ids: list[int] | None,
|
||||
) -> ColumnElement[bool] | None:
|
||||
if scope_ids is None:
|
||||
return None
|
||||
normalized_ids = sorted(set(scope_ids))
|
||||
return column.in_(normalized_ids) if normalized_ids else false()
|
||||
|
||||
@classmethod
|
||||
def _or_scope_conditions(
|
||||
cls,
|
||||
*column_scopes: tuple[InstrumentedAttribute, list[int] | None],
|
||||
) -> list[ColumnElement[bool]]:
|
||||
conditions = [
|
||||
condition
|
||||
for column, scope_ids in column_scopes
|
||||
if (condition := cls._scope_condition(column, scope_ids)) is not None
|
||||
]
|
||||
return [or_(*conditions)] if conditions else []
|
||||
|
||||
@staticmethod
|
||||
async def _paginate(
|
||||
db: AsyncSession,
|
||||
statement: Select,
|
||||
query_object: Any,
|
||||
) -> PageModel:
|
||||
return await PageUtil.paginate(
|
||||
db,
|
||||
statement,
|
||||
query_object.page_num,
|
||||
query_object.page_size,
|
||||
True,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
async def get_pallet_inventory_list(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
query_object: PalletInventoryReportQueryModel,
|
||||
warehouse_ids: list[int] | None = None,
|
||||
site_ids: list[int] | None = None,
|
||||
) -> PageModel[PalletInventoryReportModel]:
|
||||
current_warehouse = aliased(BizWarehouse, name='current_warehouse')
|
||||
current_site = aliased(BizCustomerSite, name='current_site')
|
||||
current_line = aliased(BizProductionLine, name='current_line')
|
||||
active_filling = aliased(BizFillingRecord, name='active_filling')
|
||||
|
||||
conditions = cls._or_scope_conditions(
|
||||
(BizPalletInventory.warehouse_id, warehouse_ids),
|
||||
(BizPalletInventory.site_id, site_ids),
|
||||
)
|
||||
if query_object.pallet_code:
|
||||
conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
|
||||
if query_object.flow_state:
|
||||
conditions.append(BizPalletInventory.flow_state == query_object.flow_state.value)
|
||||
if query_object.warehouse_id:
|
||||
conditions.append(BizPalletInventory.warehouse_id == query_object.warehouse_id)
|
||||
if query_object.site_id:
|
||||
conditions.append(BizPalletInventory.site_id == query_object.site_id)
|
||||
if query_object.line_id:
|
||||
conditions.append(BizPalletInventory.current_line_id == query_object.line_id)
|
||||
if query_object.strain_id:
|
||||
conditions.append(active_filling.strain_id == query_object.strain_id)
|
||||
if query_object.batch_no:
|
||||
conditions.append(active_filling.batch_no.like(f'%{query_object.batch_no}%'))
|
||||
conditions.extend(
|
||||
cls._time_conditions(
|
||||
BizPalletInventory.updated_at,
|
||||
query_object.begin_time,
|
||||
query_object.end_time,
|
||||
)
|
||||
)
|
||||
|
||||
statement = (
|
||||
select(
|
||||
BizPalletInventory.pallet_id.label('pallet_id'),
|
||||
BizPallet.pallet_code.label('pallet_code'),
|
||||
BizPalletInventory.flow_state.label('flow_state'),
|
||||
BizPalletInventory.warehouse_id.label('warehouse_id'),
|
||||
current_warehouse.warehouse_name.label('warehouse_name'),
|
||||
BizPalletInventory.site_id.label('site_id'),
|
||||
current_site.site_name.label('site_name'),
|
||||
BizPalletInventory.current_line_id.label('line_id'),
|
||||
current_line.line_name.label('line_name'),
|
||||
active_filling.strain_id.label('strain_id'),
|
||||
active_filling.strain_code_snapshot.label('strain_code'),
|
||||
active_filling.strain_name_snapshot.label('strain_name'),
|
||||
active_filling.batch_no.label('batch_no'),
|
||||
active_filling.initial_packages.label('initial_packages'),
|
||||
BizPalletInventory.remaining_packages.label('remaining_packages'),
|
||||
BizPalletInventory.updated_at.label('updated_at'),
|
||||
)
|
||||
.select_from(BizPalletInventory)
|
||||
.join(BizPallet, BizPallet.pallet_id == BizPalletInventory.pallet_id)
|
||||
.outerjoin(active_filling, active_filling.filling_id == BizPalletInventory.active_filling_id)
|
||||
.outerjoin(current_warehouse, current_warehouse.warehouse_id == BizPalletInventory.warehouse_id)
|
||||
.outerjoin(current_site, current_site.site_id == BizPalletInventory.site_id)
|
||||
.outerjoin(current_line, current_line.line_id == BizPalletInventory.current_line_id)
|
||||
.where(*conditions)
|
||||
.order_by(BizPalletInventory.pallet_id)
|
||||
)
|
||||
return await cls._paginate(db, statement, query_object)
|
||||
|
||||
@classmethod
|
||||
async def get_filling_report_list(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
query_object: FillingReportQueryModel,
|
||||
warehouse_ids: list[int] | None = None,
|
||||
) -> PageModel[FillingReportModel]:
|
||||
operator_user = aliased(SysUser, name='operator_user')
|
||||
conditions = []
|
||||
warehouse_scope = cls._scope_condition(BizFillingRecord.warehouse_id, warehouse_ids)
|
||||
if warehouse_scope is not None:
|
||||
conditions.append(warehouse_scope)
|
||||
if query_object.warehouse_id:
|
||||
conditions.append(BizFillingRecord.warehouse_id == query_object.warehouse_id)
|
||||
if query_object.pallet_code:
|
||||
conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
|
||||
if query_object.strain_id:
|
||||
conditions.append(BizFillingRecord.strain_id == query_object.strain_id)
|
||||
if query_object.batch_no:
|
||||
conditions.append(BizFillingRecord.batch_no.like(f'%{query_object.batch_no}%'))
|
||||
conditions.extend(
|
||||
cls._time_conditions(
|
||||
BizFillingRecord.operated_at,
|
||||
query_object.begin_time,
|
||||
query_object.end_time,
|
||||
)
|
||||
)
|
||||
|
||||
statement = (
|
||||
select(
|
||||
BizFillingRecord.filling_id.label('filling_id'),
|
||||
BizFillingRecord.filling_no.label('filling_no'),
|
||||
BizFillingRecord.operated_at.label('operated_at'),
|
||||
BizFillingRecord.warehouse_id.label('warehouse_id'),
|
||||
BizWarehouse.warehouse_name.label('warehouse_name'),
|
||||
BizFillingRecord.pallet_id.label('pallet_id'),
|
||||
BizPallet.pallet_code.label('pallet_code'),
|
||||
BizFillingRecord.strain_id.label('strain_id'),
|
||||
BizFillingRecord.strain_code_snapshot.label('strain_code_snapshot'),
|
||||
BizFillingRecord.strain_name_snapshot.label('strain_name_snapshot'),
|
||||
BizFillingRecord.batch_no.label('batch_no'),
|
||||
BizFillingRecord.package_spec.label('package_spec'),
|
||||
BizFillingRecord.initial_packages.label('initial_packages'),
|
||||
BizFillingRecord.operator_id.label('operator_id'),
|
||||
operator_user.user_name.label('operator_name'),
|
||||
)
|
||||
.select_from(BizFillingRecord)
|
||||
.join(BizPallet, BizPallet.pallet_id == BizFillingRecord.pallet_id)
|
||||
.outerjoin(BizWarehouse, BizWarehouse.warehouse_id == BizFillingRecord.warehouse_id)
|
||||
.outerjoin(operator_user, operator_user.user_id == BizFillingRecord.operator_id)
|
||||
.where(*conditions)
|
||||
.order_by(BizFillingRecord.filling_id)
|
||||
)
|
||||
return await cls._paginate(db, statement, query_object)
|
||||
|
||||
@classmethod
|
||||
async def get_shipment_report_list(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
query_object: ShipmentReportQueryModel,
|
||||
warehouse_ids: list[int] | None = None,
|
||||
site_ids: list[int] | None = None,
|
||||
) -> PageModel[ShipmentReportModel]:
|
||||
source_warehouse = aliased(BizWarehouse, name='source_warehouse')
|
||||
target_site = aliased(BizCustomerSite, name='target_site')
|
||||
target_customer = aliased(BizCustomer, name='target_customer')
|
||||
conditions = []
|
||||
warehouse_scope = cls._scope_condition(BizShipmentOrder.source_warehouse_id, warehouse_ids)
|
||||
site_scope = cls._scope_condition(BizShipmentOrder.target_site_id, site_ids)
|
||||
if warehouse_scope is not None:
|
||||
conditions.append(warehouse_scope)
|
||||
if site_scope is not None:
|
||||
conditions.append(site_scope)
|
||||
if query_object.shipment_no:
|
||||
conditions.append(BizShipmentOrder.shipment_no.like(f'%{query_object.shipment_no}%'))
|
||||
if query_object.source_warehouse_id:
|
||||
conditions.append(BizShipmentOrder.source_warehouse_id == query_object.source_warehouse_id)
|
||||
if query_object.target_site_id:
|
||||
conditions.append(BizShipmentOrder.target_site_id == query_object.target_site_id)
|
||||
if query_object.pallet_code:
|
||||
conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
|
||||
if query_object.status:
|
||||
conditions.append(BizShipmentOrder.status == query_object.status.value)
|
||||
conditions.extend(
|
||||
cls._time_conditions(
|
||||
BizShipmentOrder.confirmed_at,
|
||||
query_object.begin_time,
|
||||
query_object.end_time,
|
||||
)
|
||||
)
|
||||
# 报表行契约要求确认时间和正数包数，仅确认后的事实明细可形成报表行。
|
||||
conditions.extend(
|
||||
(
|
||||
BizShipmentOrder.confirmed_at.is_not(None),
|
||||
BizShipmentItem.package_qty.is_not(None),
|
||||
)
|
||||
)
|
||||
|
||||
statement = (
|
||||
select(
|
||||
BizShipmentOrder.shipment_id.label('shipment_id'),
|
||||
BizShipmentOrder.shipment_no.label('shipment_no'),
|
||||
BizShipmentOrder.confirmed_at.label('confirmed_at'),
|
||||
BizShipmentOrder.source_warehouse_id.label('source_warehouse_id'),
|
||||
source_warehouse.warehouse_name.label('source_warehouse_name'),
|
||||
BizShipmentOrder.target_site_id.label('target_site_id'),
|
||||
target_site.site_name.label('target_site_name'),
|
||||
target_site.customer_id.label('customer_id'),
|
||||
target_customer.customer_name.label('customer_name'),
|
||||
BizShipmentItem.pallet_id.label('pallet_id'),
|
||||
BizPallet.pallet_code.label('pallet_code'),
|
||||
BizShipmentItem.active_filling_id.label('active_filling_id'),
|
||||
BizShipmentItem.package_qty.label('package_qty'),
|
||||
)
|
||||
.select_from(BizShipmentOrder)
|
||||
.join(BizShipmentItem, BizShipmentItem.shipment_id == BizShipmentOrder.shipment_id)
|
||||
.join(BizPallet, BizPallet.pallet_id == BizShipmentItem.pallet_id)
|
||||
.outerjoin(source_warehouse, source_warehouse.warehouse_id == BizShipmentOrder.source_warehouse_id)
|
||||
.outerjoin(target_site, target_site.site_id == BizShipmentOrder.target_site_id)
|
||||
.outerjoin(target_customer, target_customer.customer_id == target_site.customer_id)
|
||||
.where(*conditions)
|
||||
.order_by(BizShipmentOrder.shipment_id, BizShipmentItem.shipment_item_id)
|
||||
)
|
||||
return await cls._paginate(db, statement, query_object)
|
||||
|
||||
@classmethod
|
||||
async def get_consumption_report_list(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
query_object: ConsumptionReportQueryModel,
|
||||
site_ids: list[int] | None = None,
|
||||
) -> PageModel[ConsumptionReportModel]:
|
||||
operator_user = aliased(SysUser, name='operator_user')
|
||||
conditions = []
|
||||
site_scope = cls._scope_condition(BizUsageRecord.site_id, site_ids)
|
||||
if site_scope is not None:
|
||||
conditions.append(site_scope)
|
||||
if query_object.customer_id:
|
||||
conditions.append(BizCustomerSite.customer_id == query_object.customer_id)
|
||||
if query_object.site_id:
|
||||
conditions.append(BizUsageRecord.site_id == query_object.site_id)
|
||||
if query_object.line_id:
|
||||
conditions.append(BizUsageRecord.line_id == query_object.line_id)
|
||||
if query_object.pallet_code:
|
||||
conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
|
||||
if query_object.strain_id:
|
||||
conditions.append(BizFillingRecord.strain_id == query_object.strain_id)
|
||||
if query_object.batch_no:
|
||||
conditions.append(BizFillingRecord.batch_no.like(f'%{query_object.batch_no}%'))
|
||||
conditions.extend(
|
||||
cls._time_conditions(
|
||||
BizUsageRecord.occurred_at,
|
||||
query_object.begin_time,
|
||||
query_object.end_time,
|
||||
)
|
||||
)
|
||||
|
||||
statement = (
|
||||
select(
|
||||
BizUsageRecord.usage_id.label('usage_id'),
|
||||
BizUsageRecord.usage_no.label('usage_no'),
|
||||
BizUsageRecord.occurred_at.label('occurred_at'),
|
||||
BizCustomerSite.customer_id.label('customer_id'),
|
||||
BizCustomer.customer_name.label('customer_name'),
|
||||
BizUsageRecord.site_id.label('site_id'),
|
||||
BizCustomerSite.site_name.label('site_name'),
|
||||
BizUsageRecord.line_id.label('line_id'),
|
||||
BizProductionLine.line_name.label('line_name'),
|
||||
BizUsageRecord.pallet_id.label('pallet_id'),
|
||||
BizPallet.pallet_code.label('pallet_code'),
|
||||
BizFillingRecord.strain_id.label('strain_id'),
|
||||
BizFillingRecord.strain_code_snapshot.label('strain_code'),
|
||||
BizFillingRecord.strain_name_snapshot.label('strain_name'),
|
||||
BizFillingRecord.batch_no.label('batch_no'),
|
||||
BizUsageRecord.used_packages.label('used_packages'),
|
||||
BizUsageRecord.remaining_packages_after.label('remaining_packages_after'),
|
||||
BizUsageRecord.operator_id.label('operator_id'),
|
||||
operator_user.user_name.label('operator_name'),
|
||||
)
|
||||
.select_from(BizUsageRecord)
|
||||
.join(BizPallet, BizPallet.pallet_id == BizUsageRecord.pallet_id)
|
||||
.join(BizFillingRecord, BizFillingRecord.filling_id == BizUsageRecord.active_filling_id)
|
||||
.join(BizCustomerSite, BizCustomerSite.site_id == BizUsageRecord.site_id)
|
||||
.join(BizProductionLine, BizProductionLine.line_id == BizUsageRecord.line_id)
|
||||
.outerjoin(BizCustomer, BizCustomer.customer_id == BizCustomerSite.customer_id)
|
||||
.outerjoin(operator_user, operator_user.user_id == BizUsageRecord.operator_id)
|
||||
.where(*conditions)
|
||||
.order_by(BizUsageRecord.usage_id)
|
||||
)
|
||||
return await cls._paginate(db, statement, query_object)
|
||||
|
||||
@classmethod
|
||||
async def get_return_report_list(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
query_object: ReturnReportQueryModel,
|
||||
site_ids: list[int] | None = None,
|
||||
warehouse_ids: list[int] | None = None,
|
||||
) -> PageModel[ReturnReportModel]:
|
||||
source_site = aliased(BizCustomerSite, name='source_site')
|
||||
target_warehouse = aliased(BizWarehouse, name='target_warehouse')
|
||||
conditions = cls._or_scope_conditions(
|
||||
(BizReturnOrder.source_site_id, site_ids),
|
||||
(BizReturnOrder.target_warehouse_id, warehouse_ids),
|
||||
)
|
||||
if query_object.return_no:
|
||||
conditions.append(BizReturnOrder.return_no.like(f'%{query_object.return_no}%'))
|
||||
if query_object.source_site_id:
|
||||
conditions.append(BizReturnOrder.source_site_id == query_object.source_site_id)
|
||||
if query_object.target_warehouse_id:
|
||||
conditions.append(BizReturnOrder.target_warehouse_id == query_object.target_warehouse_id)
|
||||
if query_object.pallet_code:
|
||||
conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
|
||||
if query_object.status:
|
||||
conditions.append(BizReturnOrder.status == query_object.status.value)
|
||||
conditions.extend(
|
||||
cls._time_conditions(
|
||||
BizReturnOrder.requested_at,
|
||||
query_object.begin_time,
|
||||
query_object.end_time,
|
||||
)
|
||||
)
|
||||
|
||||
statement = (
|
||||
select(
|
||||
BizReturnOrder.return_id.label('return_id'),
|
||||
BizReturnOrder.return_no.label('return_no'),
|
||||
BizReturnOrder.status.label('status'),
|
||||
BizReturnOrder.requested_at.label('requested_at'),
|
||||
BizReturnOrder.confirmed_at.label('confirmed_at'),
|
||||
BizReturnOrder.source_site_id.label('source_site_id'),
|
||||
source_site.site_name.label('source_site_name'),
|
||||
BizReturnOrder.target_warehouse_id.label('target_warehouse_id'),
|
||||
target_warehouse.warehouse_name.label('target_warehouse_name'),
|
||||
BizReturnItem.pallet_id.label('pallet_id'),
|
||||
BizPallet.pallet_code.label('pallet_code'),
|
||||
BizReturnItem.active_filling_id_snapshot.label('active_filling_id_snapshot'),
|
||||
BizReturnItem.remaining_packages_snapshot.label('remaining_packages_snapshot'),
|
||||
)
|
||||
.select_from(BizReturnOrder)
|
||||
.join(BizReturnItem, BizReturnItem.return_id == BizReturnOrder.return_id)
|
||||
.join(BizPallet, BizPallet.pallet_id == BizReturnItem.pallet_id)
|
||||
.outerjoin(source_site, source_site.site_id == BizReturnOrder.source_site_id)
|
||||
.outerjoin(target_warehouse, target_warehouse.warehouse_id == BizReturnOrder.target_warehouse_id)
|
||||
.where(*conditions)
|
||||
.order_by(BizReturnOrder.return_id, BizReturnItem.return_item_id)
|
||||
)
|
||||
return await cls._paginate(db, statement, query_object)
|
||||
|
||||
@classmethod
|
||||
async def get_pallet_trace_list(
|
||||
cls,
|
||||
db: AsyncSession,
|
||||
query_object: PalletTraceReportQueryModel,
|
||||
warehouse_ids: list[int] | None = None,
|
||||
site_ids: list[int] | None = None,
|
||||
) -> PageModel[PalletTraceReportModel]:
|
||||
before_warehouse = aliased(BizWarehouse, name='before_warehouse')
|
||||
after_warehouse = aliased(BizWarehouse, name='after_warehouse')
|
||||
before_site = aliased(BizCustomerSite, name='before_site')
|
||||
after_site = aliased(BizCustomerSite, name='after_site')
|
||||
before_line = aliased(BizProductionLine, name='before_line')
|
||||
after_line = aliased(BizProductionLine, name='after_line')
|
||||
operator_user = aliased(SysUser, name='operator_user')
|
||||
conditions = cls._or_scope_conditions(
|
||||
(BizPalletEvent.before_warehouse_id, warehouse_ids),
|
||||
(BizPalletEvent.after_warehouse_id, warehouse_ids),
|
||||
(BizPalletEvent.before_site_id, site_ids),
|
||||
(BizPalletEvent.after_site_id, site_ids),
|
||||
)
|
||||
if query_object.pallet_code:
|
||||
conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%'))
|
||||
if query_object.event_type:
|
||||
conditions.append(BizPalletEvent.event_type == query_object.event_type.value)
|
||||
conditions.extend(
|
||||
cls._time_conditions(
|
||||
BizPalletEvent.occurred_at,
|
||||
query_object.begin_time,
|
||||
query_object.end_time,
|
||||
)
|
||||
)
|
||||
|
||||
statement = (
|
||||
select(
|
||||
BizPalletEvent.event_id.label('event_id'),
|
||||
BizPalletEvent.event_seq.label('event_seq'),
|
||||
BizPalletEvent.event_type.label('event_type'),
|
||||
BizPalletEvent.source_type.label('source_type'),
|
||||
BizPalletEvent.source_id.label('source_id'),
|
||||
BizPalletEvent.pallet_id.label('pallet_id'),
|
||||
BizPallet.pallet_code.label('pallet_code'),
|
||||
BizPalletEvent.before_state.label('before_state'),
|
||||
BizPalletEvent.after_state.label('after_state'),
|
||||
BizPalletEvent.before_warehouse_id.label('before_warehouse_id'),
|
||||
before_warehouse.warehouse_name.label('before_warehouse_name'),
|
||||
BizPalletEvent.before_site_id.label('before_site_id'),
|
||||
before_site.site_name.label('before_site_name'),
|
||||
BizPalletEvent.before_line_id.label('before_line_id'),
|
||||
before_line.line_name.label('before_line_name'),
|
||||
BizPalletEvent.after_warehouse_id.label('after_warehouse_id'),
|
||||
after_warehouse.warehouse_name.label('after_warehouse_name'),
|
||||
BizPalletEvent.after_site_id.label('after_site_id'),
|
||||
after_site.site_name.label('after_site_name'),
|
||||
BizPalletEvent.after_line_id.label('after_line_id'),
|
||||
after_line.line_name.label('after_line_name'),
|
||||
BizPalletEvent.before_remaining_packages.label('before_remaining_packages'),
|
||||
BizPalletEvent.after_remaining_packages.label('after_remaining_packages'),
|
||||
BizPalletEvent.operator_id.label('operator_id'),
|
||||
operator_user.user_name.label('operator_name'),
|
||||
BizPalletEvent.occurred_at.label('occurred_at'),
|
||||
)
|
||||
.select_from(BizPalletEvent)
|
||||
.join(BizPallet, BizPallet.pallet_id == BizPalletEvent.pallet_id)
|
||||
.outerjoin(before_warehouse, before_warehouse.warehouse_id == BizPalletEvent.before_warehouse_id)
|
||||
.outerjoin(after_warehouse, after_warehouse.warehouse_id == BizPalletEvent.after_warehouse_id)
|
||||
.outerjoin(before_site, before_site.site_id == BizPalletEvent.before_site_id)
|
||||
.outerjoin(after_site, after_site.site_id == BizPalletEvent.after_site_id)
|
||||
.outerjoin(before_line, before_line.line_id == BizPalletEvent.before_line_id)
|
||||
.outerjoin(after_line, after_line.line_id == BizPalletEvent.after_line_id)
|
||||
.outerjoin(operator_user, operator_user.user_id == BizPalletEvent.operator_id)
|
||||
.where(*conditions)
|
||||
.order_by(BizPalletEvent.pallet_id, BizPalletEvent.event_seq, BizPalletEvent.event_id)
|
||||
)
|
||||
return await cls._paginate(db, statement, query_object)
|
||||
@@ -0,0 +1,196 @@
|
||||
from typing import Any, TypeVar
|
||||
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from common.vo import PageModel
|
||||
from exceptions.exception import ServiceException
|
||||
from module_admin.entity.vo.user_vo import CurrentUserModel
|
||||
from module_business.dao.access_dao import AccessDao
|
||||
from module_business.dao.report_dao import ReportDao
|
||||
from module_business.entity.vo.report_vo import (
|
||||
ConsumptionReportModel,
|
||||
ConsumptionReportQueryModel,
|
||||
FillingReportModel,
|
||||
FillingReportQueryModel,
|
||||
PalletInventoryReportModel,
|
||||
PalletInventoryReportQueryModel,
|
||||
PalletTraceReportModel,
|
||||
PalletTraceReportQueryModel,
|
||||
ReturnReportModel,
|
||||
ReturnReportQueryModel,
|
||||
ShipmentReportModel,
|
||||
ShipmentReportQueryModel,
|
||||
)
|
||||
|
||||
RowModelT = TypeVar('RowModelT', bound=BaseModel)
|
||||
|
||||
|
||||
class ReportService:
|
||||
"""六类只读业务报表服务。"""
|
||||
|
||||
@staticmethod
|
||||
def _is_admin(current_user: CurrentUserModel) -> bool:
|
||||
return bool(current_user.user and current_user.user.admin)
|
||||
|
||||
@classmethod
|
||||
def _user_id(cls, current_user: CurrentUserModel) -> int:
|
||||
if current_user.user is None or current_user.user.user_id is None:
|
||||
raise ServiceException(message='当前用户信息无效')
|
||||
return current_user.user.user_id
|
||||
|
||||
@classmethod
|
||||
async def _warehouse_scope(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
current_user: CurrentUserModel,
|
||||
) -> list[int] | None:
|
||||
if cls._is_admin(current_user):
|
||||
return None
|
||||
warehouse_ids = await AccessDao.list_user_warehouse_ids(query_db, cls._user_id(current_user))
|
||||
return sorted(set(warehouse_ids))
|
||||
|
||||
@classmethod
|
||||
async def _site_scope(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
current_user: CurrentUserModel,
|
||||
) -> list[int] | None:
|
||||
if cls._is_admin(current_user):
|
||||
return None
|
||||
site_ids = await AccessDao.list_user_site_ids(query_db, cls._user_id(current_user))
|
||||
return sorted(set(site_ids))
|
||||
|
||||
@classmethod
|
||||
async def _scopes(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
current_user: CurrentUserModel,
|
||||
) -> tuple[list[int] | None, list[int] | None]:
|
||||
if cls._is_admin(current_user):
|
||||
return None, None
|
||||
return (
|
||||
await cls._warehouse_scope(query_db, current_user),
|
||||
await cls._site_scope(query_db, current_user),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _empty_page(query_object: Any) -> PageModel:
|
||||
return PageModel(
|
||||
rows=[],
|
||||
pageNum=query_object.page_num,
|
||||
pageSize=query_object.page_size,
|
||||
total=0,
|
||||
hasNext=False,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _convert_page(page: PageModel, row_model: type[RowModelT]) -> PageModel[RowModelT]:
|
||||
rows = [row_model.model_validate(row) for row in page.rows]
|
||||
return page.model_copy(update={'rows': rows})
|
||||
|
||||
@classmethod
|
||||
async def get_pallet_inventory_list_services(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
query_object: PalletInventoryReportQueryModel,
|
||||
current_user: CurrentUserModel,
|
||||
) -> PageModel[PalletInventoryReportModel]:
|
||||
warehouse_ids, site_ids = await cls._scopes(query_db, current_user)
|
||||
if warehouse_ids == [] and site_ids == []:
|
||||
return cls._empty_page(query_object)
|
||||
page = await ReportDao.get_pallet_inventory_list(
|
||||
query_db,
|
||||
query_object,
|
||||
warehouse_ids=warehouse_ids,
|
||||
site_ids=site_ids,
|
||||
)
|
||||
return cls._convert_page(page, PalletInventoryReportModel)
|
||||
|
||||
@classmethod
|
||||
async def get_filling_report_list_services(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
query_object: FillingReportQueryModel,
|
||||
current_user: CurrentUserModel,
|
||||
) -> PageModel[FillingReportModel]:
|
||||
warehouse_ids = await cls._warehouse_scope(query_db, current_user)
|
||||
if warehouse_ids == []:
|
||||
return cls._empty_page(query_object)
|
||||
page = await ReportDao.get_filling_report_list(
|
||||
query_db,
|
||||
query_object,
|
||||
warehouse_ids=warehouse_ids,
|
||||
)
|
||||
return cls._convert_page(page, FillingReportModel)
|
||||
|
||||
@classmethod
|
||||
async def get_shipment_report_list_services(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
query_object: ShipmentReportQueryModel,
|
||||
current_user: CurrentUserModel,
|
||||
) -> PageModel[ShipmentReportModel]:
|
||||
warehouse_ids, site_ids = await cls._scopes(query_db, current_user)
|
||||
if warehouse_ids == [] or site_ids == []:
|
||||
return cls._empty_page(query_object)
|
||||
page = await ReportDao.get_shipment_report_list(
|
||||
query_db,
|
||||
query_object,
|
||||
warehouse_ids=warehouse_ids,
|
||||
site_ids=site_ids,
|
||||
)
|
||||
return cls._convert_page(page, ShipmentReportModel)
|
||||
|
||||
@classmethod
|
||||
async def get_consumption_report_list_services(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
query_object: ConsumptionReportQueryModel,
|
||||
current_user: CurrentUserModel,
|
||||
) -> PageModel[ConsumptionReportModel]:
|
||||
site_ids = await cls._site_scope(query_db, current_user)
|
||||
if site_ids == []:
|
||||
return cls._empty_page(query_object)
|
||||
page = await ReportDao.get_consumption_report_list(
|
||||
query_db,
|
||||
query_object,
|
||||
site_ids=site_ids,
|
||||
)
|
||||
return cls._convert_page(page, ConsumptionReportModel)
|
||||
|
||||
@classmethod
|
||||
async def get_return_report_list_services(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
query_object: ReturnReportQueryModel,
|
||||
current_user: CurrentUserModel,
|
||||
) -> PageModel[ReturnReportModel]:
|
||||
warehouse_ids, site_ids = await cls._scopes(query_db, current_user)
|
||||
if warehouse_ids == [] and site_ids == []:
|
||||
return cls._empty_page(query_object)
|
||||
page = await ReportDao.get_return_report_list(
|
||||
query_db,
|
||||
query_object,
|
||||
site_ids=site_ids,
|
||||
warehouse_ids=warehouse_ids,
|
||||
)
|
||||
return cls._convert_page(page, ReturnReportModel)
|
||||
|
||||
@classmethod
|
||||
async def get_pallet_trace_list_services(
|
||||
cls,
|
||||
query_db: AsyncSession,
|
||||
query_object: PalletTraceReportQueryModel,
|
||||
current_user: CurrentUserModel,
|
||||
) -> PageModel[PalletTraceReportModel]:
|
||||
warehouse_ids, site_ids = await cls._scopes(query_db, current_user)
|
||||
if warehouse_ids == [] and site_ids == []:
|
||||
return cls._empty_page(query_object)
|
||||
page = await ReportDao.get_pallet_trace_list(
|
||||
query_db,
|
||||
query_object,
|
||||
warehouse_ids=warehouse_ids,
|
||||
site_ids=site_ids,
|
||||
)
|
||||
return cls._convert_page(page, PalletTraceReportModel)
|
||||
@@ -0,0 +1,989 @@
|
||||
import ast
|
||||
import importlib.util
|
||||
import inspect
|
||||
import unittest
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from fastapi.routing import APIRoute
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import Select
|
||||
from sqlalchemy.dialects import mysql, postgresql
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from common.aspect.interface_auth import CheckUserInterfaceAuth
|
||||
from common.aspect.pre_auth import PreAuth
|
||||
from common.vo import PageModel, PageResponseModel
|
||||
from module_admin.entity.vo.user_vo import CurrentUserModel, UserInfoModel
|
||||
from module_admin.service.login_service import LoginService
|
||||
from module_business.controller import report_controller as report_controller_module
|
||||
from module_business.dao import report_dao as report_dao_module
|
||||
from module_business.dao.access_dao import AccessDao
|
||||
from module_business.domain.enums import DocumentStatus, PalletEventType, PalletFlowState, ReturnStatus
|
||||
from module_business.entity.vo.report_vo import (
|
||||
ConsumptionReportModel,
|
||||
ConsumptionReportQueryModel,
|
||||
FillingReportModel,
|
||||
FillingReportQueryModel,
|
||||
PalletInventoryReportModel,
|
||||
PalletInventoryReportQueryModel,
|
||||
PalletTraceReportModel,
|
||||
PalletTraceReportQueryModel,
|
||||
ReturnReportModel,
|
||||
ReturnReportQueryModel,
|
||||
ShipmentReportModel,
|
||||
ShipmentReportQueryModel,
|
||||
)
|
||||
from module_business.service import report_service as report_service_module
|
||||
from server import create_app
|
||||
from utils.page_util import PageUtil
|
||||
|
||||
ReportDao = getattr(report_dao_module, 'ReportDao', None)
|
||||
ReportService = getattr(report_service_module, 'ReportService', None)
|
||||
report_controller = getattr(report_controller_module, 'report_controller', None)
|
||||
|
||||
|
||||
def make_user(user_id: int = 7) -> CurrentUserModel:
|
||||
return CurrentUserModel(
|
||||
permissions=[],
|
||||
roles=[],
|
||||
user=UserInfoModel(userId=user_id, userName='admin' if user_id == 1 else f'user-{user_id}'),
|
||||
)
|
||||
|
||||
|
||||
def report_row(model_type: type[BaseModel]) -> dict[str, Any]:
|
||||
now = datetime(2026, 7, 26, 10, 30)
|
||||
rows = {
|
||||
PalletInventoryReportModel: {
|
||||
'pallet_id': 1,
|
||||
'pallet_code': 'P-001',
|
||||
'flow_state': PalletFlowState.FILLED_IN_WAREHOUSE,
|
||||
'warehouse_id': 7,
|
||||
'warehouse_name': '一号仓',
|
||||
'site_id': None,
|
||||
'site_name': None,
|
||||
'line_id': None,
|
||||
'line_name': None,
|
||||
'strain_id': 11,
|
||||
'strain_code': 'S-011',
|
||||
'strain_name': '香菇 11',
|
||||
'batch_no': 'B-202607',
|
||||
'initial_packages': 20,
|
||||
'remaining_packages': 12,
|
||||
'updated_at': now,
|
||||
},
|
||||
FillingReportModel: {
|
||||
'filling_id': 2,
|
||||
'filling_no': 'FL-002',
|
||||
'operated_at': now,
|
||||
'warehouse_id': 7,
|
||||
'warehouse_name': '一号仓',
|
||||
'pallet_id': 1,
|
||||
'pallet_code': 'P-001',
|
||||
'strain_id': 11,
|
||||
'strain_code_snapshot': 'S-011',
|
||||
'strain_name_snapshot': '香菇 11',
|
||||
'batch_no': 'B-202607',
|
||||
'package_spec': '20kg',
|
||||
'initial_packages': 20,
|
||||
'operator_id': 7,
|
||||
'operator_name': 'user-7',
|
||||
},
|
||||
ShipmentReportModel: {
|
||||
'shipment_id': 3,
|
||||
'shipment_no': 'SH-003',
|
||||
'confirmed_at': now,
|
||||
'source_warehouse_id': 7,
|
||||
'source_warehouse_name': '一号仓',
|
||||
'target_site_id': 9,
|
||||
'target_site_name': '一号现场',
|
||||
'customer_id': 3,
|
||||
'customer_name': '一号客户',
|
||||
'pallet_id': 1,
|
||||
'pallet_code': 'P-001',
|
||||
'active_filling_id': 2,
|
||||
'package_qty': 12,
|
||||
},
|
||||
ConsumptionReportModel: {
|
||||
'usage_id': 4,
|
||||
'usage_no': 'US-004',
|
||||
'occurred_at': now,
|
||||
'customer_id': 3,
|
||||
'customer_name': '一号客户',
|
||||
'site_id': 9,
|
||||
'site_name': '一号现场',
|
||||
'line_id': 4,
|
||||
'line_name': '一号线',
|
||||
'pallet_id': 1,
|
||||
'pallet_code': 'P-001',
|
||||
'strain_id': 11,
|
||||
'strain_code': 'S-011',
|
||||
'strain_name': '香菇 11',
|
||||
'batch_no': 'B-202607',
|
||||
'used_packages': 3,
|
||||
'remaining_packages_after': 9,
|
||||
'operator_id': 7,
|
||||
'operator_name': 'user-7',
|
||||
},
|
||||
ReturnReportModel: {
|
||||
'return_id': 5,
|
||||
'return_no': 'RT-005',
|
||||
'status': ReturnStatus.REQUESTED,
|
||||
'requested_at': now,
|
||||
'confirmed_at': None,
|
||||
'source_site_id': 9,
|
||||
'source_site_name': '一号现场',
|
||||
'target_warehouse_id': 7,
|
||||
'target_warehouse_name': '一号仓',
|
||||
'pallet_id': 1,
|
||||
'pallet_code': 'P-001',
|
||||
'active_filling_id_snapshot': 2,
|
||||
'remaining_packages_snapshot': 9,
|
||||
},
|
||||
PalletTraceReportModel: {
|
||||
'event_id': 6,
|
||||
'event_seq': 3,
|
||||
'event_type': PalletEventType.USAGE_RECORDED,
|
||||
'source_type': 'USAGE',
|
||||
'source_id': 4,
|
||||
'pallet_id': 1,
|
||||
'pallet_code': 'P-001',
|
||||
'before_state': PalletFlowState.AT_SITE_UNUSED,
|
||||
'after_state': PalletFlowState.AT_SITE_PARTIAL,
|
||||
'before_warehouse_id': None,
|
||||
'before_warehouse_name': None,
|
||||
'before_site_id': 9,
|
||||
'before_site_name': '一号现场',
|
||||
'before_line_id': None,
|
||||
'before_line_name': None,
|
||||
'after_warehouse_id': None,
|
||||
'after_warehouse_name': None,
|
||||
'after_site_id': 9,
|
||||
'after_site_name': '一号现场',
|
||||
'after_line_id': 4,
|
||||
'after_line_name': '一号线',
|
||||
'before_remaining_packages': 12,
|
||||
'after_remaining_packages': 9,
|
||||
'operator_id': 7,
|
||||
'operator_name': 'user-7',
|
||||
'occurred_at': now,
|
||||
},
|
||||
}
|
||||
return rows[model_type]
|
||||
|
||||
|
||||
class TestReportBootstrap(unittest.TestCase):
|
||||
"""报表三层模块必须能够被项目自动发现。"""
|
||||
|
||||
def test_report_modules_exist(self) -> None:
|
||||
module_names = (
|
||||
'module_business.dao.report_dao',
|
||||
'module_business.service.report_service',
|
||||
'module_business.controller.report_controller',
|
||||
)
|
||||
|
||||
for module_name in module_names:
|
||||
with self.subTest(module_name=module_name):
|
||||
self.assertIsNotNone(importlib.util.find_spec(module_name), f'{module_name} 尚未实现')
|
||||
|
||||
def test_report_controller_router_exists(self) -> None:
|
||||
self.assertIsNotNone(report_controller, 'report_controller 路由尚未实现')
|
||||
|
||||
|
||||
class TestReportPublicContract(unittest.TestCase):
|
||||
def test_dao_exposes_exact_public_methods_and_signatures(self) -> None:
|
||||
report_dao = getattr(report_dao_module, 'ReportDao', None)
|
||||
self.assertIsNotNone(report_dao, 'ReportDao 尚未实现')
|
||||
expected_signatures = {
|
||||
'get_pallet_inventory_list': ('db', 'query_object', 'warehouse_ids', 'site_ids'),
|
||||
'get_filling_report_list': ('db', 'query_object', 'warehouse_ids'),
|
||||
'get_shipment_report_list': ('db', 'query_object', 'warehouse_ids', 'site_ids'),
|
||||
'get_consumption_report_list': ('db', 'query_object', 'site_ids'),
|
||||
'get_return_report_list': ('db', 'query_object', 'site_ids', 'warehouse_ids'),
|
||||
'get_pallet_trace_list': ('db', 'query_object', 'warehouse_ids', 'site_ids'),
|
||||
}
|
||||
|
||||
for method_name, expected_parameters in expected_signatures.items():
|
||||
with self.subTest(method_name=method_name):
|
||||
method = getattr(report_dao, method_name, None)
|
||||
self.assertIsNotNone(method, f'ReportDao.{method_name} 尚未实现')
|
||||
self.assertEqual(expected_parameters, tuple(inspect.signature(method).parameters))
|
||||
|
||||
def test_service_exposes_exact_public_methods_and_signatures(self) -> None:
|
||||
report_service = getattr(report_service_module, 'ReportService', None)
|
||||
self.assertIsNotNone(report_service, 'ReportService 尚未实现')
|
||||
method_names = (
|
||||
'get_pallet_inventory_list_services',
|
||||
'get_filling_report_list_services',
|
||||
'get_shipment_report_list_services',
|
||||
'get_consumption_report_list_services',
|
||||
'get_return_report_list_services',
|
||||
'get_pallet_trace_list_services',
|
||||
)
|
||||
|
||||
for method_name in method_names:
|
||||
with self.subTest(method_name=method_name):
|
||||
method = getattr(report_service, method_name, None)
|
||||
self.assertIsNotNone(method, f'ReportService.{method_name} 尚未实现')
|
||||
self.assertEqual(
|
||||
('query_db', 'query_object', 'current_user'),
|
||||
tuple(inspect.signature(method).parameters),
|
||||
)
|
||||
|
||||
def test_dao_and_service_have_no_write_calls(self) -> None:
|
||||
forbidden_calls = {
|
||||
'add',
|
||||
'add_all',
|
||||
'commit',
|
||||
'delete',
|
||||
'execute_delete',
|
||||
'execute_update',
|
||||
'flush',
|
||||
'rollback',
|
||||
'update',
|
||||
}
|
||||
for module in (report_dao_module, report_service_module):
|
||||
with self.subTest(module=module.__name__):
|
||||
tree = ast.parse(inspect.getsource(module))
|
||||
called_attributes = {
|
||||
node.func.attr
|
||||
for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
|
||||
}
|
||||
self.assertTrue(forbidden_calls.isdisjoint(called_attributes))
|
||||
|
||||
|
||||
@unittest.skipUnless(ReportDao is not None, 'ReportDao 尚未实现')
|
||||
class TestReportDaoQueries(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self) -> None:
|
||||
self.db = MagicMock(spec=AsyncSession)
|
||||
|
||||
async def _capture_statement(
|
||||
self,
|
||||
method: Callable[..., Any],
|
||||
query_object: Any,
|
||||
**scope: Any,
|
||||
) -> Select:
|
||||
page = PageModel(
|
||||
rows=[],
|
||||
pageNum=query_object.page_num,
|
||||
pageSize=query_object.page_size,
|
||||
total=0,
|
||||
hasNext=False,
|
||||
)
|
||||
paginate = AsyncMock(return_value=page)
|
||||
with patch.object(PageUtil, 'paginate', new=paginate):
|
||||
try:
|
||||
result = await method(self.db, query_object, **scope)
|
||||
except NotImplementedError:
|
||||
self.fail(f'{method.__name__} 尚未构造只读分页查询')
|
||||
|
||||
self.assertIs(page, result)
|
||||
paginate.assert_awaited_once()
|
||||
args = paginate.await_args.args
|
||||
self.assertIs(self.db, args[0])
|
||||
self.assertEqual(query_object.page_num, args[2])
|
||||
self.assertEqual(query_object.page_size, args[3])
|
||||
self.assertIs(True, args[4])
|
||||
return args[1]
|
||||
|
||||
def _assert_compiles_for_supported_databases(self, statement: Select) -> tuple[str, str]:
|
||||
compiled = tuple(
|
||||
' '.join(str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).lower().split())
|
||||
for dialect in (mysql.dialect(), postgresql.dialect())
|
||||
)
|
||||
self.assertTrue(all(sql.startswith('select ') for sql in compiled))
|
||||
return compiled
|
||||
|
||||
def _assert_labels(self, statement: Select, row_model: type[BaseModel]) -> None:
|
||||
self.assertEqual(tuple(row_model.model_fields), tuple(column.key for column in statement.selected_columns))
|
||||
|
||||
async def test_inventory_query_uses_current_facts_snapshots_all_filters_and_or_scope(self) -> None:
|
||||
query = PalletInventoryReportQueryModel(
|
||||
pageNum=2,
|
||||
pageSize=7,
|
||||
palletCode='P-001',
|
||||
flowState=PalletFlowState.AT_SITE_PARTIAL,
|
||||
warehouseId=7,
|
||||
siteId=9,
|
||||
lineId=4,
|
||||
strainId=11,
|
||||
batchNo='B-202607',
|
||||
beginTime='2026-07-01',
|
||||
endTime='2026-07-31',
|
||||
)
|
||||
statement = await self._capture_statement(
|
||||
ReportDao.get_pallet_inventory_list,
|
||||
query,
|
||||
warehouse_ids=[8, 7, 7],
|
||||
site_ids=[10, 9, 9],
|
||||
)
|
||||
|
||||
self._assert_labels(statement, PalletInventoryReportModel)
|
||||
for sql in self._assert_compiles_for_supported_databases(statement):
|
||||
self.assertIn('from biz_pallet_inventory', sql)
|
||||
self.assertIn('join biz_pallet ', sql)
|
||||
self.assertIn('left outer join biz_filling_record as active_filling', sql)
|
||||
self.assertIn('active_filling.strain_code_snapshot as strain_code', sql)
|
||||
self.assertIn('active_filling.strain_name_snapshot as strain_name', sql)
|
||||
self.assertNotIn('biz_strain', sql)
|
||||
self.assertIn(
|
||||
'(biz_pallet_inventory.warehouse_id in (7, 8) or biz_pallet_inventory.site_id in (9, 10))',
|
||||
sql,
|
||||
)
|
||||
for column in (
|
||||
'pallet_code',
|
||||
'flow_state',
|
||||
'warehouse_id',
|
||||
'site_id',
|
||||
'current_line_id',
|
||||
'strain_id',
|
||||
'batch_no',
|
||||
):
|
||||
self.assertIn(column, sql)
|
||||
self.assertIn('biz_pallet_inventory.updated_at >=', sql)
|
||||
self.assertIn('biz_pallet_inventory.updated_at <', sql)
|
||||
self.assertIn('order by biz_pallet_inventory.pallet_id', sql)
|
||||
|
||||
async def test_filling_query_uses_snapshot_columns_all_filters_and_warehouse_scope(self) -> None:
|
||||
query = FillingReportQueryModel(
|
||||
pageNum=3,
|
||||
pageSize=5,
|
||||
warehouseId=7,
|
||||
palletCode='P-001',
|
||||
strainId=11,
|
||||
batchNo='B-202607',
|
||||
beginTime='2026-07-01',
|
||||
endTime='2026-07-31',
|
||||
)
|
||||
statement = await self._capture_statement(
|
||||
ReportDao.get_filling_report_list,
|
||||
query,
|
||||
warehouse_ids=[8, 7, 7],
|
||||
)
|
||||
|
||||
self._assert_labels(statement, FillingReportModel)
|
||||
for sql in self._assert_compiles_for_supported_databases(statement):
|
||||
self.assertIn('from biz_filling_record', sql)
|
||||
self.assertIn('join biz_pallet ', sql)
|
||||
self.assertIn('join biz_warehouse ', sql)
|
||||
self.assertIn('strain_code_snapshot', sql)
|
||||
self.assertIn('strain_name_snapshot', sql)
|
||||
self.assertNotIn('biz_strain', sql)
|
||||
self.assertIn('biz_filling_record.warehouse_id in (7, 8)', sql)
|
||||
for column in ('warehouse_id', 'pallet_code', 'strain_id', 'batch_no'):
|
||||
self.assertIn(column, sql)
|
||||
self.assertIn('biz_filling_record.operated_at >=', sql)
|
||||
self.assertIn('biz_filling_record.operated_at <', sql)
|
||||
self.assertIn('order by biz_filling_record.filling_id', sql)
|
||||
|
||||
async def test_shipment_query_is_one_row_per_item_uses_snapshots_and_and_scope(self) -> None:
|
||||
query = ShipmentReportQueryModel(
|
||||
shipmentNo='SH-001',
|
||||
sourceWarehouseId=7,
|
||||
targetSiteId=9,
|
||||
palletCode='P-001',
|
||||
status=DocumentStatus.CONFIRMED,
|
||||
beginTime='2026-07-01',
|
||||
endTime='2026-07-31',
|
||||
)
|
||||
statement = await self._capture_statement(
|
||||
ReportDao.get_shipment_report_list,
|
||||
query,
|
||||
warehouse_ids=[8, 7],
|
||||
site_ids=[10, 9],
|
||||
)
|
||||
|
||||
self._assert_labels(statement, ShipmentReportModel)
|
||||
for sql in self._assert_compiles_for_supported_databases(statement):
|
||||
self.assertIn('from biz_shipment_order', sql)
|
||||
self.assertIn('join biz_shipment_item', sql)
|
||||
self.assertIn('biz_shipment_item.active_filling_id as active_filling_id', sql)
|
||||
self.assertIn('biz_shipment_item.package_qty as package_qty', sql)
|
||||
self.assertIn('biz_shipment_order.source_warehouse_id in (7, 8)', sql)
|
||||
self.assertIn('biz_shipment_order.target_site_id in (9, 10)', sql)
|
||||
self.assertNotIn(
|
||||
'source_warehouse_id in (7, 8) or biz_shipment_order.target_site_id',
|
||||
sql,
|
||||
)
|
||||
for column in ('shipment_no', 'source_warehouse_id', 'target_site_id', 'pallet_code', 'status'):
|
||||
self.assertIn(column, sql)
|
||||
self.assertIn('biz_shipment_order.confirmed_at >=', sql)
|
||||
self.assertIn('biz_shipment_order.confirmed_at <', sql)
|
||||
self.assertIn(
|
||||
'order by biz_shipment_order.shipment_id, biz_shipment_item.shipment_item_id',
|
||||
sql,
|
||||
)
|
||||
|
||||
async def test_consumption_query_uses_usage_filling_snapshots_all_filters_and_site_scope(self) -> None:
|
||||
query = ConsumptionReportQueryModel(
|
||||
customerId=3,
|
||||
siteId=9,
|
||||
lineId=4,
|
||||
palletCode='P-001',
|
||||
strainId=11,
|
||||
batchNo='B-202607',
|
||||
beginTime='2026-07-01',
|
||||
endTime='2026-07-31',
|
||||
)
|
||||
statement = await self._capture_statement(
|
||||
ReportDao.get_consumption_report_list,
|
||||
query,
|
||||
site_ids=[10, 9, 9],
|
||||
)
|
||||
|
||||
self._assert_labels(statement, ConsumptionReportModel)
|
||||
for sql in self._assert_compiles_for_supported_databases(statement):
|
||||
self.assertIn('from biz_usage_record', sql)
|
||||
self.assertIn(
|
||||
'join biz_filling_record on biz_filling_record.filling_id = biz_usage_record.active_filling_id',
|
||||
sql,
|
||||
)
|
||||
self.assertIn('biz_filling_record.strain_code_snapshot as strain_code', sql)
|
||||
self.assertIn('biz_filling_record.strain_name_snapshot as strain_name', sql)
|
||||
self.assertNotIn('biz_strain', sql)
|
||||
self.assertIn('biz_usage_record.site_id in (9, 10)', sql)
|
||||
for column in ('customer_id', 'site_id', 'line_id', 'pallet_code', 'strain_id', 'batch_no'):
|
||||
self.assertIn(column, sql)
|
||||
self.assertIn('biz_usage_record.occurred_at >=', sql)
|
||||
self.assertIn('biz_usage_record.occurred_at <', sql)
|
||||
self.assertIn('order by biz_usage_record.usage_id', sql)
|
||||
|
||||
async def test_return_query_is_one_row_per_item_uses_snapshots_and_or_scope(self) -> None:
|
||||
query = ReturnReportQueryModel(
|
||||
returnNo='RT-001',
|
||||
sourceSiteId=9,
|
||||
targetWarehouseId=7,
|
||||
palletCode='P-001',
|
||||
status=ReturnStatus.REQUESTED,
|
||||
beginTime='2026-07-01',
|
||||
endTime='2026-07-31',
|
||||
)
|
||||
statement = await self._capture_statement(
|
||||
ReportDao.get_return_report_list,
|
||||
query,
|
||||
site_ids=[10, 9],
|
||||
warehouse_ids=[8, 7],
|
||||
)
|
||||
|
||||
self._assert_labels(statement, ReturnReportModel)
|
||||
for sql in self._assert_compiles_for_supported_databases(statement):
|
||||
self.assertIn('from biz_return_order', sql)
|
||||
self.assertIn('join biz_return_item', sql)
|
||||
self.assertIn('biz_return_item.active_filling_id_snapshot as active_filling_id_snapshot', sql)
|
||||
self.assertIn('biz_return_item.remaining_packages_snapshot as remaining_packages_snapshot', sql)
|
||||
self.assertIn(
|
||||
'(biz_return_order.source_site_id in (9, 10) or biz_return_order.target_warehouse_id in (7, 8))',
|
||||
sql,
|
||||
)
|
||||
for column in ('return_no', 'source_site_id', 'target_warehouse_id', 'pallet_code', 'status'):
|
||||
self.assertIn(column, sql)
|
||||
self.assertIn('biz_return_order.requested_at >=', sql)
|
||||
self.assertIn('biz_return_order.requested_at <', sql)
|
||||
self.assertIn('order by biz_return_order.return_id, biz_return_item.return_item_id', sql)
|
||||
|
||||
async def test_trace_query_reads_event_history_aliases_all_filters_and_event_scope(self) -> None:
|
||||
query = PalletTraceReportQueryModel(
|
||||
palletCode='P-001',
|
||||
eventType=PalletEventType.USAGE_RECORDED,
|
||||
beginTime='2026-07-01',
|
||||
endTime='2026-07-31',
|
||||
)
|
||||
statement = await self._capture_statement(
|
||||
ReportDao.get_pallet_trace_list,
|
||||
query,
|
||||
warehouse_ids=[8, 7],
|
||||
site_ids=[10, 9],
|
||||
)
|
||||
|
||||
self._assert_labels(statement, PalletTraceReportModel)
|
||||
for sql in self._assert_compiles_for_supported_databases(statement):
|
||||
self.assertIn('from biz_pallet_event', sql)
|
||||
self.assertNotIn('biz_pallet_inventory', sql)
|
||||
self.assertNotIn('sys_oper_log', sql)
|
||||
for column in (
|
||||
'before_state',
|
||||
'after_state',
|
||||
'before_warehouse_id',
|
||||
'after_warehouse_id',
|
||||
'before_site_id',
|
||||
'after_site_id',
|
||||
'before_line_id',
|
||||
'after_line_id',
|
||||
'before_remaining_packages',
|
||||
'after_remaining_packages',
|
||||
'event_seq',
|
||||
'event_type',
|
||||
'source_type',
|
||||
'source_id',
|
||||
'operator_id',
|
||||
'occurred_at',
|
||||
):
|
||||
self.assertIn(f'biz_pallet_event.{column}', sql)
|
||||
self.assertIn('biz_pallet_event.before_warehouse_id in (7, 8)', sql)
|
||||
self.assertIn('biz_pallet_event.after_warehouse_id in (7, 8)', sql)
|
||||
self.assertIn('biz_pallet_event.before_site_id in (9, 10)', sql)
|
||||
self.assertIn('biz_pallet_event.after_site_id in (9, 10)', sql)
|
||||
self.assertIn('biz_pallet_event.occurred_at >=', sql)
|
||||
self.assertIn('biz_pallet_event.occurred_at <', sql)
|
||||
self.assertIn(
|
||||
'order by biz_pallet_event.pallet_id, biz_pallet_event.event_seq, biz_pallet_event.event_id',
|
||||
sql,
|
||||
)
|
||||
|
||||
|
||||
@unittest.skipUnless(ReportService is not None, 'ReportService 尚未实现')
|
||||
class TestReportServices(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self) -> None:
|
||||
self.db = MagicMock(spec=AsyncSession)
|
||||
self.user = make_user()
|
||||
|
||||
async def _assert_scoped_conversion(
|
||||
self,
|
||||
service_name: str,
|
||||
dao_name: str,
|
||||
query_object: Any,
|
||||
row_model: type[BaseModel],
|
||||
expected_scope: dict[str, list[int] | None],
|
||||
) -> None:
|
||||
page = PageModel(
|
||||
rows=[report_row(row_model)],
|
||||
pageNum=query_object.page_num,
|
||||
pageSize=query_object.page_size,
|
||||
total=17,
|
||||
hasNext=True,
|
||||
)
|
||||
dao_method = AsyncMock(return_value=page)
|
||||
warehouse_scope = AsyncMock(return_value=[8, 7, 7])
|
||||
site_scope = AsyncMock(return_value=[10, 9, 9])
|
||||
with (
|
||||
patch.object(ReportDao, dao_name, new=dao_method),
|
||||
patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope),
|
||||
patch.object(AccessDao, 'list_user_site_ids', new=site_scope),
|
||||
):
|
||||
try:
|
||||
result = await getattr(ReportService, service_name)(self.db, query_object, self.user)
|
||||
except NotImplementedError:
|
||||
self.fail(f'ReportService.{service_name} 尚未实现数据范围和行转换')
|
||||
|
||||
dao_method.assert_awaited_once_with(self.db, query_object, **expected_scope)
|
||||
if 'warehouse_ids' in expected_scope:
|
||||
warehouse_scope.assert_awaited_once_with(self.db, 7)
|
||||
else:
|
||||
warehouse_scope.assert_not_awaited()
|
||||
if 'site_ids' in expected_scope:
|
||||
site_scope.assert_awaited_once_with(self.db, 7)
|
||||
else:
|
||||
site_scope.assert_not_awaited()
|
||||
self.assertEqual(17, result.total)
|
||||
self.assertEqual(query_object.page_num, result.page_num)
|
||||
self.assertEqual(query_object.page_size, result.page_size)
|
||||
self.assertTrue(result.has_next)
|
||||
self.assertIsInstance(result.rows[0], row_model)
|
||||
|
||||
async def test_inventory_service_pushes_both_scopes_and_converts_page(self) -> None:
|
||||
await self._assert_scoped_conversion(
|
||||
'get_pallet_inventory_list_services',
|
||||
'get_pallet_inventory_list',
|
||||
PalletInventoryReportQueryModel(pageNum=2, pageSize=7),
|
||||
PalletInventoryReportModel,
|
||||
{'warehouse_ids': [7, 8], 'site_ids': [9, 10]},
|
||||
)
|
||||
|
||||
async def test_filling_service_pushes_warehouse_scope_and_converts_page(self) -> None:
|
||||
await self._assert_scoped_conversion(
|
||||
'get_filling_report_list_services',
|
||||
'get_filling_report_list',
|
||||
FillingReportQueryModel(pageNum=2, pageSize=7),
|
||||
FillingReportModel,
|
||||
{'warehouse_ids': [7, 8]},
|
||||
)
|
||||
|
||||
async def test_shipment_service_pushes_both_scopes_and_converts_page(self) -> None:
|
||||
await self._assert_scoped_conversion(
|
||||
'get_shipment_report_list_services',
|
||||
'get_shipment_report_list',
|
||||
ShipmentReportQueryModel(pageNum=2, pageSize=7),
|
||||
ShipmentReportModel,
|
||||
{'warehouse_ids': [7, 8], 'site_ids': [9, 10]},
|
||||
)
|
||||
|
||||
async def test_consumption_service_pushes_site_scope_and_converts_page(self) -> None:
|
||||
await self._assert_scoped_conversion(
|
||||
'get_consumption_report_list_services',
|
||||
'get_consumption_report_list',
|
||||
ConsumptionReportQueryModel(pageNum=2, pageSize=7),
|
||||
ConsumptionReportModel,
|
||||
{'site_ids': [9, 10]},
|
||||
)
|
||||
|
||||
async def test_return_service_pushes_both_scopes_and_converts_page(self) -> None:
|
||||
await self._assert_scoped_conversion(
|
||||
'get_return_report_list_services',
|
||||
'get_return_report_list',
|
||||
ReturnReportQueryModel(pageNum=2, pageSize=7),
|
||||
ReturnReportModel,
|
||||
{'site_ids': [9, 10], 'warehouse_ids': [7, 8]},
|
||||
)
|
||||
|
||||
async def test_trace_service_pushes_both_scopes_and_converts_page(self) -> None:
|
||||
await self._assert_scoped_conversion(
|
||||
'get_pallet_trace_list_services',
|
||||
'get_pallet_trace_list',
|
||||
PalletTraceReportQueryModel(pageNum=2, pageSize=7),
|
||||
PalletTraceReportModel,
|
||||
{'warehouse_ids': [7, 8], 'site_ids': [9, 10]},
|
||||
)
|
||||
|
||||
async def test_admin_skips_scope_queries_and_passes_none(self) -> None:
|
||||
cases = (
|
||||
(
|
||||
'get_pallet_inventory_list_services',
|
||||
'get_pallet_inventory_list',
|
||||
PalletInventoryReportQueryModel(),
|
||||
{'warehouse_ids': None, 'site_ids': None},
|
||||
),
|
||||
(
|
||||
'get_filling_report_list_services',
|
||||
'get_filling_report_list',
|
||||
FillingReportQueryModel(),
|
||||
{'warehouse_ids': None},
|
||||
),
|
||||
(
|
||||
'get_shipment_report_list_services',
|
||||
'get_shipment_report_list',
|
||||
ShipmentReportQueryModel(),
|
||||
{'warehouse_ids': None, 'site_ids': None},
|
||||
),
|
||||
(
|
||||
'get_consumption_report_list_services',
|
||||
'get_consumption_report_list',
|
||||
ConsumptionReportQueryModel(),
|
||||
{'site_ids': None},
|
||||
),
|
||||
(
|
||||
'get_return_report_list_services',
|
||||
'get_return_report_list',
|
||||
ReturnReportQueryModel(),
|
||||
{'site_ids': None, 'warehouse_ids': None},
|
||||
),
|
||||
(
|
||||
'get_pallet_trace_list_services',
|
||||
'get_pallet_trace_list',
|
||||
PalletTraceReportQueryModel(),
|
||||
{'warehouse_ids': None, 'site_ids': None},
|
||||
),
|
||||
)
|
||||
for service_name, dao_name, query_object, expected_scope in cases:
|
||||
with self.subTest(service_name=service_name):
|
||||
dao_method = AsyncMock(
|
||||
return_value=PageModel(
|
||||
rows=[],
|
||||
pageNum=1,
|
||||
pageSize=10,
|
||||
total=0,
|
||||
hasNext=False,
|
||||
)
|
||||
)
|
||||
warehouse_scope = AsyncMock()
|
||||
site_scope = AsyncMock()
|
||||
with (
|
||||
patch.object(ReportDao, dao_name, new=dao_method),
|
||||
patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope),
|
||||
patch.object(AccessDao, 'list_user_site_ids', new=site_scope),
|
||||
):
|
||||
try:
|
||||
await getattr(ReportService, service_name)(self.db, query_object, make_user(1))
|
||||
except NotImplementedError:
|
||||
self.fail(f'ReportService.{service_name} 尚未实现管理员范围语义')
|
||||
|
||||
warehouse_scope.assert_not_awaited()
|
||||
site_scope.assert_not_awaited()
|
||||
dao_method.assert_awaited_once_with(self.db, query_object, **expected_scope)
|
||||
|
||||
async def _assert_empty_page_without_dao(
|
||||
self,
|
||||
service_name: str,
|
||||
dao_name: str,
|
||||
query_object: Any,
|
||||
warehouse_ids: list[int] | None,
|
||||
site_ids: list[int] | None,
|
||||
) -> None:
|
||||
dao_method = AsyncMock()
|
||||
with (
|
||||
patch.object(ReportDao, dao_name, new=dao_method),
|
||||
patch.object(
|
||||
AccessDao,
|
||||
'list_user_warehouse_ids',
|
||||
new=AsyncMock(return_value=warehouse_ids),
|
||||
),
|
||||
patch.object(
|
||||
AccessDao,
|
||||
'list_user_site_ids',
|
||||
new=AsyncMock(return_value=site_ids),
|
||||
),
|
||||
):
|
||||
try:
|
||||
result = await getattr(ReportService, service_name)(self.db, query_object, self.user)
|
||||
except NotImplementedError:
|
||||
self.fail(f'ReportService.{service_name} 尚未实现空范围短路')
|
||||
|
||||
dao_method.assert_not_awaited()
|
||||
self.assertEqual([], result.rows)
|
||||
self.assertEqual(0, result.total)
|
||||
self.assertEqual(query_object.page_num, result.page_num)
|
||||
self.assertEqual(query_object.page_size, result.page_size)
|
||||
self.assertFalse(result.has_next)
|
||||
|
||||
async def test_inventory_return_and_trace_only_short_circuit_when_both_scopes_empty(self) -> None:
|
||||
empty_cases = (
|
||||
(
|
||||
'get_pallet_inventory_list_services',
|
||||
'get_pallet_inventory_list',
|
||||
PalletInventoryReportQueryModel(pageNum=3, pageSize=4),
|
||||
),
|
||||
(
|
||||
'get_return_report_list_services',
|
||||
'get_return_report_list',
|
||||
ReturnReportQueryModel(pageNum=3, pageSize=4),
|
||||
),
|
||||
(
|
||||
'get_pallet_trace_list_services',
|
||||
'get_pallet_trace_list',
|
||||
PalletTraceReportQueryModel(pageNum=3, pageSize=4),
|
||||
),
|
||||
)
|
||||
for service_name, dao_name, query_object in empty_cases:
|
||||
with self.subTest(service_name=service_name):
|
||||
await self._assert_empty_page_without_dao(
|
||||
service_name,
|
||||
dao_name,
|
||||
query_object,
|
||||
[],
|
||||
[],
|
||||
)
|
||||
|
||||
visible_cases = (
|
||||
(
|
||||
'get_pallet_inventory_list_services',
|
||||
'get_pallet_inventory_list',
|
||||
PalletInventoryReportQueryModel(),
|
||||
),
|
||||
('get_return_report_list_services', 'get_return_report_list', ReturnReportQueryModel()),
|
||||
(
|
||||
'get_pallet_trace_list_services',
|
||||
'get_pallet_trace_list',
|
||||
PalletTraceReportQueryModel(),
|
||||
),
|
||||
)
|
||||
for service_name, dao_name, query_object in visible_cases:
|
||||
with self.subTest(service_name=service_name, one_scope_visible=True):
|
||||
dao_method = AsyncMock(return_value=PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False))
|
||||
with (
|
||||
patch.object(ReportDao, dao_name, new=dao_method),
|
||||
patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])),
|
||||
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])),
|
||||
):
|
||||
try:
|
||||
await getattr(ReportService, service_name)(self.db, query_object, self.user)
|
||||
except NotImplementedError:
|
||||
self.fail(f'ReportService.{service_name} 尚未实现单侧 OR 范围可见性')
|
||||
dao_method.assert_awaited_once()
|
||||
|
||||
async def test_filling_and_consumption_empty_applicable_scope_short_circuits(self) -> None:
|
||||
await self._assert_empty_page_without_dao(
|
||||
'get_filling_report_list_services',
|
||||
'get_filling_report_list',
|
||||
FillingReportQueryModel(pageNum=3, pageSize=4),
|
||||
[],
|
||||
[9],
|
||||
)
|
||||
await self._assert_empty_page_without_dao(
|
||||
'get_consumption_report_list_services',
|
||||
'get_consumption_report_list',
|
||||
ConsumptionReportQueryModel(pageNum=3, pageSize=4),
|
||||
[7],
|
||||
[],
|
||||
)
|
||||
|
||||
async def test_shipment_either_empty_scope_short_circuits(self) -> None:
|
||||
for warehouse_ids, site_ids in (([], [9]), ([7], []), ([], [])):
|
||||
with self.subTest(warehouse_ids=warehouse_ids, site_ids=site_ids):
|
||||
await self._assert_empty_page_without_dao(
|
||||
'get_shipment_report_list_services',
|
||||
'get_shipment_report_list',
|
||||
ShipmentReportQueryModel(pageNum=3, pageSize=4),
|
||||
warehouse_ids,
|
||||
site_ids,
|
||||
)
|
||||
|
||||
|
||||
@unittest.skipUnless(report_controller is not None, 'report_controller 尚未实现')
|
||||
class TestReportControllerContract(unittest.TestCase):
|
||||
@staticmethod
|
||||
def _routes() -> dict[tuple[str, str], APIRoute]:
|
||||
return {
|
||||
(route.path, next(iter(route.methods))): route
|
||||
for route in report_controller.routes
|
||||
if isinstance(route, APIRoute)
|
||||
}
|
||||
|
||||
def test_create_app_openapi_has_exact_six_report_get_operations_and_aliases(self) -> None:
|
||||
schema = create_app().openapi()
|
||||
report_operations = {
|
||||
(path, method)
|
||||
for path, methods in schema['paths'].items()
|
||||
if path.startswith('/report/')
|
||||
for method in methods
|
||||
}
|
||||
expected_operations = {
|
||||
('/report/pallet-inventory/list', 'get'),
|
||||
('/report/filling/list', 'get'),
|
||||
('/report/shipment/list', 'get'),
|
||||
('/report/consumption/list', 'get'),
|
||||
('/report/return/list', 'get'),
|
||||
('/report/pallet-trace/list', 'get'),
|
||||
}
|
||||
self.assertEqual(expected_operations, report_operations)
|
||||
|
||||
expected_parameters = {
|
||||
'/report/pallet-inventory/list': {
|
||||
'pageNum',
|
||||
'pageSize',
|
||||
'beginTime',
|
||||
'endTime',
|
||||
'palletCode',
|
||||
'flowState',
|
||||
'warehouseId',
|
||||
'siteId',
|
||||
'lineId',
|
||||
'strainId',
|
||||
'batchNo',
|
||||
},
|
||||
'/report/filling/list': {
|
||||
'pageNum',
|
||||
'pageSize',
|
||||
'beginTime',
|
||||
'endTime',
|
||||
'warehouseId',
|
||||
'palletCode',
|
||||
'strainId',
|
||||
'batchNo',
|
||||
},
|
||||
'/report/shipment/list': {
|
||||
'pageNum',
|
||||
'pageSize',
|
||||
'beginTime',
|
||||
'endTime',
|
||||
'shipmentNo',
|
||||
'sourceWarehouseId',
|
||||
'targetSiteId',
|
||||
'palletCode',
|
||||
'status',
|
||||
},
|
||||
'/report/consumption/list': {
|
||||
'pageNum',
|
||||
'pageSize',
|
||||
'beginTime',
|
||||
'endTime',
|
||||
'customerId',
|
||||
'siteId',
|
||||
'lineId',
|
||||
'palletCode',
|
||||
'strainId',
|
||||
'batchNo',
|
||||
},
|
||||
'/report/return/list': {
|
||||
'pageNum',
|
||||
'pageSize',
|
||||
'beginTime',
|
||||
'endTime',
|
||||
'returnNo',
|
||||
'sourceSiteId',
|
||||
'targetWarehouseId',
|
||||
'palletCode',
|
||||
'status',
|
||||
},
|
||||
'/report/pallet-trace/list': {
|
||||
'pageNum',
|
||||
'pageSize',
|
||||
'beginTime',
|
||||
'endTime',
|
||||
'palletCode',
|
||||
'eventType',
|
||||
},
|
||||
}
|
||||
for path, parameters in expected_parameters.items():
|
||||
with self.subTest(path=path):
|
||||
actual = {parameter['name'] for parameter in schema['paths'][path]['get']['parameters']}
|
||||
self.assertEqual(parameters, actual)
|
||||
self.assertNotIn('mode', actual)
|
||||
|
||||
def test_routes_have_exact_permissions_response_models_current_user_and_no_log(self) -> None:
|
||||
expected = {
|
||||
('/report/pallet-inventory/list', 'GET'): (
|
||||
'report:palletinventory:list',
|
||||
PageResponseModel[PalletInventoryReportModel],
|
||||
'get_pallet_inventory_list_services',
|
||||
),
|
||||
('/report/filling/list', 'GET'): (
|
||||
'report:filling:list',
|
||||
PageResponseModel[FillingReportModel],
|
||||
'get_filling_report_list_services',
|
||||
),
|
||||
('/report/shipment/list', 'GET'): (
|
||||
'report:shipment:list',
|
||||
PageResponseModel[ShipmentReportModel],
|
||||
'get_shipment_report_list_services',
|
||||
),
|
||||
('/report/consumption/list', 'GET'): (
|
||||
'report:consumption:list',
|
||||
PageResponseModel[ConsumptionReportModel],
|
||||
'get_consumption_report_list_services',
|
||||
),
|
||||
('/report/return/list', 'GET'): (
|
||||
'report:return:list',
|
||||
PageResponseModel[ReturnReportModel],
|
||||
'get_return_report_list_services',
|
||||
),
|
||||
('/report/pallet-trace/list', 'GET'): (
|
||||
'report:pallettrace:list',
|
||||
PageResponseModel[PalletTraceReportModel],
|
||||
'get_pallet_trace_list_services',
|
||||
),
|
||||
}
|
||||
routes = self._routes()
|
||||
self.assertEqual(set(expected), set(routes))
|
||||
for key, (permission, response_model, service_method) in expected.items():
|
||||
route = routes[key]
|
||||
permissions = [
|
||||
dependency.dependency
|
||||
for dependency in route.dependencies
|
||||
if isinstance(dependency.dependency, CheckUserInterfaceAuth)
|
||||
]
|
||||
self.assertEqual(1, len(permissions), key)
|
||||
self.assertEqual(permission, permissions[0].perm)
|
||||
self.assertEqual(response_model, route.response_model)
|
||||
self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies))
|
||||
self.assertTrue(
|
||||
any(
|
||||
dependency.name == 'current_user' and dependency.call == LoginService.get_current_user
|
||||
for dependency in route.dependant.dependencies
|
||||
),
|
||||
key,
|
||||
)
|
||||
self.assertFalse(hasattr(route.endpoint, '__wrapped__'), key)
|
||||
self.assertIn(service_method, inspect.getsource(route.endpoint))
|
||||
|
||||
def test_controller_source_has_no_export_write_route_or_log(self) -> None:
|
||||
source = inspect.getsource(report_controller_module)
|
||||
self.assertNotIn('@Log(', source)
|
||||
self.assertNotIn('/export', source)
|
||||
for decorator in ('.post(', '.put(', '.delete(', '.patch('):
|
||||
self.assertNotIn(decorator, source)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user