zkxw/backend/tests/test_oa_list_apps_perf.py

"""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()