import unittest from datetime import datetime 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, ) from module_business.entity.vo.test_batch_vo import ( AdoptionExecuteRequestModel, AdoptionPreviewRequestModel, AdoptionRootModel, BatchImpactModel, PurgeExecuteRequestModel, PurgePreviewModel, TestBatchCreateModel, TestBatchDetailModel, TestBatchEditModel, TestBatchModel, TestBatchOptionModel, TestBatchPreviewModel, TestBatchUserAssignModel, ) MASTER_MODEL_FIELDS = { StrainModel: { 'strain_id', 'strain_code', 'strain_name', 'status', 'create_by', 'create_time', 'update_by', 'update_time', 'remark', 'can_remove', }, 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', 'can_remove', }, 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', 'can_remove', }, 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', 'can_remove', }, 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', 'can_remove', }, 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', 'remaining_packages', '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', }, } TEST_BATCH_MODEL_FIELDS = { TestBatchCreateModel: {'batch_name', 'remark'}, TestBatchEditModel: {'remark'}, TestBatchUserAssignModel: {'user_ids'}, TestBatchModel: { 'batch_id', 'batch_code', 'batch_name', 'status', 'remark', 'create_by', 'create_time', 'update_by', 'update_time', 'closed_by', 'closed_at', 'purged_by', 'purged_at', 'purge_summary', }, TestBatchOptionModel: {'batch_id', 'batch_code', 'batch_name', 'status'}, BatchImpactModel: {'table_name', 'label', 'count', 'keys'}, TestBatchDetailModel: { 'batch_id', 'batch_code', 'batch_name', 'status', 'remark', 'create_by', 'create_time', 'update_by', 'update_time', 'closed_by', 'closed_at', 'purged_by', 'purged_at', 'purge_summary', 'user_ids', 'data_summary', }, TestBatchPreviewModel: { 'batch_id', 'batch_name', 'preview_token', 'expires_at', 'total_count', 'impacts', }, PurgePreviewModel: { 'batch_id', 'batch_name', 'preview_token', 'expires_at', 'total_count', 'impacts', 'online_purge_allowed', 'online_purge_limit', 'blocking_reason', }, AdoptionRootModel: {'resource_type', 'resource_ids'}, AdoptionPreviewRequestModel: {'roots'}, AdoptionExecuteRequestModel: {'roots', 'batch_name', 'preview_token'}, PurgeExecuteRequestModel: {'batch_name', 'preview_token'}, } 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, ) OPERATION_PAGE_QUERY_MODELS = ( InboundPageQueryModel, FillingPageQueryModel, ShipmentPageQueryModel, UsagePageQueryModel, ReturnPageQueryModel, ) PAGE_QUERY_MODELS = ( StrainPageQueryModel, CustomerPageQueryModel, SitePageQueryModel, LinePageQueryModel, WarehousePageQueryModel, PalletPageQueryModel, *OPERATION_PAGE_QUERY_MODELS, PalletInventoryReportQueryModel, FillingReportQueryModel, ShipmentReportQueryModel, ConsumptionReportQueryModel, ReturnReportQueryModel, PalletTraceReportQueryModel, ) STRICT_POSITIVE_PACKAGE_FIELDS = { PalletModel: {'capacity_packages'}, FillingModel: {'initial_packages'}, ShipmentItemModel: {'package_qty'}, UsageModel: {'used_packages'}, PalletScanResponseModel: {'capacity_packages'}, PalletInventoryReportModel: {'initial_packages'}, FillingReportModel: {'initial_packages'}, ShipmentReportModel: {'package_qty'}, ConsumptionReportModel: {'used_packages'}, } STRICT_NON_NEGATIVE_PACKAGE_FIELDS = { ShipmentItemModel: {'remaining_packages'}, UsageModel: {'remaining_packages_after'}, ReturnItemModel: {'remaining_packages_snapshot'}, PalletScanResponseModel: {'remaining_packages'}, PalletInventoryReportModel: {'remaining_packages'}, ConsumptionReportModel: {'remaining_packages_after'}, ReturnReportModel: {'remaining_packages_snapshot'}, PalletTraceReportModel: {'before_remaining_packages', 'after_remaining_packages'}, } PACKAGE_MODEL_PAYLOADS = { PalletModel: { 'pallet_id': 1, 'pallet_code': 'PLT-001', 'qr_token': 'qr-12345678', 'capacity_packages': 20, 'asset_status': '0', }, FillingModel: { 'filling_id': 1, 'filling_no': 'F202607240001', 'pallet_id': 1, 'warehouse_id': 1, 'strain_id': 1, 'strain_code_snapshot': 'STRAIN-01', 'strain_name_snapshot': '菌种一号', 'batch_no': 'B20260724', 'package_spec': '20kg/包', 'initial_packages': 20, 'idempotency_key': 'fill-key-1', 'request_hash': 'fill-hash-1', 'operator_id': 1, 'operated_at': '2026-07-24T10:00:00', }, ShipmentItemModel: { 'shipment_item_id': 1, 'shipment_id': 1, 'pallet_id': 1, 'package_qty': 20, 'remaining_packages': 8, }, UsageModel: { 'usage_id': 1, 'usage_no': 'U202607240001', 'pallet_id': 1, 'active_filling_id': 1, 'site_id': 1, 'line_id': 1, 'used_packages': 1, 'remaining_packages_after': 19, 'idempotency_key': 'usage-key-1', 'request_hash': 'usage-hash-1', 'operator_id': 1, 'occurred_at': '2026-07-24T11:00:00', }, ReturnItemModel: { 'return_item_id': 1, 'return_id': 1, 'pallet_id': 1, 'remaining_packages_snapshot': 0, }, PalletScanResponseModel: { 'pallet_id': 1, 'pallet_code': 'PLT-001', 'flow_state': 'AT_SITE_PARTIAL', 'capacity_packages': 20, 'remaining_packages': 10, 'allowed_actions': [], }, PalletInventoryReportModel: { 'pallet_id': 1, 'pallet_code': 'PLT-001', 'flow_state': 'FILLED_IN_WAREHOUSE', 'initial_packages': 20, 'remaining_packages': 20, 'updated_at': '2026-07-24T10:00:00', }, FillingReportModel: { 'filling_id': 1, 'filling_no': 'F202607240001', 'operated_at': '2026-07-24T10:00:00', 'warehouse_id': 1, 'pallet_id': 1, 'pallet_code': 'PLT-001', 'strain_id': 1, 'strain_code_snapshot': 'STRAIN-01', 'strain_name_snapshot': '菌种一号', 'batch_no': 'B20260724', 'package_spec': '20kg/包', 'initial_packages': 20, 'operator_id': 1, }, ShipmentReportModel: { 'shipment_id': 1, 'shipment_no': 'S202607240001', 'confirmed_at': '2026-07-24T12:00:00', 'source_warehouse_id': 1, 'target_site_id': 1, 'customer_id': 1, 'pallet_id': 1, 'pallet_code': 'PLT-001', 'package_qty': 20, }, ConsumptionReportModel: { 'usage_id': 1, 'usage_no': 'U202607240001', 'occurred_at': '2026-07-24T13:00:00', 'customer_id': 1, 'site_id': 1, 'line_id': 1, 'pallet_id': 1, 'pallet_code': 'PLT-001', 'strain_id': 1, 'batch_no': 'B20260724', 'used_packages': 1, 'remaining_packages_after': 19, 'operator_id': 1, }, ReturnReportModel: { 'return_id': 1, 'return_no': 'R202607240001', 'status': 'REQUESTED', 'requested_at': '2026-07-24T14:00:00', 'source_site_id': 1, 'target_warehouse_id': 1, 'pallet_id': 1, 'pallet_code': 'PLT-001', 'remaining_packages_snapshot': 0, }, PalletTraceReportModel: { 'event_id': 1, 'event_seq': 1, 'event_type': 'INBOUND_CONFIRMED', 'source_type': 'INBOUND', 'source_id': 1, 'pallet_id': 1, 'pallet_code': 'PLT-001', 'before_state': 'UNSTOCKED', 'after_state': 'EMPTY_IN_WAREHOUSE', 'before_remaining_packages': 0, 'after_remaining_packages': 0, 'operator_id': 1, 'occurred_at': '2026-07-24T15:00:00', }, } 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 | TEST_BATCH_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_all_test_batch_models_forbid_extra_fields(self) -> None: for model_class in TEST_BATCH_MODEL_FIELDS: with self.subTest(model=model_class.__name__): self.assertEqual('forbid', model_class.model_config.get('extra')) def test_test_batch_models_accept_snake_case_and_camel_case(self) -> None: camel_case = TestBatchCreateModel(batchName='验收批次', remark='camel') snake_case = TestBatchCreateModel(batch_name='验收批次', remark='snake') self.assertEqual('验收批次', camel_case.batch_name) self.assertEqual('验收批次', snake_case.batch_name) def test_test_batch_edit_forbids_immutable_fields(self) -> None: for field_name, value in (('batchName', '新名称'), ('status', 'CLOSED')): with self.subTest(field=field_name), self.assertRaises(ValidationError): TestBatchEditModel.model_validate({'remark': None, field_name: value}) def test_test_batch_positive_ids_and_non_empty_roots_are_validated(self) -> None: with self.assertRaises(ValidationError): TestBatchUserAssignModel(userIds=[0]) with self.assertRaises(ValidationError): AdoptionRootModel(resourceType='strain', resourceIds=[]) with self.assertRaises(ValidationError): AdoptionPreviewRequestModel(roots=[]) def test_test_batch_status_rejects_unknown_value(self) -> None: with self.assertRaises(ValidationError) as context: TestBatchModel( batchId=1, batchCode='TB-1', batchName='验收批次', status='UNKNOWN', createBy='admin', createTime=datetime(2026, 9, 22), updateBy='admin', updateTime=datetime(2026, 9, 22), ) self.assertIn(('status',), {tuple(error['loc']) for error in context.exception.errors()}) 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_operation_page_query_dates_require_strict_calendar_format(self) -> None: for model_class in OPERATION_PAGE_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_operation_page_query_date_range_must_not_be_inverted(self) -> None: for model_class in OPERATION_PAGE_QUERY_MODELS: 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='inverted'), self.assertRaises(ValidationError): model_class.model_validate({'beginTime': '2026-07-03', 'endTime': '2026-07-02'}) def test_operation_page_query_rejects_overflowing_end_but_allows_same_begin(self) -> None: for model_class in OPERATION_PAGE_QUERY_MODELS: with self.subTest(model=model_class.__name__, field='beginTime'): model = model_class.model_validate({'beginTime': '9999-12-31'}) self.assertEqual('9999-12-31', model.begin_time) self.assertIsNone(model.end_time) with ( self.subTest(model=model_class.__name__, field='endTime'), self.assertRaises(ValidationError), ): model_class.model_validate({'endTime': '9999-12-31'}) def test_all_page_sizes_are_capped_at_100(self) -> None: for model_class in PAGE_QUERY_MODELS: with self.subTest(model=model_class.__name__, page_size=100): self.assertEqual(100, model_class.model_validate({'pageSize': 100}).page_size) with ( self.subTest(model=model_class.__name__, page_size=101), self.assertRaises(ValidationError), ): model_class.model_validate({'pageSize': 101}) 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_output_occurrence_and_capacity_fields_are_strict_positive_integers(self) -> None: accepted_invalid_values = [] for model_class, field_names in STRICT_POSITIVE_PACKAGE_FIELDS.items(): payload = PACKAGE_MODEL_PAYLOADS[model_class] model_class.model_validate(payload) for field_name in field_names: for invalid_value in (0, '1', 1.5, True): try: model_class.model_validate(payload | {field_name: invalid_value}) except ValidationError: continue accepted_invalid_values.append((model_class.__name__, field_name, repr(invalid_value))) self.assertEqual([], accepted_invalid_values) self.assertIsNone( ShipmentItemModel.model_validate( PACKAGE_MODEL_PAYLOADS[ShipmentItemModel] | {'package_qty': None} ).package_qty ) self.assertIsNone( PalletInventoryReportModel.model_validate( PACKAGE_MODEL_PAYLOADS[PalletInventoryReportModel] | {'initial_packages': None} ).initial_packages ) def test_output_snapshot_and_remaining_fields_are_strict_non_negative_integers(self) -> None: accepted_invalid_values = [] for model_class, field_names in STRICT_NON_NEGATIVE_PACKAGE_FIELDS.items(): payload = PACKAGE_MODEL_PAYLOADS[model_class] for field_name in field_names: zero_model = model_class.model_validate(payload | {field_name: 0}) self.assertEqual(0, getattr(zero_model, field_name)) for invalid_value in (-1, '1', 1.5, True): try: model_class.model_validate(payload | {field_name: invalid_value}) except ValidationError: continue accepted_invalid_values.append((model_class.__name__, field_name, repr(invalid_value))) self.assertEqual([], accepted_invalid_values) shipment_item = ShipmentItemModel.model_validate( PACKAGE_MODEL_PAYLOADS[ShipmentItemModel] | { 'packageQty': None, 'remainingPackages': None, } ) self.assertIsNone(shipment_item.package_qty) self.assertIsNone(shipment_item.remaining_packages) self.assertIn('remainingPackages', shipment_item.model_dump(by_alias=True)) 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'), (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_user_assignment_allows_empty_and_rejects_invalid_non_empty_ids(self) -> None: try: empty_assignment = UserAssignModel.model_validate({'userIds': []}) except ValidationError as exc: self.fail(f'完整替换用户关联必须允许空数组：{exc}') self.assertEqual([], empty_assignment.user_ids) for invalid_value in ([0], [1, 1]): with self.subTest(value=invalid_value), self.assertRaises(ValidationError): UserAssignModel.model_validate({'userIds': 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()