]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
fixes for logging tests
authorMike Bayer <mike_mp@zzzcomputing.com>
Tue, 23 Jun 2026 17:53:21 +0000 (13:53 -0400)
committerMike Bayer <mike_mp@zzzcomputing.com>
Tue, 23 Jun 2026 17:54:49 +0000 (13:54 -0400)
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

lib/sqlalchemy/testing/fixtures/base.py

index cbfc77833cd504844296c09d48cc2c8b58d384ec..254ad9abf8601fdda362e77b97a498b9b3c682ec 100644 (file)
@@ -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