Files
ShiErFeng/shierfeng-fastapi-backend/tests/module_business/test_vo_contract.py
T

786 lines
22 KiB
Python

import unittest
from pydantic import ValidationError
from pydantic.alias_generators import to_camel
from module_business.entity.vo.master_vo import (
CustomerModel,
CustomerPageQueryModel,
DeleteCustomerModel,
DeleteLineModel,
DeletePalletModel,
DeleteSiteModel,
DeleteStrainModel,
DeleteWarehouseModel,
LineModel,
LinePageQueryModel,
PalletCreateModel,
PalletEditModel,
PalletModel,
PalletPageQueryModel,
SiteModel,
SitePageQueryModel,
StrainModel,
StrainPageQueryModel,
UserAssignModel,
WarehouseModel,
WarehousePageQueryModel,
)
from module_business.entity.vo.operation_vo import (
BusinessOptionModel,
ConfirmCommandModel,
FillingCreateModel,
FillingModel,
FillingPageQueryModel,
InboundCreateModel,
InboundItemModel,
InboundModel,
InboundPageQueryModel,
InboundScanQueryModel,
InboundUpdateModel,
PalletScanResponseModel,
ReturnConfirmModel,
ReturnItemModel,
ReturnModel,
ReturnPageQueryModel,
ReturnRequestModel,
ReturnScanQueryModel,
ShipmentCreateModel,
ShipmentItemModel,
ShipmentModel,
ShipmentPageQueryModel,
ShipmentScanQueryModel,
ShipmentUpdateModel,
UsageCreateModel,
UsageModel,
UsagePageQueryModel,
)
from module_business.entity.vo.report_vo import (
ConsumptionReportModel,
ConsumptionReportQueryModel,
FillingReportModel,
FillingReportQueryModel,
PalletInventoryReportModel,
PalletInventoryReportQueryModel,
PalletTraceReportModel,
PalletTraceReportQueryModel,
ReturnReportModel,
ReturnReportQueryModel,
ShipmentReportModel,
ShipmentReportQueryModel,
)
MASTER_MODEL_FIELDS = {
StrainModel: {
'strain_id',
'strain_code',
'strain_name',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
StrainPageQueryModel: {'page_num', 'page_size', 'strain_code', 'strain_name', 'status'},
DeleteStrainModel: {'strain_ids'},
CustomerModel: {
'customer_id',
'customer_code',
'customer_name',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
CustomerPageQueryModel: {'page_num', 'page_size', 'customer_code', 'customer_name', 'status'},
DeleteCustomerModel: {'customer_ids'},
SiteModel: {
'site_id',
'customer_id',
'site_code',
'site_name',
'address',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
SitePageQueryModel: {'page_num', 'page_size', 'customer_id', 'site_code', 'site_name', 'status'},
DeleteSiteModel: {'site_ids'},
LineModel: {
'line_id',
'site_id',
'line_code',
'line_name',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
LinePageQueryModel: {'page_num', 'page_size', 'site_id', 'line_code', 'line_name', 'status'},
DeleteLineModel: {'line_ids'},
WarehouseModel: {
'warehouse_id',
'warehouse_code',
'warehouse_name',
'address',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
WarehousePageQueryModel: {'page_num', 'page_size', 'warehouse_code', 'warehouse_name', 'status'},
DeleteWarehouseModel: {'warehouse_ids'},
PalletModel: {
'pallet_id',
'pallet_code',
'qr_token',
'capacity_packages',
'asset_status',
'remark',
'create_by',
'create_time',
'update_by',
'update_time',
'can_remove',
},
PalletCreateModel: {'pallet_code', 'capacity_packages', 'remark'},
PalletEditModel: {'pallet_id', 'capacity_packages', 'asset_status', 'remark'},
PalletPageQueryModel: {'page_num', 'page_size', 'pallet_code', 'asset_status'},
DeletePalletModel: {'pallet_ids'},
UserAssignModel: {'user_ids'},
}
OPERATION_MODEL_FIELDS = {
InboundCreateModel: {'warehouse_id', 'pallet_ids'},
InboundUpdateModel: {'inbound_id', 'warehouse_id', 'pallet_ids'},
ConfirmCommandModel: {'idempotency_key'},
FillingCreateModel: {
'qr_token',
'warehouse_id',
'strain_id',
'batch_no',
'package_spec',
'initial_packages',
'idempotency_key',
},
ShipmentCreateModel: {'source_warehouse_id', 'target_site_id', 'pallet_ids'},
ShipmentUpdateModel: {'shipment_id', 'source_warehouse_id', 'target_site_id', 'pallet_ids'},
UsageCreateModel: {'qr_token', 'line_id', 'used_packages', 'idempotency_key'},
ReturnRequestModel: {'source_site_id', 'target_warehouse_id', 'pallet_ids', 'idempotency_key'},
ReturnConfirmModel: {'scanned_pallet_ids', 'idempotency_key'},
InboundItemModel: {'inbound_item_id', 'inbound_id', 'pallet_id', 'create_by', 'create_time'},
InboundModel: {
'inbound_id',
'inbound_no',
'warehouse_id',
'status',
'confirm_idempotency_key',
'confirm_request_hash',
'confirmed_by',
'confirmed_at',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
'items',
},
FillingModel: {
'filling_id',
'filling_no',
'pallet_id',
'warehouse_id',
'strain_id',
'strain_code_snapshot',
'strain_name_snapshot',
'batch_no',
'package_spec',
'initial_packages',
'active_slot',
'closed_at',
'idempotency_key',
'request_hash',
'operator_id',
'operated_at',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
ShipmentItemModel: {
'shipment_item_id',
'shipment_id',
'pallet_id',
'active_filling_id',
'package_qty',
'create_by',
'create_time',
},
ShipmentModel: {
'shipment_id',
'shipment_no',
'source_warehouse_id',
'target_site_id',
'status',
'confirm_idempotency_key',
'confirm_request_hash',
'confirmed_by',
'confirmed_at',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
'items',
},
UsageModel: {
'usage_id',
'usage_no',
'pallet_id',
'active_filling_id',
'site_id',
'line_id',
'used_packages',
'remaining_packages_after',
'idempotency_key',
'request_hash',
'operator_id',
'occurred_at',
},
ReturnItemModel: {
'return_item_id',
'return_id',
'pallet_id',
'active_filling_id_snapshot',
'remaining_packages_snapshot',
'create_by',
'create_time',
},
ReturnModel: {
'return_id',
'return_no',
'source_site_id',
'target_warehouse_id',
'status',
'request_idempotency_key',
'request_hash',
'confirm_idempotency_key',
'confirm_request_hash',
'requested_by',
'requested_at',
'confirmed_by',
'confirmed_at',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
'items',
},
InboundPageQueryModel: {
'page_num',
'page_size',
'inbound_no',
'warehouse_id',
'status',
'begin_time',
'end_time',
},
FillingPageQueryModel: {
'page_num',
'page_size',
'filling_no',
'warehouse_id',
'strain_id',
'batch_no',
'pallet_code',
'begin_time',
'end_time',
},
ShipmentPageQueryModel: {
'page_num',
'page_size',
'shipment_no',
'source_warehouse_id',
'target_site_id',
'status',
'begin_time',
'end_time',
},
UsagePageQueryModel: {
'page_num',
'page_size',
'usage_no',
'site_id',
'line_id',
'pallet_code',
'begin_time',
'end_time',
},
ReturnPageQueryModel: {
'page_num',
'page_size',
'return_no',
'source_site_id',
'target_warehouse_id',
'status',
'begin_time',
'end_time',
},
InboundScanQueryModel: {'warehouse_id'},
ShipmentScanQueryModel: {'source_warehouse_id', 'target_site_id'},
ReturnScanQueryModel: {'mode', 'return_id'},
BusinessOptionModel: {'id', 'code', 'name'},
PalletScanResponseModel: {
'pallet_id',
'pallet_code',
'flow_state',
'warehouse_id',
'site_id',
'current_line_id',
'active_filling_id',
'strain_id',
'strain_code',
'strain_name',
'batch_no',
'package_spec',
'capacity_packages',
'remaining_packages',
'allowed_actions',
},
}
REPORT_MODEL_FIELDS = {
PalletInventoryReportQueryModel: {
'page_num',
'page_size',
'begin_time',
'end_time',
'pallet_code',
'flow_state',
'warehouse_id',
'site_id',
'line_id',
'strain_id',
'batch_no',
},
PalletInventoryReportModel: {
'pallet_id',
'pallet_code',
'flow_state',
'warehouse_id',
'warehouse_name',
'site_id',
'site_name',
'line_id',
'line_name',
'strain_id',
'strain_code',
'strain_name',
'batch_no',
'initial_packages',
'remaining_packages',
'updated_at',
},
FillingReportQueryModel: {
'page_num',
'page_size',
'begin_time',
'end_time',
'warehouse_id',
'pallet_code',
'strain_id',
'batch_no',
},
FillingReportModel: {
'filling_id',
'filling_no',
'operated_at',
'warehouse_id',
'warehouse_name',
'pallet_id',
'pallet_code',
'strain_id',
'strain_code_snapshot',
'strain_name_snapshot',
'batch_no',
'package_spec',
'initial_packages',
'operator_id',
'operator_name',
},
ShipmentReportQueryModel: {
'page_num',
'page_size',
'begin_time',
'end_time',
'shipment_no',
'source_warehouse_id',
'target_site_id',
'pallet_code',
'status',
},
ShipmentReportModel: {
'shipment_id',
'shipment_no',
'confirmed_at',
'source_warehouse_id',
'source_warehouse_name',
'target_site_id',
'target_site_name',
'customer_id',
'customer_name',
'pallet_id',
'pallet_code',
'active_filling_id',
'package_qty',
},
ConsumptionReportQueryModel: {
'page_num',
'page_size',
'begin_time',
'end_time',
'customer_id',
'site_id',
'line_id',
'pallet_code',
'strain_id',
'batch_no',
},
ConsumptionReportModel: {
'usage_id',
'usage_no',
'occurred_at',
'customer_id',
'customer_name',
'site_id',
'site_name',
'line_id',
'line_name',
'pallet_id',
'pallet_code',
'strain_id',
'strain_code',
'strain_name',
'batch_no',
'used_packages',
'remaining_packages_after',
'operator_id',
'operator_name',
},
ReturnReportQueryModel: {
'page_num',
'page_size',
'begin_time',
'end_time',
'return_no',
'source_site_id',
'target_warehouse_id',
'pallet_code',
'status',
},
ReturnReportModel: {
'return_id',
'return_no',
'status',
'requested_at',
'confirmed_at',
'source_site_id',
'source_site_name',
'target_warehouse_id',
'target_warehouse_name',
'pallet_id',
'pallet_code',
'active_filling_id_snapshot',
'remaining_packages_snapshot',
},
PalletTraceReportQueryModel: {
'page_num',
'page_size',
'begin_time',
'end_time',
'pallet_code',
'event_type',
},
PalletTraceReportModel: {
'event_id',
'event_seq',
'event_type',
'source_type',
'source_id',
'pallet_id',
'pallet_code',
'before_state',
'after_state',
'before_warehouse_id',
'before_warehouse_name',
'before_site_id',
'before_site_name',
'before_line_id',
'before_line_name',
'after_warehouse_id',
'after_warehouse_name',
'after_site_id',
'after_site_name',
'after_line_id',
'after_line_name',
'before_remaining_packages',
'after_remaining_packages',
'operator_id',
'operator_name',
'occurred_at',
},
}
WRITE_MODELS = (
StrainModel,
DeleteStrainModel,
CustomerModel,
DeleteCustomerModel,
SiteModel,
DeleteSiteModel,
LineModel,
DeleteLineModel,
WarehouseModel,
DeleteWarehouseModel,
PalletCreateModel,
PalletEditModel,
DeletePalletModel,
UserAssignModel,
InboundCreateModel,
InboundUpdateModel,
ConfirmCommandModel,
FillingCreateModel,
ShipmentCreateModel,
ShipmentUpdateModel,
UsageCreateModel,
ReturnRequestModel,
ReturnConfirmModel,
)
PAGE_QUERY_MODELS = (
StrainPageQueryModel,
CustomerPageQueryModel,
SitePageQueryModel,
LinePageQueryModel,
WarehousePageQueryModel,
PalletPageQueryModel,
InboundPageQueryModel,
FillingPageQueryModel,
ShipmentPageQueryModel,
UsagePageQueryModel,
ReturnPageQueryModel,
PalletInventoryReportQueryModel,
FillingReportQueryModel,
ShipmentReportQueryModel,
ConsumptionReportQueryModel,
ReturnReportQueryModel,
PalletTraceReportQueryModel,
)
class BusinessVoContractTest(unittest.TestCase):
def test_all_models_have_exact_fields_and_camel_case_aliases(self) -> None:
expected_models = MASTER_MODEL_FIELDS | OPERATION_MODEL_FIELDS | REPORT_MODEL_FIELDS
for model_class, expected_fields in expected_models.items():
with self.subTest(model=model_class.__name__):
self.assertEqual(expected_fields, set(model_class.model_fields))
self.assertTrue(model_class.model_config.get('from_attributes'))
self.assertTrue(model_class.model_config.get('populate_by_name'))
self.assertIs(to_camel, model_class.model_config.get('alias_generator'))
aliases = {name: field.alias for name, field in model_class.model_fields.items()}
self.assertEqual({name: to_camel(name) for name in expected_fields}, aliases)
def test_write_models_forbid_extra_fields(self) -> None:
for model_class in WRITE_MODELS:
with self.subTest(model=model_class.__name__):
self.assertEqual('forbid', model_class.model_config.get('extra'))
def test_page_queries_have_positive_defaults(self) -> None:
for model_class in PAGE_QUERY_MODELS:
with self.subTest(model=model_class.__name__):
model = model_class.model_validate({})
self.assertEqual((1, 10), (model.page_num, model.page_size))
with self.assertRaises(ValidationError):
model_class.model_validate({'pageNum': 0})
with self.assertRaises(ValidationError):
model_class.model_validate({'pageSize': 0})
def test_filling_accepts_camel_case_and_positive_packages(self) -> None:
model = FillingCreateModel.model_validate(
{
'qrToken': 'qr-12345678',
'warehouseId': 1,
'strainId': 2,
'batchNo': 'B20260724',
'packageSpec': '20kg/包',
'initialPackages': 10,
'idempotencyKey': 'fill-key-1',
}
)
self.assertEqual(10, model.initial_packages)
self.assertEqual(10, model.model_dump(by_alias=True)['initialPackages'])
def test_usage_rejects_zero_packages(self) -> None:
with self.assertRaises(ValidationError):
UsageCreateModel.model_validate(
{
'qrToken': 'qr-12345678',
'lineId': 3,
'usedPackages': 0,
'idempotencyKey': 'usage-key-1',
}
)
def test_usage_rejects_non_integer_packages(self) -> None:
for used_packages in (1.5, '1', True):
with self.subTest(used_packages=used_packages), self.assertRaises(ValidationError):
UsageCreateModel.model_validate(
{
'qrToken': 'qr-12345678',
'lineId': 3,
'usedPackages': used_packages,
'idempotencyKey': 'usage-key-1',
}
)
def test_other_package_inputs_are_strict_positive_integers(self) -> None:
cases = (
(
PalletCreateModel,
{'palletCode': 'PLT-001', 'capacityPackages': 10, 'remark': None},
'capacityPackages',
),
(
FillingCreateModel,
{
'qrToken': 'qr-12345678',
'warehouseId': 1,
'strainId': 2,
'batchNo': 'B20260724',
'packageSpec': '20kg/包',
'initialPackages': 10,
'idempotencyKey': 'fill-key-1',
},
'initialPackages',
),
)
for model_class, payload, field_name in cases:
for invalid_value in (0, 1.5, '1', True):
with self.subTest(model=model_class.__name__, value=invalid_value), self.assertRaises(ValidationError):
model_class.model_validate(payload | {field_name: invalid_value})
def test_id_lists_reject_empty_non_positive_and_duplicate_values(self) -> None:
cases = (
(DeleteStrainModel, {}, 'strainIds'),
(DeleteCustomerModel, {}, 'customerIds'),
(DeleteSiteModel, {}, 'siteIds'),
(DeleteLineModel, {}, 'lineIds'),
(DeleteWarehouseModel, {}, 'warehouseIds'),
(DeletePalletModel, {}, 'palletIds'),
(UserAssignModel, {}, 'userIds'),
(InboundCreateModel, {'warehouseId': 1}, 'palletIds'),
(InboundUpdateModel, {'inboundId': 1, 'warehouseId': 1}, 'palletIds'),
(ShipmentCreateModel, {'sourceWarehouseId': 1, 'targetSiteId': 2}, 'palletIds'),
(
ShipmentUpdateModel,
{'shipmentId': 1, 'sourceWarehouseId': 1, 'targetSiteId': 2},
'palletIds',
),
(
ReturnRequestModel,
{'sourceSiteId': 1, 'targetWarehouseId': 2, 'idempotencyKey': 'return-key-1'},
'palletIds',
),
(ReturnConfirmModel, {'idempotencyKey': 'return-confirm-key-1'}, 'scannedPalletIds'),
)
for model_class, payload, field_name in cases:
for invalid_value in ([], [0], [1, 1]):
with self.subTest(model=model_class.__name__, value=invalid_value), self.assertRaises(ValidationError):
model_class.model_validate(payload | {field_name: invalid_value})
def test_pallet_edit_forbids_identity_fields(self) -> None:
valid_payload = {
'palletId': 1,
'capacityPackages': 20,
'assetStatus': '0',
'remark': None,
}
for field_name in ('palletCode', 'qrToken'):
with self.subTest(field=field_name), self.assertRaises(ValidationError):
PalletEditModel.model_validate(valid_payload | {field_name: 'immutable'})
def test_return_scan_mode_controls_return_id(self) -> None:
request_model = ReturnScanQueryModel.model_validate({'mode': 'request'})
confirm_model = ReturnScanQueryModel.model_validate({'mode': 'confirm', 'returnId': 9})
self.assertIsNone(request_model.return_id)
self.assertEqual(9, confirm_model.return_id)
for payload in (
{'mode': 'confirm'},
{'mode': 'request', 'returnId': 9},
{'mode': 'other'},
):
with self.subTest(payload=payload), self.assertRaises(ValidationError):
ReturnScanQueryModel.model_validate(payload)
def test_required_business_strings_reject_whitespace(self) -> None:
with self.assertRaises(ValidationError):
ConfirmCommandModel.model_validate({'idempotencyKey': ' '})
with self.assertRaises(ValidationError):
StrainModel.model_validate({'strainCode': '\t', 'strainName': '菌种 A', 'status': '0'})
with self.assertRaises(ValidationError):
FillingCreateModel.model_validate(
{
'qrToken': 'qr-12345678',
'warehouseId': 1,
'strainId': 2,
'batchNo': ' ',
'packageSpec': '20kg/包',
'initialPackages': 10,
'idempotencyKey': 'fill-key-1',
}
)
def test_business_ids_must_be_positive(self) -> None:
for model_class, payload in (
(InboundScanQueryModel, {'warehouseId': 0}),
(ShipmentScanQueryModel, {'sourceWarehouseId': 1, 'targetSiteId': -1}),
(BusinessOptionModel, {'id': 0, 'code': 'WH-01', 'name': '一号仓'}),
):
with self.subTest(model=model_class.__name__), self.assertRaises(ValidationError):
model_class.model_validate(payload)
def test_non_negative_output_quantities_are_validated(self) -> None:
with self.assertRaises(ValidationError):
PalletScanResponseModel.model_validate(
{
'palletId': 1,
'palletCode': 'PLT-001',
'flowState': 'AT_SITE_PARTIAL',
'capacityPackages': 20,
'remainingPackages': -1,
'allowedActions': [],
}
)
if __name__ == '__main__':
unittest.main()