niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调整部分前端布局和后端增删改逻辑

Browse Source
This commit is contained in:
焦龙言
2026-09-26 23:11:05 +08:00
parent 977e7f996e
commit 17d991a74a
117 changed files with 311994 additions and 3192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py
+18 -25
View File
@@ -5,7 +5,7 @@ from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from sqlalchemy.dialects import mysql, postgresql
from sqlalchemy.dialects import mysql
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import operators
@@ -39,7 +39,7 @@ from utils.page_util import PageUtil
def compile_sql(statement: Any, dialect: Any | None = None) -> str:
compiled = statement.compile(
dialect=dialect or postgresql.dialect(),
dialect=dialect or mysql.dialect(),
compile_kwargs={'literal_binds': True},
)
return ' '.join(str(compiled).split()).replace('%%', '%')
@@ -289,7 +289,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
sql = compile_sql(statement, dialect)
self.assertIn('biz_pallet_inventory.pallet_id = 7', sql)
self.assertIn('FOR UPDATE', sql)
@@ -310,7 +310,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletInventory.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
sql = compile_sql(statement, dialect)
self.assertIn('biz_pallet_inventory.pallet_id IN (2, 5, 9)', sql)
self.assertIn('ORDER BY biz_pallet_inventory.pallet_id', sql)
@@ -353,7 +353,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(BizPallet.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
sql = compile_sql(statement, dialect)
self.assertIn('biz_pallet.pallet_id IN (2, 5, 9)', sql)
self.assertIn('ORDER BY biz_pallet.pallet_id', sql)
@@ -412,7 +412,7 @@ class TestInventoryDao(unittest.IsolatedAsyncioTestCase):
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.selected_columns[0].compare(BizPalletEvent.event_seq.property.columns[0]))
self.assertTrue(statement.get_final_froms()[0].compare(BizPalletEvent.__table__))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
sql = compile_sql(statement, dialect)
self.assertNotIn('max(', sql.lower())
self.assertIn('biz_pallet_event.event_id = 40', sql)
@@ -483,7 +483,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
statement = self.db.execute.await_args.args[0]
self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
self.assertIs(True, statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
sql = compile_sql(statement, dialect)
self.assertIn(f'{model.__tablename__}.{id_field} = {record_id}', sql)
self.assertIn('FOR UPDATE', sql)
@@ -519,7 +519,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertTrue(item_statement.get_final_froms()[0].compare(BizShipmentItem.__table__))
self.assertIs(True, parent_statement.get_execution_options().get('populate_existing'))
self.assertIs(True, item_statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
parent_sql = compile_sql(parent_statement, dialect)
item_sql = compile_sql(item_statement, dialect)
@@ -551,9 +551,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.db.execute.assert_awaited_once()
parent_statement = self.db.execute.await_args.args[0]
mysql_sql = compile_sql(parent_statement, mysql.dialect())
postgresql_sql = compile_sql(parent_statement, postgresql.dialect())
self.assertIn('false = 1', mysql_sql)
self.assertIn('false', postgresql_sql)
async def test_inbound_for_update_locks_parent_then_current_items(self) -> None:
inbound = SimpleNamespace(inbound_id=11)
@@ -578,7 +576,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertTrue(item_statement.get_final_froms()[0].compare(BizInboundItem.__table__))
self.assertIs(True, parent_statement.get_execution_options().get('populate_existing'))
self.assertIs(True, item_statement.get_execution_options().get('populate_existing'))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
parent_sql = compile_sql(parent_statement, dialect)
item_sql = compile_sql(item_statement, dialect)
@@ -610,7 +608,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIsNone(actual)
self.db.execute.assert_awaited_once()
parent_statement = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
sql = compile_sql(parent_statement, dialect)
self.assertIn('biz_inbound_order.warehouse_id IN (3, 7)', sql)
@@ -684,7 +682,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIsNone(actual)
statement = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
sql = compile_sql(statement, dialect)
self.assertIn("biz_inbound_order.confirm_idempotency_key = 'confirm-key'", sql)
@@ -850,7 +848,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIs(marker, actual)
statement = paginate.await_args.args[1]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
sql = compile_sql(statement, dialect)
self.assertIn('biz_inbound_order.warehouse_id IN (2, 5)', sql)
@@ -871,9 +869,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIs(marker, actual)
statement = paginate.await_args.args[1]
mysql_sql = compile_sql(statement, mysql.dialect())
postgresql_sql = compile_sql(statement, postgresql.dialect())
self.assertIn('WHERE false = 1', mysql_sql)
self.assertIn('WHERE false', postgresql_sql)
self.assertEqual((query.page_num, query.page_size, True), paginate.await_args.args[2:])
async def test_inbound_list_none_scope_has_no_authorization_predicate(self) -> None:
@@ -906,7 +902,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.assertIs(marker, actual)
statement = paginate.await_args.args[1]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
sql = compile_sql(statement, dialect)
self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', sql)
@@ -922,7 +918,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
)
empty_statement = empty_paginate.await_args.args[1]
self.assertIn('false = 1', compile_sql(empty_statement, mysql.dialect()))
self.assertIn('false', compile_sql(empty_statement, postgresql.dialect()))
async def test_shipment_detail_and_confirm_key_push_both_scopes_into_parent_sql(self) -> None:
parent_result = MagicMock()
@@ -953,7 +948,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
)
)
key_statement = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
detail_sql = compile_sql(detail_statement, dialect)
key_sql = compile_sql(key_statement, dialect)
@@ -994,7 +989,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
)
key_statement = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=dialect.name):
for statement in (list_statement, detail_statement, key_statement):
sql = compile_sql(statement, dialect)
@@ -1016,7 +1011,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
)
empty_key = self.db.execute.await_args.args[0]
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
with self.subTest(dialect=f'{dialect.name}-empty'):
for statement in (empty_list, empty_detail, empty_key):
sql = compile_sql(statement, dialect)
@@ -1055,7 +1050,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
statement = self.db.execute.await_args.args[0]
self.assertFalse(statement.get_execution_options().get('populate_existing', False))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
sql = compile_sql(statement, dialect)
self.assertNotIn('FOR UPDATE', sql)
if expected_scope is not None:
@@ -1154,9 +1149,7 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
self.db.execute.assert_awaited_once()
parent_statement = self.db.execute.await_args.args[0]
mysql_sql = compile_sql(parent_statement, mysql.dialect())
postgresql_sql = compile_sql(parent_statement, postgresql.dialect())
self.assertIn('false = 1', mysql_sql)
self.assertIn('false', postgresql_sql)
async def test_simple_details_target_primary_keys(self) -> None:
cases = (
@@ -1315,7 +1308,7 @@ class TestOperationDaoWrites(unittest.IsolatedAsyncioTestCase):
order_statement = self.db.execute.await_args_list[1].args[0]
self.assertTrue(item_statement.table.compare(item_model.__table__))
self.assertTrue(order_statement.table.compare(order_model.__table__))
for dialect in (mysql.dialect(), postgresql.dialect()):
for dialect in (mysql.dialect(),):
self.assertIn(
f'{item_model.__tablename__}.{id_field} = {record_id}',
compile_sql(item_statement, dialect),