From 246e69f5c88f31725dc3d126f7871843abf91c5f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Mon, 27 Jul 2026 04:51:29 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=BC=BA=E4=B8=BB=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=8F=AF=E5=88=A0=E9=99=A4=E6=80=A7=E6=9F=A5=E8=AF=A2=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/test_master_service.py | 82 +++++++++++++------ 1 file changed, 57 insertions(+), 25 deletions(-) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py index 61b2eed..a4ae7e6 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py @@ -1,4 +1,5 @@ import inspect +import re import unittest from contextlib import ExitStack from datetime import datetime @@ -8,6 +9,7 @@ from unittest.mock import AsyncMock, MagicMock, patch from fastapi import FastAPI from fastapi.routing import APIRoute +from sqlalchemy.dialects import mysql, postgresql from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -753,36 +755,48 @@ class TestMasterDao(unittest.IsolatedAsyncioTestCase): async def test_master_removability_uses_the_same_reference_constraints_as_delete(self) -> None: expected = { - 'strain': (('biz_filling_record', ('strain_id',)),), - 'customer': (('biz_customer_site', ('customer_id',)),), + 'strain': ('biz_strain', 'strain_id', (('biz_filling_record', ('strain_id',)),)), + 'customer': ('biz_customer', 'customer_id', (('biz_customer_site', ('customer_id',)),)), 'site': ( - ('biz_production_line', ('site_id',)), - ('biz_user_site', ('site_id',)), - ('biz_shipment_order', ('target_site_id',)), - ('biz_usage_record', ('site_id',)), - ('biz_return_order', ('source_site_id',)), - ('biz_pallet_inventory', ('site_id',)), - ('biz_pallet_event', ('before_site_id', 'after_site_id')), + 'biz_customer_site', + 'site_id', + ( + ('biz_production_line', ('site_id',)), + ('biz_user_site', ('site_id',)), + ('biz_shipment_order', ('target_site_id',)), + ('biz_usage_record', ('site_id',)), + ('biz_return_order', ('source_site_id',)), + ('biz_pallet_inventory', ('site_id',)), + ('biz_pallet_event', ('before_site_id', 'after_site_id')), + ), ), 'line': ( - ('biz_usage_record', ('line_id',)), - ('biz_pallet_inventory', ('current_line_id',)), - ('biz_pallet_event', ('before_line_id', 'after_line_id')), + 'biz_production_line', + 'line_id', + ( + ('biz_usage_record', ('line_id',)), + ('biz_pallet_inventory', ('current_line_id',)), + ('biz_pallet_event', ('before_line_id', 'after_line_id')), + ), ), 'warehouse': ( - ('biz_user_warehouse', ('warehouse_id',)), - ('biz_inbound_order', ('warehouse_id',)), - ('biz_filling_record', ('warehouse_id',)), - ('biz_shipment_order', ('source_warehouse_id',)), - ('biz_return_order', ('target_warehouse_id',)), - ('biz_pallet_inventory', ('warehouse_id',)), - ('biz_pallet_event', ('before_warehouse_id', 'after_warehouse_id')), + 'biz_warehouse', + 'warehouse_id', + ( + ('biz_user_warehouse', ('warehouse_id',)), + ('biz_inbound_order', ('warehouse_id',)), + ('biz_filling_record', ('warehouse_id',)), + ('biz_shipment_order', ('source_warehouse_id',)), + ('biz_return_order', ('target_warehouse_id',)), + ('biz_pallet_inventory', ('warehouse_id',)), + ('biz_pallet_event', ('before_warehouse_id', 'after_warehouse_id')), + ), ), } method = getattr(MasterDao, 'get_master_removability', None) self.assertIsNotNone(method) - for resource, references in expected.items(): + for resource, (outer_table, outer_id, references) in expected.items(): result = MagicMock() result.all.return_value = [(1, True), (2, False)] self.db.execute = AsyncMock(return_value=result) @@ -790,11 +804,29 @@ class TestMasterDao(unittest.IsolatedAsyncioTestCase): removability = await method(self.db, resource, [1, 2, 3]) self.assertEqual({1: True, 2: False, 3: False}, removability) - sql = compile_sql(self.db.execute.await_args.args[0]) - for table, columns in references: - self.assertIn(f'FROM {table}', sql) - for column in columns: - self.assertIn(f'{table}.{column}', sql) + statement = self.db.execute.await_args.args[0] + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(resource=resource, dialect=dialect.name): + sql = ' '.join( + str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).split() + ) + self.assertEqual(len(references), len(re.findall(r'NOT\s+\(?EXISTS\s+\(', sql))) + for table, columns in references: + self.assertRegex( + sql, + rf'NOT\s+\(?EXISTS\s+\(SELECT {table}\.{columns[0]} FROM {table} WHERE ', + ) + for column in columns: + self.assertIn(f'{table}.{column} = {outer_table}.{outer_id}', sql) + + async def test_master_removability_short_circuits_empty_resource_ids(self) -> None: + self.db.execute = AsyncMock() + + for resource in RESOURCE_IDS: + with self.subTest(resource=resource): + self.assertEqual({}, await MasterDao.get_master_removability(self.db, resource, [])) + + self.db.execute.assert_not_awaited() class TestMasterControllers(unittest.TestCase):