修复:保留 PostgreSQL 时间微秒精度

This commit is contained in:
焦龙言
2026-07-27 13:33:08 +08:00
parent b858340890
commit 7a786800bd
2 changed files with 85 additions and 34 deletions
@@ -165,6 +165,27 @@ EXPECTED_COLUMNS = {
},
}
EXPECTED_DATETIME_COLUMNS = {
'biz_strain': {'create_time', 'update_time'},
'biz_warehouse': {'create_time', 'update_time'},
'biz_customer': {'create_time', 'update_time'},
'biz_customer_site': {'create_time', 'update_time'},
'biz_production_line': {'create_time', 'update_time'},
'biz_pallet': {'create_time', 'update_time'},
'biz_user_warehouse': {'create_time'},
'biz_user_site': {'create_time'},
'biz_inbound_order': {'confirmed_at', 'create_time', 'update_time'},
'biz_inbound_item': {'create_time'},
'biz_filling_record': {'closed_at', 'operated_at', 'create_time', 'update_time'},
'biz_shipment_order': {'confirmed_at', 'create_time', 'update_time'},
'biz_shipment_item': {'create_time'},
'biz_usage_record': {'occurred_at'},
'biz_return_order': {'requested_at', 'confirmed_at', 'create_time', 'update_time'},
'biz_return_item': {'create_time'},
'biz_pallet_inventory': {'updated_at'},
'biz_pallet_event': {'occurred_at'},
}
EXPECTED_PRIMARY_KEYS = {
'biz_strain': ('strain_id',),
'biz_warehouse': ('warehouse_id',),
@@ -577,6 +598,17 @@ def _table_contract(
return columns, primary_key, unique_keys, foreign_keys, checks
def _column_definitions(body: str) -> dict[str, str]:
definitions = {}
reserved = {'constraint', 'primary', 'unique', 'foreign', 'check', 'key', 'index'}
for clause in _split_values(body):
clean = clause.strip()
identifier_match = re.match(r'[`"]?([a-z_][a-z0-9_]*)[`"]?', clean, flags=re.I)
if identifier_match and identifier_match.group(1).lower() not in reserved:
definitions[identifier_match.group(1).lower()] = clean
return definitions
def _business_menu_rows(sql: str) -> dict[int, list]:
rows = _insert_rows(sql, 'sys_menu')
selected = [row for row in rows if row and str(row[0]).isdigit() and int(row[0]) in BUSINESS_MENU_IDS]
@@ -607,6 +639,25 @@ class SqlContractTest(unittest.TestCase):
self.assertEqual(EXPECTED_FOREIGN_KEYS[table], foreign_keys)
self.assertEqual(EXPECTED_CHECKS[table], checks)
def test_postgres_business_datetimes_preserve_default_microsecond_precision(self) -> None:
self.assertEqual(34, sum(len(columns) for columns in EXPECTED_DATETIME_COLUMNS.values()))
tables = _create_table_bodies((SQL_ROOT / 'shierfeng-fastapi-pg.sql').read_text(encoding='utf-8'))
zero_precision_columns = []
for table, expected_columns in EXPECTED_DATETIME_COLUMNS.items():
definitions = _column_definitions(tables[table])
actual_columns = {
column
for column, definition in definitions.items()
if re.search(r'\btimestamp\b', definition, flags=re.I)
}
self.assertEqual(expected_columns, actual_columns, table)
for column in expected_columns:
definition = definitions[column]
self.assertRegex(definition, r'\btimestamp(?:\s+without\s+time\s+zone)?\b')
if re.search(r'\btimestamp\s*\(\s*0\s*\)', definition, flags=re.I):
zero_precision_columns.append(f'{table}.{column}')
self.assertEqual([], sorted(zero_precision_columns))
def test_initializers_define_fixed_business_menus_and_permissions(self) -> None:
expected_permissions = {page[-1] for page in EXPECTED_PAGES.values()} | {
button[-1] for button in EXPECTED_BUTTONS.values()