1069 lines
44 KiB
Python
1069 lines
44 KiB
Python
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, ValidationError
|
|
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)
|
|
|
|
REPORT_QUERY_MODELS = (
|
|
PalletInventoryReportQueryModel,
|
|
FillingReportQueryModel,
|
|
ShipmentReportQueryModel,
|
|
ConsumptionReportQueryModel,
|
|
ReturnReportQueryModel,
|
|
PalletTraceReportQueryModel,
|
|
)
|
|
|
|
|
|
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))
|
|
|
|
|
|
class TestReportQueryValidation(unittest.TestCase):
|
|
def test_dates_require_strict_real_calendar_values(self) -> None:
|
|
for model_class in REPORT_QUERY_MODELS:
|
|
with self.subTest(model=model_class.__name__, case='begin-only'):
|
|
model = model_class.model_validate({'beginTime': '2026-07-01'})
|
|
self.assertEqual('2026-07-01', model.begin_time)
|
|
self.assertIsNone(model.end_time)
|
|
with self.subTest(model=model_class.__name__, case='end-only'):
|
|
model = model_class.model_validate({'endTime': '2026-07-31'})
|
|
self.assertIsNone(model.begin_time)
|
|
self.assertEqual('2026-07-31', model.end_time)
|
|
for field_name in ('beginTime', 'endTime'):
|
|
for invalid_date in ('not-a-date', '2026-02-30', '2026-7-01', '2026-07-1'):
|
|
with (
|
|
self.subTest(
|
|
model=model_class.__name__,
|
|
field=field_name,
|
|
value=invalid_date,
|
|
),
|
|
self.assertRaises(ValidationError),
|
|
):
|
|
model_class.model_validate({field_name: invalid_date})
|
|
|
|
def test_date_range_allows_none_and_equal_but_rejects_inverted_or_overflowing_end(self) -> None:
|
|
for model_class in REPORT_QUERY_MODELS:
|
|
with self.subTest(model=model_class.__name__, case='none'):
|
|
model = model_class.model_validate({'beginTime': None, 'endTime': None})
|
|
self.assertIsNone(model.begin_time)
|
|
self.assertIsNone(model.end_time)
|
|
with self.subTest(model=model_class.__name__, case='equal'):
|
|
model = model_class.model_validate({'beginTime': '2026-07-02', 'endTime': '2026-07-02'})
|
|
self.assertEqual(model.begin_time, model.end_time)
|
|
with self.subTest(model=model_class.__name__, case='max-begin'):
|
|
model = model_class.model_validate({'beginTime': '9999-12-31'})
|
|
self.assertEqual('9999-12-31', model.begin_time)
|
|
with self.subTest(model=model_class.__name__, case='inverted'), self.assertRaises(ValidationError):
|
|
model_class.model_validate({'beginTime': '2026-07-03', 'endTime': '2026-07-02'})
|
|
with self.subTest(model=model_class.__name__, case='max-end'), self.assertRaises(ValidationError):
|
|
model_class.model_validate({'endTime': '9999-12-31'})
|
|
|
|
def test_page_size_is_capped_at_100(self) -> None:
|
|
for model_class in REPORT_QUERY_MODELS:
|
|
with self.subTest(model=model_class.__name__, page_size=100):
|
|
self.assertEqual(100, model_class.model_validate({'pageSize': 100}).page_size)
|
|
for invalid_size in (101, 10**9):
|
|
with (
|
|
self.subTest(model=model_class.__name__, page_size=invalid_size),
|
|
self.assertRaises(ValidationError),
|
|
):
|
|
model_class.model_validate({'pageSize': invalid_size})
|
|
|
|
|
|
@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_shipment_query_requires_confirmed_fact_in_addition_to_optional_status(self) -> None:
|
|
statement = await self._capture_statement(
|
|
ReportDao.get_shipment_report_list,
|
|
ShipmentReportQueryModel(status=DocumentStatus.DRAFT),
|
|
warehouse_ids=None,
|
|
site_ids=None,
|
|
)
|
|
|
|
for sql in self._assert_compiles_for_supported_databases(statement):
|
|
self.assertIn("biz_shipment_order.status = 'draft'", sql)
|
|
self.assertIn("biz_shipment_order.status = 'confirmed'", sql)
|
|
self.assertEqual(2, sql.count('biz_shipment_order.status ='))
|
|
|
|
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):
|
|
operation_parameters = schema['paths'][path]['get']['parameters']
|
|
actual = {parameter['name'] for parameter in operation_parameters}
|
|
self.assertEqual(parameters, actual)
|
|
self.assertNotIn('mode', actual)
|
|
page_size_parameter = next(
|
|
parameter for parameter in operation_parameters if parameter['name'] == 'pageSize'
|
|
)
|
|
self.assertEqual(100, page_size_parameter['schema'].get('maximum'))
|
|
|
|
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()
|