修复：校正业务模型审计默认值

2026-07-24 20:45:10 +08:00
parent f814f916a5
commit eab725a541
3 changed files with 157 additions and 51 deletions
@@ -1,6 +1,8 @@
 import unittest

 from sqlalchemy import CHAR, BigInteger, CheckConstraint, DateTime, Integer, String, Table, UniqueConstraint
+from sqlalchemy.dialects import mysql, postgresql
+from sqlalchemy.schema import CreateColumn

 from config.database import Base
 from module_business.entity.do import master_do, operation_do
@@ -694,6 +696,63 @@ EXPECTED_NULLABILITY = {
    },
 }

+EXPECTED_CALLABLE_DATETIME_DEFAULTS = {
+    ('biz_strain', 'create_time'),
+    ('biz_strain', 'update_time'),
+    ('biz_warehouse', 'create_time'),
+    ('biz_warehouse', 'update_time'),
+    ('biz_customer', 'create_time'),
+    ('biz_customer', 'update_time'),
+    ('biz_customer_site', 'create_time'),
+    ('biz_customer_site', 'update_time'),
+    ('biz_production_line', 'create_time'),
+    ('biz_production_line', 'update_time'),
+    ('biz_pallet', 'create_time'),
+    ('biz_pallet', 'update_time'),
+    ('biz_user_warehouse', 'create_time'),
+    ('biz_user_site', 'create_time'),
+    ('biz_inbound_order', 'create_time'),
+    ('biz_inbound_order', 'update_time'),
+    ('biz_inbound_item', 'create_time'),
+    ('biz_filling_record', 'create_time'),
+    ('biz_filling_record', 'update_time'),
+    ('biz_shipment_order', 'create_time'),
+    ('biz_shipment_order', 'update_time'),
+    ('biz_shipment_item', 'create_time'),
+    ('biz_return_order', 'create_time'),
+    ('biz_return_order', 'update_time'),
+    ('biz_return_item', 'create_time'),
+    ('biz_pallet_inventory', 'updated_at'),
+}
+
+EXPECTED_EMPTY_AUDIT_SERVER_DEFAULTS = {
+    ('biz_strain', 'create_by'),
+    ('biz_strain', 'update_by'),
+    ('biz_warehouse', 'create_by'),
+    ('biz_warehouse', 'update_by'),
+    ('biz_customer', 'create_by'),
+    ('biz_customer', 'update_by'),
+    ('biz_customer_site', 'create_by'),
+    ('biz_customer_site', 'update_by'),
+    ('biz_production_line', 'create_by'),
+    ('biz_production_line', 'update_by'),
+    ('biz_pallet', 'create_by'),
+    ('biz_pallet', 'update_by'),
+    ('biz_user_warehouse', 'create_by'),
+    ('biz_user_site', 'create_by'),
+    ('biz_inbound_order', 'create_by'),
+    ('biz_inbound_order', 'update_by'),
+    ('biz_inbound_item', 'create_by'),
+    ('biz_filling_record', 'create_by'),
+    ('biz_filling_record', 'update_by'),
+    ('biz_shipment_order', 'create_by'),
+    ('biz_shipment_order', 'update_by'),
+    ('biz_shipment_item', 'create_by'),
+    ('biz_return_order', 'create_by'),
+    ('biz_return_order', 'update_by'),
+    ('biz_return_item', 'create_by'),
+}
+
 EXPECTED_PRIMARY_KEYS = {
    'biz_strain': ('strain_id',),
    'biz_warehouse': ('warehouse_id',),
@@ -930,6 +989,53 @@ class BusinessModelContractTest(unittest.TestCase):
                with self.subTest(table_name=table_name, column_name=column_name):
                    self.assertIs(expected_nullable, table.c[column_name].nullable)

+    def test_datetime_python_defaults_are_callable(self) -> None:
+        business_tables = [table for table in Base.metadata.tables.values() if table.name.startswith('biz_')]
+        actual_defaults = {
+            (table.name, column.name)
+            for table in business_tables
+            for column in table.columns
+            if type(column.type) is DateTime and column.default is not None
+        }
+        self.assertEqual(EXPECTED_CALLABLE_DATETIME_DEFAULTS, actual_defaults)
+
+        violations = []
+        for table_name, column_name in sorted(EXPECTED_CALLABLE_DATETIME_DEFAULTS):
+            column_default = Base.metadata.tables[table_name].c[column_name].default
+            if column_default is None or not column_default.is_callable or column_default.is_scalar:
+                violations.append(
+                    (
+                        table_name,
+                        column_name,
+                        None if column_default is None else column_default.is_callable,
+                        None if column_default is None else column_default.is_scalar,
+                    )
+                )
+        self.assertEqual([], violations)
+
+    def test_audit_user_server_defaults_compile_to_empty_strings(self) -> None:
+        business_tables = [table for table in Base.metadata.tables.values() if table.name.startswith('biz_')]
+        actual_audit_columns = {
+            (table.name, column.name)
+            for table in business_tables
+            for column in table.columns
+            if column.name in {'create_by', 'update_by'}
+        }
+        self.assertEqual(EXPECTED_EMPTY_AUDIT_SERVER_DEFAULTS, actual_audit_columns)
+
+        violations = []
+        dialects = {'mysql': mysql.dialect(), 'postgresql': postgresql.dialect()}
+        for table_name, column_name in sorted(EXPECTED_EMPTY_AUDIT_SERVER_DEFAULTS):
+            column = Base.metadata.tables[table_name].c[column_name]
+            for dialect_name, dialect in dialects.items():
+                compiled_column = str(CreateColumn(column).compile(dialect=dialect))
+                _, separator, compiled_default = compiled_column.partition(' DEFAULT ')
+                if separator:
+                    compiled_default = compiled_default.split(' COMMENT ', maxsplit=1)[0].strip()
+                if compiled_default != "''":
+                    violations.append((dialect_name, table_name, column_name, compiled_default))
+        self.assertEqual([], violations)
+
    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]