From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Tue, 23 Jun 2026 17:53:21 +0000 (-0400)
Subject: fixes for logging tests
X-Git-Tag: rel_2_1_0b3~8^2
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=695cbfbff2688618add13de21e850d21392592c6;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

fixes for logging tests

Repair some things that went wrong in 0dcfa5e9ea49e481a7a35130f654cbdd3a8c1405
re #13363

* add "logging.handlers" import
* ensure DB connects before capturing log messages so we skip
  initial connect overhead

Change-Id: I10f3ac06b982633d617ec2d9c2a57371038b5a54
---

diff --git a/lib/sqlalchemy/testing/fixtures/base.py b/lib/sqlalchemy/testing/fixtures/base.py
index cbfc77833c..254ad9abf8 100644
--- a/lib/sqlalchemy/testing/fixtures/base.py
+++ b/lib/sqlalchemy/testing/fixtures/base.py
@@ -12,6 +12,7 @@ from __future__ import annotations
 from collections.abc import Callable
 from collections.abc import Sequence
 import logging
+import logging.handlers
 
 import sqlalchemy as sa
 from .. import assertions
@@ -384,19 +385,32 @@ class TestBase:
 
     @config.fixture()
     def debug_logging_engine(self, testing_engine):
+        """a fixture that provides an engine which will capture log messages
+        in a buffer.
+
+        """
         log = logging.getLogger("sqlalchemy.engine")
         existing_level = log.level
 
         buf = logging.handlers.BufferingHandler(100)
         log.addHandler(buf)
 
-        def get_testing_engine(echo=None, log_level=None):
-            options = {"sqlite_share_pool": True}
+        def get_testing_engine(
+            echo=None, log_level=None, skip_initial_connect=True
+        ):
+            te = testing_engine(options={"sqlite_share_pool": True})
+
+            # perform initial connection first so that logging occurs
+            # before we begin capturing
+            if skip_initial_connect:
+                te.connect().close()
+
+            # then set up echo or logging
             if echo is not None:
-                options["echo"] = echo
+                te.echo = echo
             if log_level:
                 log.setLevel(logging.DEBUG)
-            return testing_engine(options=options)
+            return te
 
         try:
             yield get_testing_engine, buf