hyoban · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026
@@ -1,3 +1,4 @@
+import gc
 import os
 from collections.abc import Iterator
 
@@ -10,6 +11,17 @@
 os.environ.setdefault("HUMAN_INPUT_IM_EVENT_TRANSPORT_MODE", "WEBHOOK")
 
 
+def pytest_collection_finish(session: pytest.Session) -> None:
+    """Move everything loaded during collection out of the garbage collector's reach.
+
+    Collection imports the whole application, so every automatic full collection afterwards
+    rescanned those long-lived objects and stalled whichever test triggered it. Objects created
+    by tests themselves are still collected normally.
+    """
+    gc.collect()
+    gc.freeze()
+
+
 @pytest.fixture(autouse=True)
 def _bind_workflow_file_runtime() -> Iterator[None]:
     with use_workflow_file_runtime(create_dify_workflow_file_runtime()):

@@ -56,7 +56,7 @@ def do_HEAD(self) -> None:
     http_pool = http_client_pooling.HttpClientPoolFactory()
     monkeypatch.setattr(http_client_pooling, "_factory", http_pool)
     server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
-    thread = Thread(target=server.serve_forever, daemon=True)
+    thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     thread.start()
     try:
         yield f"http://127.0.0.1:{server.server_port}", requests

@@ -95,7 +95,7 @@ def respond(self, data: object, *, stream: bool, code: int = 0, message: str = "
             self.wfile.write(body)
 
     server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
-    thread = Thread(target=server.serve_forever, daemon=True)
+    thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     original_url = plugin_http.plugin_daemon_inner_api_baseurl
     # Change real connection settings, preserving every production client and method.
     plugin_http.plugin_daemon_inner_api_baseurl = URL(f"http://127.0.0.1:{server.server_port}")

@@ -5,8 +5,10 @@
 from collections.abc import Callable, Iterator
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from threading import Lock
 from typing import TYPE_CHECKING
 from unittest.mock import MagicMock, create_autospec, patch
+from weakref import WeakKeyDictionary
 
 # Ensure OpenDAL fs writes to tmp to avoid polluting workspace
 os.environ.setdefault("OPENDAL_SCHEME", "fs")
@@ -17,7 +19,7 @@
 from dify_trace_sdk.testing import RecordingEgress
 from flask import Flask
 from redis import Redis
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, event
 from sqlalchemy.engine import URL, Engine
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -132,7 +134,10 @@ def _patch_redis_clients() -> Iterator[None]:
 @pytest.fixture(autouse=True)
 def reset_redis_mock(_patch_redis_clients: None) -> None:
     """Reset the shared Redis mock after per-test client rebinding."""
-    redis_mock.reset_mock()
+    # ``mock_calls`` records calls on every attached child and return value, so when it is empty the
+    # recursive reset has nothing to clear. Most tests never touch Redis; skip walking ~370 child mocks.
+    if redis_mock.mock_calls:
+        redis_mock.reset_mock()
     # Restoring a monkeypatched method can leave it detached from the parent's reset traversal.
     redis_mock.delete.reset_mock()
     redis_mock.get.reset_mock()
@@ -178,19 +183,43 @@ def apply(**values: object) -> None:
     return apply
 
 
+# The callable that materializes each live ``_sqlite_engine``'s database file.
+_sqlite_copiers: WeakKeyDictionary[Engine, Callable[[], None]] = WeakKeyDictionary()
+
+
+def _copy_sqlite_template(template: Path, database_path: Path, lock: Lock) -> None:
+    """Create ``database_path`` as a copy of ``template`` unless it already exists.
+
+    Idempotent and safe across threads sharing ``lock``: the copy happens at most once, so an
+    open connection never sees its file overwritten.
+    """
+    with lock:
+        if not database_path.exists():
+            shutil.copyfile(template, database_path)
+
+
 @pytest.fixture
 def _sqlite_engine(_sqlite_database_template: Path) -> Iterator[Engine]:
     """Copy the schema into an isolated directory without pytest's numbered scan.
 
     ``tmp_path`` searches all preceding test directories for a free number. This
     autouse dependency needs only a unique, disposable directory, including any
     SQLite journal files, so keep it separate from test-owned ``tmp_path`` data.
+
+    Most tests never connect, so the schema is copied on the engine's first connection
+    rather than for every test. Code that opens the file without this engine must go
+    through ``sqlite_engine``, which copies it up front.
     """
     with TemporaryDirectory(prefix="case-", dir=_sqlite_database_template.parent) as directory:
         database_path = Path(directory) / "unit-tests.sqlite3"
-        shutil.copyfile(_sqlite_database_template, database_path)
         engine = create_engine(URL.create("sqlite", database=str(database_path)))
+        lock = Lock()
+
+        def copy() -> None:
+            _copy_sqlite_template(_sqlite_database_template, database_path, lock)
 
+        event.listen(engine, "do_connect", lambda *_args: copy())
+        _sqlite_copiers[engine] = copy
         try:
             yield engine
         finally:
@@ -236,8 +265,13 @@ def _unbound_session_factory(
 
 @pytest.fixture
 def sqlite_engine(_sqlite_engine: Engine) -> Engine:
-    """Expose the pristine full-schema SQLite engine to tests."""
+    """Expose the pristine full-schema SQLite engine to tests.
+
+    The database file exists on return, so tests may hand ``sqlite_engine.url`` to code
+    that opens it with an engine of its own.
+    """
 
+    _sqlite_copiers[_sqlite_engine]()
     return _sqlite_engine
 
 

@@ -320,10 +320,25 @@ class AdmittedAccount:
     status: str
 
 
+@pytest.fixture(scope="module")
+def console_app() -> Flask:
+    """One app with the console namespace, which takes a few hundred milliseconds to mount.
+
+    Each test registers its own ``extensions["application_services"]`` for the duration of that test.
+    """
+    app = Flask(__name__)
+    app.config.update(TESTING=False, PROPAGATE_EXCEPTIONS=False)
+    api = ExternalApi(app)
+    api.error_handlers = console_api.error_handlers.copy()
+    api.add_namespace(console_ns, path="/console/api")
+    return app
+
+
 class TestDataSourceCases:
     @staticmethod
     @pytest.fixture
     def collaborators(
+        console_app: Flask,
         config_overrides: Callable[..., None],
         monkeypatch: pytest.MonkeyPatch,
     ) -> ControllerCollaborators:
@@ -333,12 +348,6 @@ def collaborators(
             RBAC_ENABLED=False,
             CONSOLE_API_URL="https://console-api.example.com",
         )
-        app = Flask(__name__)
-        app.config.update(TESTING=False, PROPAGATE_EXCEPTIONS=False)
-        api = ExternalApi(app)
-        api.error_handlers = console_api.error_handlers.copy()
-        api.add_namespace(console_ns, path="/console/api")
-
         account = AdmittedAccount(id=ACCOUNT_ID, role="owner", status="active")
         account_with_tenant = AccountWithTenant(account=cast("Account", account), tenant_id=TENANT_ID)
         monkeypatch.setattr(
@@ -393,7 +402,7 @@ def collaborators(
         )
         estimates = RecordingEstimateService()
         dispatcher = RecordingDispatcher()
-        app.extensions["application_services"] = ApplicationServiceRegistry(
+        services = ApplicationServiceRegistry(
             data_sources=DataSourceApplicationServices(
                 bindings=DataSourceBindingApplicationService(bindings=bindings),
                 notion_imports=NotionImportApplicationService(
@@ -412,8 +421,9 @@ def collaborators(
                 indexing_estimates=estimates,
             ),
         )
+        monkeypatch.setitem(console_app.extensions, "application_services", services)
         return ControllerCollaborators(
-            client=app.test_client(),
+            client=console_app.test_client(),
             account=account,
             bindings=bindings,
             notion_source=notion_source,

@@ -50,21 +50,32 @@ def current_account(sqlite_session_factory: sessionmaker[Session]) -> Account:
     return account
 
 
-@pytest.fixture(params=[False, True], ids=["legacy-roles", "rbac"])
-def client(current_account: Account, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> FlaskClient:
-    monkeypatch.setattr(dify_config, "DEPLOYMENT_EDITION", DeploymentEdition.CLOUD)
-    monkeypatch.setattr(dify_config, "LOGIN_DISABLED", True)
-    monkeypatch.setattr(dify_config, "RBAC_ENABLED", request.param)
+@pytest.fixture(scope="module")
+def console_app() -> Flask:
+    """One app with the whole console blueprint, which takes a few hundred milliseconds to register.
 
+    Each request logs in the account the running test stored under ``extensions["test_login_user"]``.
+    """
     app = Flask(__name__)
     LoginManager().init_app(app)
     app.register_blueprint(console_bp)
 
     @app.before_request
     def authenticate() -> None:
-        g._login_user = current_account
+        g._login_user = app.extensions["test_login_user"]
+
+    return app
 
-    return app.test_client()
+
+@pytest.fixture(params=[False, True], ids=["legacy-roles", "rbac"])
+def client(
+    console_app: Flask, current_account: Account, request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch
+) -> FlaskClient:
+    monkeypatch.setattr(dify_config, "DEPLOYMENT_EDITION", DeploymentEdition.CLOUD)
+    monkeypatch.setattr(dify_config, "LOGIN_DISABLED", True)
+    monkeypatch.setattr(dify_config, "RBAC_ENABLED", request.param)
+    monkeypatch.setitem(console_app.extensions, "test_login_user", current_account)
+    return console_app.test_client()
 
 
 @pytest.mark.parametrize(

@@ -50,8 +50,19 @@
 }
 
 
+@pytest.fixture(scope="module")
+def console_app() -> Flask:
+    """One app with the whole console blueprint, which takes a few hundred milliseconds to register."""
+    application = Flask(__name__)
+    LoginManager().init_app(application)
+    application.config["TESTING"] = False
+    application.config["RESTX_ERROR_404_HELP"] = False
+    application.register_blueprint(console_bp)
+    return application
+
+
 @pytest.fixture
-def wire_app(monkeypatch: pytest.MonkeyPatch) -> Flask:
+def wire_app(console_app: Flask, monkeypatch: pytest.MonkeyPatch) -> Flask:
     account = Account(name="Owner", email="owner@example.com", status=AccountStatus.ACTIVE)
     account.id = "account-1"
     account.role = TenantAccountRole.OWNER
@@ -64,13 +75,7 @@ def wire_app(monkeypatch: pytest.MonkeyPatch) -> Flask:
         "controllers.console.wraps.current_account_with_tenant",
         lambda: AccountWithTenant(account=account, tenant_id="workspace-1"),
     )
-
-    application = Flask(__name__)
-    LoginManager().init_app(application)
-    application.config["TESTING"] = False
-    application.config["RESTX_ERROR_404_HELP"] = False
-    application.register_blueprint(console_bp)
-    return application
+    return console_app
 
 
 def test_console_blueprint_registers_only_the_canonical_channel_routes(wire_app: Flask) -> None:

@@ -56,8 +56,34 @@ class ConsoleForm:
     actor: Account | None
 
 
+@pytest.fixture(scope="module")
+def console_app() -> DifyApp:
+    """One app with the console namespace, which takes a few hundred milliseconds to mount.
+
+    Each request logs in the actor of the scenario the running test stored under
+    ``extensions["test_console_form"]``, read per request so tests can change it.
+    """
+    app = DifyApp(__name__)
+    manager = DifyLoginManager(app)
+
+    @manager.unauthorized_handler
+    def unauthorized() -> Response:
+        return Response(
+            '{"code":"unauthorized","message":"Login required","status":401}', status=401, mimetype="application/json"
+        )
+
+    @app.before_request
+    def load_session_actor() -> None:
+        g._login_user = app.extensions["test_console_form"].actor
+
+    ExternalApi(app).add_namespace(controller.console_ns, path="/console/api")
+    return app
+
+
 @pytest.fixture
-def console_form(sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> ConsoleForm:
+def console_form(
+    console_app: DifyApp, sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch
+) -> ConsoleForm:
     tenant = make_tenant(tenant_id=str(uuid4()))
     active_tenant = make_tenant(tenant_id=str(uuid4()))
     account = make_account(account_id=str(uuid4()), tenant=active_tenant)
@@ -113,22 +139,10 @@ def console_form(sqlite_session_factory: sessionmaker[Session], monkeypatch: pyt
     monkeypatch.setattr("services.human_input_v2.submission_service.naive_utc_now", lambda: _NOW)
     # Isolate session decoding and CSRF; keep the route's login and Contact authorization real.
     monkeypatch.setattr("libs.login.check_csrf_token", lambda _request, _user_id: None)
-    app = DifyApp(__name__)
-    manager = DifyLoginManager(app)
-
-    @manager.unauthorized_handler
-    def unauthorized() -> Response:
-        return Response(
-            '{"code":"unauthorized","message":"Login required","status":401}', status=401, mimetype="application/json"
-        )
-
-    scenario = ConsoleForm(app.test_client(), sqlite_session_factory, form, recipient, ContactId(identity.id), account)
-
-    @app.before_request
-    def load_session_actor() -> None:
-        g._login_user = scenario.actor
-
-    ExternalApi(app).add_namespace(controller.console_ns, path="/console/api")
+    scenario = ConsoleForm(
+        console_app.test_client(), sqlite_session_factory, form, recipient, ContactId(identity.id), account
+    )
+    monkeypatch.setitem(console_app.extensions, "test_console_form", scenario)
     return scenario
 
 

@@ -100,8 +100,21 @@ def submission(self) -> dict[str, JsonValue] | None:
             return json.loads(form.raw_submission_inputs) if form.raw_submission_inputs is not None else None
 
 
+@pytest.fixture(scope="module")
+def service_app() -> Flask:
+    """One app with the service API namespace, which takes a few hundred milliseconds to mount.
+
+    Each test binds its own SQLite engine to the app for the duration of that test.
+    """
+    application_http = Flask(__name__)
+    LoginManager().init_app(application_http)
+    ExternalApi(application_http, register_default_root=False).add_namespace(controller.service_api_ns, path="/v1")
+    return application_http
+
+
 @pytest.fixture(params=["1", "2"])
 def approval(
+    service_app: Flask,
     sqlite_session_factory: sessionmaker[Session],
     sqlite_engine: Engine,
     monkeypatch: pytest.MonkeyPatch,
@@ -270,12 +283,9 @@ def publish(*, kwargs: dict[str, object]) -> None:
         published.append(kwargs)
 
     monkeypatch.setattr(resume_app_execution, "apply_async", publish)
-    application_http = Flask(__name__)
-    database._app_engines[application_http] = {None: sqlite_engine}
-    LoginManager().init_app(application_http)
-    ExternalApi(application_http, register_default_root=False).add_namespace(controller.service_api_ns, path="/v1")
+    monkeypatch.setitem(database._app_engines, service_app, {None: sqlite_engine})
     yield Approval(
-        application_http.test_client(),
+        service_app.test_client(),
         sqlite_session_factory,
         version,
         tenant.id,

@@ -588,7 +588,7 @@ def sdk_server(monkeypatch: pytest.MonkeyPatch) -> Generator[_RunningServer]:
     state = _ServerState()
     server = _SDKHTTPServer(("127.0.0.1", 0), _SDKRequestHandler)
     server.state = state
-    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
     thread.start()
     try:
         host, port = server.server_address[:2]

@@ -184,7 +184,7 @@ def connector_server() -> Generator[_ConnectorServer]:
     state = _ConnectorState()
     server = _ConnectorHTTPServer(("127.0.0.1", 0), _ConnectorHandler)
     server.state = state
-    thread = threading.Thread(target=server.serve_forever)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
     thread.start()
     try:
         yield _ConnectorServer(server, state)

@@ -161,7 +161,7 @@ def slack_api_server() -> Generator[_SlackApiServer]:
     state = _SlackApiState()
     server = _SlackApiHTTPServer(("127.0.0.1", 0), _SlackApiHandler)
     server.state = state
-    thread = threading.Thread(target=server.serve_forever)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
     thread.start()
     try:
         yield _SlackApiServer(server, state)