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):