niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

功能：新增托盘业务数据模型

Browse Source
This commit is contained in:
焦龙言
2026-07-24 20:09:36 +08:00
parent cb3bf5bf4c
commit f6e9566f70
3 changed files with 1279 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_model_contract.py
+609
View File
@@ -0,0 +1,609 @@
import unittest
from sqlalchemy import CHAR, BigInteger, CheckConstraint, Integer, String, Table, UniqueConstraint
from config.database import Base
from module_business.entity.do import master_do, operation_do
EXPECTED_COLUMNS = {
'biz_strain': {
'strain_id',
'strain_code',
'strain_name',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
'biz_warehouse': {
'warehouse_id',
'warehouse_code',
'warehouse_name',
'address',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
'biz_customer': {
'customer_id',
'customer_code',
'customer_name',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
'biz_customer_site': {
'site_id',
'customer_id',
'site_code',
'site_name',
'address',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
'biz_production_line': {
'line_id',
'site_id',
'line_code',
'line_name',
'status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
'biz_pallet': {
'pallet_id',
'pallet_code',
'qr_token',
'capacity_packages',
'asset_status',
'create_by',
'create_time',
'update_by',
'update_time',
'remark',
},
'biz_user_warehouse': {'user_id', 'warehouse_id', 'create_by', 'create_time'},
'biz_user_site': {'user_id', 'site_id', 'create_by', 'create_time'},
'biz_inbound_order': {
'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',
},
'biz_inbound_item': {
'inbound_item_id',
'inbound_id',
'pallet_id',
'create_by',
'create_time',
},
'biz_filling_record': {
'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',
},
'biz_shipment_order': {
'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',
},
'biz_shipment_item': {
'shipment_item_id',
'shipment_id',
'pallet_id',
'active_filling_id',
'package_qty',
'create_by',
'create_time',
},
'biz_usage_record': {
'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',
},
'biz_return_order': {
'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',
},
'biz_return_item': {
'return_item_id',
'return_id',
'pallet_id',
'active_filling_id_snapshot',
'remaining_packages_snapshot',
'create_by',
'create_time',
},
'biz_pallet_inventory': {
'pallet_id',
'active_filling_id',
'warehouse_id',
'site_id',
'current_line_id',
'flow_state',
'remaining_packages',
'last_event_id',
'version',
'updated_at',
},
'biz_pallet_event': {
'event_id',
'pallet_id',
'event_seq',
'event_type',
'source_type',
'source_id',
'before_state',
'after_state',
'before_warehouse_id',
'before_site_id',
'before_line_id',
'after_warehouse_id',
'after_site_id',
'after_line_id',
'before_remaining_packages',
'after_remaining_packages',
'operator_id',
'occurred_at',
'idempotency_key',
'request_hash',
},
}
EXPECTED_PRIMARY_KEYS = {
'biz_strain': ('strain_id',),
'biz_warehouse': ('warehouse_id',),
'biz_customer': ('customer_id',),
'biz_customer_site': ('site_id',),
'biz_production_line': ('line_id',),
'biz_pallet': ('pallet_id',),
'biz_user_warehouse': ('user_id', 'warehouse_id'),
'biz_user_site': ('user_id', 'site_id'),
'biz_inbound_order': ('inbound_id',),
'biz_inbound_item': ('inbound_item_id',),
'biz_filling_record': ('filling_id',),
'biz_shipment_order': ('shipment_id',),
'biz_shipment_item': ('shipment_item_id',),
'biz_usage_record': ('usage_id',),
'biz_return_order': ('return_id',),
'biz_return_item': ('return_item_id',),
'biz_pallet_inventory': ('pallet_id',),
'biz_pallet_event': ('event_id',),
}
EXPECTED_FOREIGN_KEYS = {
'biz_strain': set(),
'biz_warehouse': set(),
'biz_customer': set(),
'biz_customer_site': {('customer_id', 'biz_customer.customer_id')},
'biz_production_line': {('site_id', 'biz_customer_site.site_id')},
'biz_pallet': set(),
'biz_user_warehouse': {
('user_id', 'sys_user.user_id'),
('warehouse_id', 'biz_warehouse.warehouse_id'),
},
'biz_user_site': {
('user_id', 'sys_user.user_id'),
('site_id', 'biz_customer_site.site_id'),
},
'biz_inbound_order': {
('warehouse_id', 'biz_warehouse.warehouse_id'),
('confirmed_by', 'sys_user.user_id'),
},
'biz_inbound_item': {
('inbound_id', 'biz_inbound_order.inbound_id'),
('pallet_id', 'biz_pallet.pallet_id'),
},
'biz_filling_record': {
('pallet_id', 'biz_pallet.pallet_id'),
('warehouse_id', 'biz_warehouse.warehouse_id'),
('strain_id', 'biz_strain.strain_id'),
('operator_id', 'sys_user.user_id'),
},
'biz_shipment_order': {
('source_warehouse_id', 'biz_warehouse.warehouse_id'),
('target_site_id', 'biz_customer_site.site_id'),
('confirmed_by', 'sys_user.user_id'),
},
'biz_shipment_item': {
('shipment_id', 'biz_shipment_order.shipment_id'),
('pallet_id', 'biz_pallet.pallet_id'),
('active_filling_id', 'biz_filling_record.filling_id'),
},
'biz_usage_record': {
('pallet_id', 'biz_pallet.pallet_id'),
('active_filling_id', 'biz_filling_record.filling_id'),
('site_id', 'biz_customer_site.site_id'),
('line_id', 'biz_production_line.line_id'),
('operator_id', 'sys_user.user_id'),
},
'biz_return_order': {
('source_site_id', 'biz_customer_site.site_id'),
('target_warehouse_id', 'biz_warehouse.warehouse_id'),
('requested_by', 'sys_user.user_id'),
('confirmed_by', 'sys_user.user_id'),
},
'biz_return_item': {
('return_id', 'biz_return_order.return_id'),
('pallet_id', 'biz_pallet.pallet_id'),
('active_filling_id_snapshot', 'biz_filling_record.filling_id'),
},
'biz_pallet_inventory': {
('pallet_id', 'biz_pallet.pallet_id'),
('active_filling_id', 'biz_filling_record.filling_id'),
('warehouse_id', 'biz_warehouse.warehouse_id'),
('site_id', 'biz_customer_site.site_id'),
('current_line_id', 'biz_production_line.line_id'),
('last_event_id', 'biz_pallet_event.event_id'),
},
'biz_pallet_event': {
('pallet_id', 'biz_pallet.pallet_id'),
('before_warehouse_id', 'biz_warehouse.warehouse_id'),
('before_site_id', 'biz_customer_site.site_id'),
('before_line_id', 'biz_production_line.line_id'),
('after_warehouse_id', 'biz_warehouse.warehouse_id'),
('after_site_id', 'biz_customer_site.site_id'),
('after_line_id', 'biz_production_line.line_id'),
('operator_id', 'sys_user.user_id'),
},
}
EXPECTED_UNIQUE_SETS = {
'biz_strain': {frozenset({'strain_code'})},
'biz_warehouse': {frozenset({'warehouse_code'})},
'biz_customer': {frozenset({'customer_code'})},
'biz_customer_site': {frozenset({'customer_id', 'site_code'})},
'biz_production_line': {frozenset({'site_id', 'line_code'})},
'biz_pallet': {frozenset({'pallet_code'}), frozenset({'qr_token'})},
'biz_user_warehouse': set(),
'biz_user_site': set(),
'biz_inbound_order': {
frozenset({'inbound_no'}),
frozenset({'confirm_idempotency_key'}),
},
'biz_inbound_item': {frozenset({'inbound_id', 'pallet_id'})},
'biz_filling_record': {
frozenset({'filling_no'}),
frozenset({'idempotency_key'}),
frozenset({'pallet_id', 'active_slot'}),
},
'biz_shipment_order': {
frozenset({'shipment_no'}),
frozenset({'confirm_idempotency_key'}),
},
'biz_shipment_item': {frozenset({'shipment_id', 'pallet_id'})},
'biz_usage_record': {
frozenset({'usage_no'}),
frozenset({'idempotency_key'}),
},
'biz_return_order': {
frozenset({'return_no'}),
frozenset({'request_idempotency_key'}),
frozenset({'confirm_idempotency_key'}),
},
'biz_return_item': {frozenset({'return_id', 'pallet_id'})},
'biz_pallet_inventory': set(),
'biz_pallet_event': {
frozenset({'pallet_id', 'event_seq'}),
frozenset({'pallet_id', 'idempotency_key'}),
},
}
def _unique_column_sets(table: Table) -> set[frozenset[str]]:
return {
frozenset(column.name for column in constraint.columns)
for constraint in table.constraints
if isinstance(constraint, UniqueConstraint)
}
def _normalized_check_sql(table: Table) -> dict[str, str]:
return {
constraint.name: ''.join(str(constraint.sqltext).split())
for constraint in table.constraints
if isinstance(constraint, CheckConstraint) and constraint.name is not None
}
class BusinessModelContractTest(unittest.TestCase):
def test_exact_business_tables_are_registered(self) -> None:
expected = {
'biz_strain',
'biz_warehouse',
'biz_customer',
'biz_customer_site',
'biz_production_line',
'biz_pallet',
'biz_user_warehouse',
'biz_user_site',
'biz_inbound_order',
'biz_inbound_item',
'biz_filling_record',
'biz_shipment_order',
'biz_shipment_item',
'biz_usage_record',
'biz_return_order',
'biz_return_item',
'biz_pallet_inventory',
'biz_pallet_event',
}
actual = {name for name in Base.metadata.tables if name.startswith('biz_')}
self.assertEqual(expected, actual)
def test_exact_model_classes_and_columns(self) -> None:
models = {
master_do.BizStrain: 'biz_strain',
master_do.BizWarehouse: 'biz_warehouse',
master_do.BizCustomer: 'biz_customer',
master_do.BizCustomerSite: 'biz_customer_site',
master_do.BizProductionLine: 'biz_production_line',
master_do.BizPallet: 'biz_pallet',
master_do.BizUserWarehouse: 'biz_user_warehouse',
master_do.BizUserSite: 'biz_user_site',
operation_do.BizInboundOrder: 'biz_inbound_order',
operation_do.BizInboundItem: 'biz_inbound_item',
operation_do.BizFillingRecord: 'biz_filling_record',
operation_do.BizShipmentOrder: 'biz_shipment_order',
operation_do.BizShipmentItem: 'biz_shipment_item',
operation_do.BizUsageRecord: 'biz_usage_record',
operation_do.BizReturnOrder: 'biz_return_order',
operation_do.BizReturnItem: 'biz_return_item',
operation_do.BizPalletInventory: 'biz_pallet_inventory',
operation_do.BizPalletEvent: 'biz_pallet_event',
}
self.assertEqual(set(EXPECTED_COLUMNS), set(models.values()))
for model, table_name in models.items():
with self.subTest(table_name=table_name):
self.assertEqual(table_name, model.__tablename__)
self.assertEqual(EXPECTED_COLUMNS[table_name], set(model.__table__.columns.keys()))
def test_primary_keys_and_foreign_keys_match_contract(self) -> None:
for table_name, expected_primary_key in EXPECTED_PRIMARY_KEYS.items():
table = Base.metadata.tables[table_name]
with self.subTest(table_name=table_name, constraint='primary_key'):
self.assertEqual(expected_primary_key, tuple(table.primary_key.columns.keys()))
actual_foreign_keys = {
(foreign_key.parent.name, foreign_key.target_fullname) for foreign_key in table.foreign_keys
}
with self.subTest(table_name=table_name, constraint='foreign_keys'):
self.assertEqual(EXPECTED_FOREIGN_KEYS[table_name], actual_foreign_keys)
for table_name, primary_key in EXPECTED_PRIMARY_KEYS.items():
for column_name in primary_key:
with self.subTest(table_name=table_name, column_name=column_name):
self.assertIsInstance(Base.metadata.tables[table_name].c[column_name].type, BigInteger)
def test_all_required_unique_constraints_match_contract(self) -> None:
for table_name, expected_unique_sets in EXPECTED_UNIQUE_SETS.items():
with self.subTest(table_name=table_name):
self.assertEqual(expected_unique_sets, _unique_column_sets(Base.metadata.tables[table_name]))
def test_pallet_identifiers_are_unique(self) -> None:
table = Base.metadata.tables['biz_pallet']
unique_columns = {column.name for column in table.columns if column.unique}
self.assertEqual({'pallet_code', 'qr_token'}, unique_columns)
def test_checks_lock_quantities_states_active_slot_and_location(self) -> None:
expected_checks = {
'biz_strain': {'ck_biz_strain_status': "statusIN('0','1')"},
'biz_warehouse': {'ck_biz_warehouse_status': "statusIN('0','1')"},
'biz_customer': {'ck_biz_customer_status': "statusIN('0','1')"},
'biz_customer_site': {'ck_biz_customer_site_status': "statusIN('0','1')"},
'biz_production_line': {'ck_biz_production_line_status': "statusIN('0','1')"},
'biz_pallet': {
'ck_biz_pallet_capacity_packages': 'capacity_packages>0',
'ck_biz_pallet_asset_status': "asset_statusIN('0','1')",
},
'biz_inbound_order': {
'ck_biz_inbound_order_status': "statusIN('DRAFT','CONFIRMED')",
},
'biz_filling_record': {
'ck_biz_filling_record_initial_packages': 'initial_packages>0',
'ck_biz_filling_record_active_slot': "active_slotISNULLORactive_slot='ACTIVE'",
},
'biz_shipment_order': {
'ck_biz_shipment_order_status': "statusIN('DRAFT','CONFIRMED')",
},
'biz_shipment_item': {
'ck_biz_shipment_item_package_qty': 'package_qtyISNULLORpackage_qty>0',
},
'biz_usage_record': {
'ck_biz_usage_record_used_packages': 'used_packages>0',
'ck_biz_usage_record_remaining_packages_after': 'remaining_packages_after>=0',
},
'biz_return_order': {
'ck_biz_return_order_status': "statusIN('REQUESTED','CONFIRMED')",
},
'biz_return_item': {
'ck_biz_return_item_remaining_packages_snapshot': 'remaining_packages_snapshot>=0',
},
'biz_pallet_inventory': {
'ck_biz_pallet_inventory_flow_state': (
"flow_stateIN('UNSTOCKED','EMPTY_IN_WAREHOUSE','FILLED_IN_WAREHOUSE',"
"'AT_SITE_UNUSED','AT_SITE_PARTIAL','AT_SITE_EMPTY','RETURNING')"
),
'ck_biz_pallet_inventory_remaining_packages': 'remaining_packages>=0',
'ck_biz_pallet_inventory_version': 'version>=0',
'ck_biz_pallet_inventory_location': 'NOT(warehouse_idISNOTNULLANDsite_idISNOTNULL)',
},
'biz_pallet_event': {
'ck_biz_pallet_event_event_type': (
"event_typeIN('INBOUND_CONFIRMED','FILLING_CONFIRMED','SHIPMENT_CONFIRMED',"
"'USAGE_RECORDED','RETURN_REQUESTED','RETURN_CONFIRMED')"
),
},
}
for table_name, expected in expected_checks.items():
with self.subTest(table_name=table_name):
self.assertEqual(expected, _normalized_check_sql(Base.metadata.tables[table_name]))
def test_status_fields_use_portable_string_types(self) -> None:
char_statuses = {
'biz_strain': 'status',
'biz_warehouse': 'status',
'biz_customer': 'status',
'biz_customer_site': 'status',
'biz_production_line': 'status',
'biz_pallet': 'asset_status',
}
for table_name, column_name in char_statuses.items():
status_type = Base.metadata.tables[table_name].c[column_name].type
with self.subTest(table_name=table_name, column_name=column_name):
self.assertIs(type(status_type), CHAR)
self.assertEqual(1, status_type.length)
string_statuses = {
'biz_inbound_order': {'status'},
'biz_shipment_order': {'status'},
'biz_return_order': {'status'},
'biz_pallet_inventory': {'flow_state'},
'biz_pallet_event': {'event_type', 'before_state', 'after_state'},
}
for table_name, column_names in string_statuses.items():
for column_name in column_names:
status_type = Base.metadata.tables[table_name].c[column_name].type
with self.subTest(table_name=table_name, column_name=column_name):
self.assertIs(type(status_type), String)
self.assertEqual(32, status_type.length)
def test_package_counts_use_integer_and_nullable_snapshots_are_preserved(self) -> None:
integer_columns = {
'biz_pallet': {'capacity_packages'},
'biz_filling_record': {'initial_packages'},
'biz_shipment_item': {'package_qty'},
'biz_usage_record': {'used_packages', 'remaining_packages_after'},
'biz_return_item': {'remaining_packages_snapshot'},
'biz_pallet_inventory': {'remaining_packages', 'version'},
'biz_pallet_event': {'before_remaining_packages', 'after_remaining_packages'},
}
for table_name, column_names in integer_columns.items():
for column_name in column_names:
with self.subTest(table_name=table_name, column_name=column_name):
self.assertIs(type(Base.metadata.tables[table_name].c[column_name].type), Integer)
nullable_columns = {
'biz_inbound_order': {
'confirm_idempotency_key',
'confirm_request_hash',
'confirmed_by',
'confirmed_at',
},
'biz_filling_record': {'active_slot', 'closed_at'},
'biz_shipment_order': {
'confirm_idempotency_key',
'confirm_request_hash',
'confirmed_by',
'confirmed_at',
},
'biz_shipment_item': {'active_filling_id', 'package_qty'},
'biz_return_order': {
'confirm_idempotency_key',
'confirm_request_hash',
'confirmed_by',
'confirmed_at',
},
'biz_return_item': {'active_filling_id_snapshot'},
'biz_pallet_inventory': {
'active_filling_id',
'warehouse_id',
'site_id',
'current_line_id',
'last_event_id',
},
'biz_pallet_event': {
'before_warehouse_id',
'before_site_id',
'before_line_id',
'after_warehouse_id',
'after_site_id',
'after_line_id',
},
}
for table_name, column_names in nullable_columns.items():
for column_name in column_names:
with self.subTest(table_name=table_name, column_name=column_name):
self.assertTrue(Base.metadata.tables[table_name].c[column_name].nullable)
if __name__ == '__main__':
unittest.main()