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-27 04:51:29 +08:00
parent 33f29df40e
commit 246e69f5c8
1 changed files with 57 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_master_service.py
+57 -25
View File
@@ -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):