245 lines
14 KiB
Python
245 lines
14 KiB
Python
"""标签只读回归：执行真实服务/DAO，以记录型会话替代数据库连接。"""
|
|
|
|
import base64
|
|
import importlib
|
|
import importlib.util
|
|
import json
|
|
import unittest
|
|
from datetime import date
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from fastapi import FastAPI, Request
|
|
from sqlalchemy.dialects import mysql
|
|
|
|
from common.aspect.interface_auth import CheckUserInterfaceAuth
|
|
from exceptions.exception import PermissionException, ServiceException
|
|
from module_business.controller import shipment_controller as shipment_controller_module
|
|
from module_business.dependency.test_batch_dependency import TestBatchScopeResolver
|
|
from module_business.domain.test_batch_scope import TestBatchScope
|
|
from module_business.entity.do.master_do import BizCustomer, BizCustomerSite, BizPallet
|
|
from module_business.entity.do.operation_do import BizFillingRecord, BizShipmentItem, BizShipmentOrder
|
|
from module_business.service.pallet_service import PalletService
|
|
from tests.module_business.test_batch_fixtures import make_admin, make_current_user, make_test_scope
|
|
|
|
LabelRecords = tuple[BizShipmentOrder, BizShipmentItem, BizPallet, BizFillingRecord, BizCustomerSite, BizCustomer]
|
|
|
|
|
|
def sql(statement: Any) -> str:
|
|
return ' '.join(str(statement.compile(dialect=mysql.dialect(), compile_kwargs={'literal_binds': True})).split())
|
|
|
|
|
|
class ReadOnlySession:
|
|
"""仅支持 SELECT，不提供写入、提交或真实连接能力。"""
|
|
|
|
def __init__(self, *results: Any) -> None:
|
|
self.results = list(results)
|
|
self.statements = []
|
|
|
|
async def execute(self, statement: Any) -> MagicMock:
|
|
assert statement.is_select, '标签不得产生业务写入'
|
|
self.statements.append(sql(statement))
|
|
assert self.results, '出现非预期查询'
|
|
value = self.results.pop(0)
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = value
|
|
result.scalars.return_value.all.return_value = value
|
|
result.all.return_value = value
|
|
return result
|
|
|
|
|
|
class ShipmentLabelsTest(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
spec = importlib.util.find_spec('module_business.service.shipment_label_service')
|
|
self.assertIsNotNone(spec, '缺少可执行的发货标签服务')
|
|
self.service = importlib.import_module('module_business.service.shipment_label_service').ShipmentLabelService
|
|
|
|
def records(self, status: str = 'CONFIRMED', batch_id: int | None = None, dates: bool = True) -> LabelRecords:
|
|
order = BizShipmentOrder(shipment_id=12, shipment_no='原发货-12', source_warehouse_id=7,
|
|
target_site_id=9, status=status, test_batch_id=batch_id)
|
|
item = BizShipmentItem(shipment_item_id=1, shipment_id=12, pallet_id=5,
|
|
active_filling_id=105, package_qty=37, test_batch_id=batch_id)
|
|
pallet = BizPallet(pallet_id=5, pallet_code='原托盘-005', qr_token='fixed-qr-005',
|
|
asset_status='0', capacity_packages=100, test_batch_id=batch_id)
|
|
filling = BizFillingRecord(filling_id=105, pallet_id=5, strain_id=3,
|
|
strain_name_snapshot='<img src=x onerror=alert(1)>',
|
|
strain_code_snapshot='S03', batch_no='原生产批号-03',
|
|
active_slot='ACTIVE' if status == 'DRAFT' else None,
|
|
production_date=date(2026, 9, 1) if dates else None,
|
|
expiry_date=date(2027, 9, 1) if dates else None,
|
|
test_batch_id=batch_id)
|
|
site = BizCustomerSite(site_id=9, customer_id=4, site_name='原现场', test_batch_id=batch_id)
|
|
customer = BizCustomer(customer_id=4, customer_name='客户 <script>攻击</script>', test_batch_id=batch_id)
|
|
order.items = [item]
|
|
return order, item, pallet, filling, site, customer
|
|
|
|
def session(self, records: LabelRecords, *, admin: bool = True, qty: int | None = None) -> ReadOnlySession:
|
|
order, item, pallet, filling, site, customer = records
|
|
current_qty = 62 if qty is None else qty
|
|
detail_items = [(item, current_qty)] if order.status == 'DRAFT' else [item]
|
|
rows = [(item, pallet, filling, current_qty if order.status == 'DRAFT' else item.package_qty)]
|
|
values = [order, detail_items, site, customer, rows]
|
|
return ReadOnlySession(*([] if admin else [[7], [9]]), *values)
|
|
|
|
async def test_confirmed_uses_original_filling_and_package_snapshot_without_inventory(self) -> None:
|
|
records = self.records()
|
|
db = self.session(records)
|
|
labels = await self.service.get_labels_services(db, 12, make_admin(), TestBatchScope.formal())
|
|
self.assertEqual(1, len(labels))
|
|
payload = labels[0].model_dump(mode='json', by_alias=True)
|
|
self.assertEqual('原托盘-005', payload['palletCode'])
|
|
self.assertEqual('原发货-12', payload['shipmentNo'])
|
|
self.assertEqual('原生产批号-03', payload['batchNo'])
|
|
self.assertEqual('客户 <script>攻击</script>', payload['customerName'])
|
|
self.assertEqual('原现场', payload['siteName'])
|
|
self.assertEqual('<img src=x onerror=alert(1)>', payload['strainName'])
|
|
self.assertEqual(37, payload['packageQty'])
|
|
self.assertEqual('2026-09-01', payload['productionDate'])
|
|
self.assertEqual('2027-09-01', payload['expiryDate'])
|
|
prefix, encoded = payload['qrCodeDataUrl'].split(',', 1)
|
|
self.assertEqual('data:image/png;base64', prefix)
|
|
self.assertEqual(PalletService.render_qr_png('fixed-qr-005'), base64.b64decode(encoded))
|
|
self.assertNotIn('biz_pallet_inventory', ' '.join(db.statements))
|
|
self.assertIn('biz_shipment_item.active_filling_id = biz_filling_record.filling_id', db.statements[-1])
|
|
self.assertNotIn("active_slot = 'ACTIVE'", db.statements[-1])
|
|
self.assertEqual([], db.results)
|
|
|
|
async def test_draft_uses_current_quantity_and_active_filling_with_null_dates(self) -> None:
|
|
records = self.records('DRAFT', dates=False)
|
|
records[1].active_filling_id = None
|
|
records[1].package_qty = None
|
|
db = self.session(records)
|
|
labels = await self.service.get_labels_services(db, 12, make_admin(), TestBatchScope.formal())
|
|
self.assertEqual(62, labels[0].package_qty)
|
|
self.assertIsNone(labels[0].production_date)
|
|
self.assertIsNone(labels[0].expiry_date)
|
|
statement = db.statements[-1]
|
|
for fragment in (
|
|
'biz_pallet_inventory.active_filling_id = biz_filling_record.filling_id',
|
|
'biz_pallet_inventory.warehouse_id = 7',
|
|
"biz_pallet_inventory.flow_state = 'FILLED_IN_WAREHOUSE'",
|
|
'biz_pallet_inventory.site_id IS NULL',
|
|
'biz_pallet_inventory.current_line_id IS NULL',
|
|
'biz_pallet_inventory.remaining_packages > 0',
|
|
"biz_filling_record.active_slot = 'ACTIVE'",
|
|
):
|
|
self.assertIn(fragment, statement)
|
|
|
|
async def test_both_business_scopes_and_each_related_batch_are_in_real_queries(self) -> None:
|
|
for scope in (TestBatchScope.formal(), make_test_scope(31)):
|
|
for status in ('DRAFT', 'CONFIRMED'):
|
|
with self.subTest(scope=scope, status=status):
|
|
db = self.session(self.records(status, scope.batch_id), admin=False)
|
|
await self.service.get_labels_services(db, 12, make_current_user(), scope)
|
|
self.assertIn('source_warehouse_id IN (7)', db.statements[2])
|
|
self.assertIn('target_site_id IN (9)', db.statements[2])
|
|
combined = ' '.join(db.statements)
|
|
tables = ['biz_shipment_order', 'biz_shipment_item', 'biz_pallet',
|
|
'biz_filling_record', 'biz_customer_site', 'biz_customer']
|
|
if status == 'DRAFT':
|
|
tables.append('biz_pallet_inventory')
|
|
for table in tables:
|
|
predicate = 'IS NULL' if scope.batch_id is None else '= 31'
|
|
self.assertIn(f'{table}.test_batch_id {predicate}', combined)
|
|
self.assertNotIn('FOR UPDATE', combined)
|
|
|
|
async def test_inaccessible_order_stops_before_related_queries_or_qr_rendering(self) -> None:
|
|
for warehouses, sites in (([], [9]), ([7], []), ([8], [10])):
|
|
with self.subTest(warehouses=warehouses, sites=sites):
|
|
db = ReadOnlySession(warehouses, sites, None)
|
|
with (
|
|
patch.object(PalletService, 'render_qr_png', side_effect=AssertionError('未授权生成二维码')),
|
|
self.assertRaises(PermissionException),
|
|
):
|
|
await self.service.get_labels_services(db, 12, make_current_user(), TestBatchScope.formal())
|
|
self.assertEqual(3, len(db.statements))
|
|
|
|
async def test_cross_batch_relations_are_rejected_before_qr_generation(self) -> None:
|
|
for index in range(6):
|
|
with self.subTest(record=index):
|
|
records = self.records()
|
|
records[index].test_batch_id = 31
|
|
db = self.session(records)
|
|
with (
|
|
patch.object(PalletService, 'render_qr_png', side_effect=AssertionError('跨批次生成二维码')),
|
|
self.assertRaises(ServiceException),
|
|
):
|
|
await self.service.get_labels_services(db, 12, make_admin(), TestBatchScope.formal())
|
|
|
|
async def test_incomplete_label_does_not_print_partial_or_fall_back_to_current_inventory(self) -> None:
|
|
for status in ('DRAFT', 'CONFIRMED'):
|
|
for missing in ('pallet', 'filling', 'quantity', 'snapshot'):
|
|
with self.subTest(status=status, missing=missing):
|
|
records = self.records(status)
|
|
if missing == 'snapshot':
|
|
records[3].pallet_id = 99
|
|
db = self.session(records)
|
|
row = list(db.results[-1][0])
|
|
if missing != 'snapshot':
|
|
row[{'pallet': 1, 'filling': 2, 'quantity': 3}[missing]] = None
|
|
db.results[-1] = [tuple(row)]
|
|
with self.assertRaises(ServiceException):
|
|
await self.service.get_labels_services(db, 12, make_admin(), TestBatchScope.formal())
|
|
|
|
async def test_missing_later_pallet_prevents_any_qr_generation(self) -> None:
|
|
records = self.records()
|
|
db = self.session(records)
|
|
second = BizShipmentItem(shipment_item_id=2, shipment_id=12, pallet_id=6,
|
|
active_filling_id=106, package_qty=20, test_batch_id=None)
|
|
db.results[1].append(second)
|
|
db.results[-1].append((second, None, None, 20))
|
|
with (
|
|
patch.object(PalletService, 'render_qr_png', side_effect=AssertionError('未完成校验已生成二维码')),
|
|
self.assertRaises(ServiceException),
|
|
):
|
|
await self.service.get_labels_services(db, 12, make_admin(), TestBatchScope.formal())
|
|
|
|
async def test_empty_shipment_is_not_printable(self) -> None:
|
|
records = self.records()
|
|
db = self.session(records)
|
|
db.results[1] = []
|
|
db.results[-1] = []
|
|
with self.assertRaises(ServiceException):
|
|
await self.service.get_labels_services(db, 12, make_admin(), TestBatchScope.formal())
|
|
|
|
async def test_route_serializes_label_dates_and_prevents_cross_scope_browser_cache(self) -> None:
|
|
db = self.session(self.records())
|
|
response = await shipment_controller_module.get_business_shipment_labels(
|
|
Request({'type': 'http'}), 12, db, make_admin(), TestBatchScope.formal(),
|
|
)
|
|
payload = json.loads(response.body)
|
|
self.assertEqual(200, response.status_code)
|
|
self.assertEqual('2026-09-01', payload['data'][0]['productionDate'])
|
|
self.assertEqual('CONFIRMED', payload['data'][0]['shipmentStatus'])
|
|
self.assertTrue(payload['data'][0]['qrCodeDataUrl'].startswith('data:image/png;base64,'))
|
|
self.assertEqual('no-store', response.headers.get('Cache-Control'))
|
|
|
|
async def test_draft_rejects_disabled_closed_or_excess_quantity(self) -> None:
|
|
for field in ('asset_status', 'active_slot', 'capacity_packages'):
|
|
with self.subTest(field=field):
|
|
records = self.records('DRAFT')
|
|
target = records[3] if field == 'active_slot' else records[2]
|
|
setattr(target, field, {'asset_status': '1', 'active_slot': None, 'capacity_packages': 20}[field])
|
|
with self.assertRaises(ServiceException):
|
|
await self.service.get_labels_services(
|
|
self.session(records), 12, make_admin(), TestBatchScope.formal(),
|
|
)
|
|
|
|
|
|
class ShipmentLabelRouteTest(unittest.TestCase):
|
|
def test_label_route_reuses_query_permission_and_read_scope(self) -> None:
|
|
app = FastAPI()
|
|
app.include_router(shipment_controller_module.shipment_controller)
|
|
routes = [route for route in app.routes if route.path == '/business/shipment/{shipment_id}/labels']
|
|
self.assertEqual(1, len(routes), '缺少标签 GET 路由')
|
|
route = routes[0]
|
|
self.assertEqual({'GET'}, route.methods)
|
|
checks = [dep.call for dep in route.dependant.dependencies if isinstance(dep.call, CheckUserInterfaceAuth)]
|
|
self.assertEqual(['business:shipment:query'], [check.perm for check in checks])
|
|
scopes = [dep.call for dep in route.dependant.dependencies if isinstance(dep.call, TestBatchScopeResolver)]
|
|
self.assertEqual([False], [scope.write for scope in scopes])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|