126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""OA 审批应用列表：避免按应用 N+1 拉取版本大字段。"""
|
|
import json
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
|
|
class TestListAppsPerf(unittest.TestCase):
|
|
def setUp(self):
|
|
from app.models.models import OaApp, OaAppGroup, OaAppVersion
|
|
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
OaAppGroup.__table__.create(self.engine)
|
|
OaApp.__table__.create(self.engine)
|
|
OaAppVersion.__table__.create(self.engine)
|
|
Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
|
|
self.db = Session()
|
|
self.OaApp = OaApp
|
|
self.OaAppGroup = OaAppGroup
|
|
self.OaAppVersion = OaAppVersion
|
|
self.now = datetime.now()
|
|
self.query_count = 0
|
|
|
|
@event.listens_for(self.engine, "before_cursor_execute")
|
|
def _count(_conn, _cursor, _statement, _parameters, _context, _executemany):
|
|
self.query_count += 1
|
|
|
|
self._listener = _count
|
|
|
|
def tearDown(self):
|
|
event.remove(self.engine, "before_cursor_execute", self._listener)
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def _seed_apps(self, n: int = 8):
|
|
bulky = json.dumps({"rule": [{"x": "y" * 2000}] * 50}, ensure_ascii=False)
|
|
self.db.add(
|
|
self.OaAppGroup(
|
|
id="g1",
|
|
group_name="人事",
|
|
tenant_no="zkxw",
|
|
create_dept_id="",
|
|
create_time=self.now,
|
|
update_time=self.now,
|
|
del_flag=0,
|
|
)
|
|
)
|
|
for i in range(n):
|
|
app_id = f"app{i}"
|
|
self.db.add(
|
|
self.OaApp(
|
|
id=app_id,
|
|
group_id="g1",
|
|
app_name=f"应用{i}",
|
|
initiator_scope="all",
|
|
admin_scope="all_oa_admin",
|
|
status="1",
|
|
tenant_no="zkxw",
|
|
order_num=i,
|
|
create_dept_id="",
|
|
create_time=self.now,
|
|
update_time=self.now,
|
|
del_flag=0,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.OaAppVersion(
|
|
id=f"draft{i}",
|
|
app_id=app_id,
|
|
version_no=0,
|
|
status="draft",
|
|
is_latest_published="0",
|
|
form_schema_json=bulky,
|
|
flow_tree_json=bulky,
|
|
advanced_settings_json="{}",
|
|
tenant_no="zkxw",
|
|
create_dept_id="",
|
|
create_time=self.now,
|
|
update_time=self.now,
|
|
del_flag=0,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.OaAppVersion(
|
|
id=f"pub{i}",
|
|
app_id=app_id,
|
|
version_no=i + 1,
|
|
status="published",
|
|
is_latest_published="1",
|
|
form_schema_json=bulky,
|
|
flow_tree_json=bulky,
|
|
advanced_settings_json="{}",
|
|
tenant_no="zkxw",
|
|
create_dept_id="",
|
|
create_time=self.now,
|
|
update_time=self.now,
|
|
del_flag=0,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
def test_list_apps_query_count_bounded_and_fields_correct(self):
|
|
from app.crud.oa_approval import list_apps
|
|
|
|
n = 8
|
|
self._seed_apps(n)
|
|
self.query_count = 0
|
|
items = list_apps(self.db, tenant_no="zkxw")
|
|
|
|
self.assertEqual(len(items), n)
|
|
self.assertTrue(all(it["hasDraft"] for it in items))
|
|
self.assertEqual(
|
|
{it["id"]: it["latestPublishedVersion"] for it in items},
|
|
{f"app{i}": i + 1 for i in range(n)},
|
|
)
|
|
self.assertEqual(items[0]["groupName"], "人事")
|
|
# apps + groups + drafts batch + published batch（不应随应用数线性增长）
|
|
self.assertLessEqual(self.query_count, 5)
|
|
self.assertLess(self.query_count, 2 + 2 * n)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|