Fix FD leak in LogSingleDriver.change_format

fastapi_startkit/src/fastapi_startkit/logging/drivers/LogSingleDriver.py

@@ -1,66 +1,64 @@
 import logging
+import os
+
 from .BaseDriver import BaseDriver
 
 
 class LogSingleDriver(BaseDriver):
     def __init__(self, *args, path: str, max_level: str | None = None, **kwargs):
         self.max_level = max_level
         self.path = path
-        self.log = logging.getLogger("root")
+        # A dedicated, non-propagating logger per file keeps records away from the root
+        # logger (and its LoggingHandler bridge, which would feed them back into Logger).
+        self.log = logging.getLogger(f"fastapi_startkit.logging.single.{os.path.abspath(path)}")
+        self.log.setLevel(logging.DEBUG)
+        self.log.propagate = False
+        self.handler = self._file_handler()
+        self.handler.setFormatter(
+            logging.Formatter("{} - %(levelname)s - %(message)s".format(self.get_time().to_datetime_string()))
+        )
+
+    def _file_handler(self) -> logging.FileHandler:
+        # Drivers are rebuilt per channel instance; reuse the handler so each file holds one descriptor.
+        for handler in self.log.handlers:
+            if isinstance(handler, logging.FileHandler):
+                return handler
 
         handler = logging.FileHandler(self.path, "a")
-        formatter = logging.Formatter("{} - %(levelname)s - %(message)s".format(self.get_time().to_datetime_string()))
-
-        handler.setFormatter(formatter)
         self.log.addHandler(handler)
+        return handler
 
     def change_format(self, changed_format):
-        for hdlr in self.log.handlers[:]:  # remove all old handlers
-            self.log.removeHandler(hdlr)
+        self.handler.setFormatter(logging.Formatter(changed_format))
 
-        handler = logging.FileHandler(self.path, "a")
-        formatter = logging.Formatter(changed_format)
+    def close(self):
+        self.log.removeHandler(self.handler)
+        self.handler.close()
 
-        handler.setFormatter(formatter)
-
-        self.log.addHandler(handler)
+    def _write(self, level: int, label: str, message):
+        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), label))
+        return self.log.log(level, message)
 
     def emergency(self, message, *args, **kwargs):
-        self.log.setLevel(logging.CRITICAL)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "EMERGENCY"))
-        return self.log.critical(message)
+        return self._write(logging.CRITICAL, "EMERGENCY", message)
 
     def alert(self, message, *args, **kwargs):
-        self.log.setLevel(logging.CRITICAL)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "ALERT"))
-        return self.log.critical(message)
+        return self._write(logging.CRITICAL, "ALERT", message)
 
     def critical(self, message, *args, **kwargs):
-        self.log.setLevel(logging.CRITICAL)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "CRITICAL"))
-        return self.log.critical(message)
+        return self._write(logging.CRITICAL, "CRITICAL", message)
 
     def error(self, message, *args, **kwargs):
-        self.log.setLevel(logging.ERROR)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "ERROR"))
-        return self.log.error(message)
+        return self._write(logging.ERROR, "ERROR", message)
 
     def warning(self, message, *args, **kwargs):
-        self.log.setLevel(logging.WARNING)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "WARNING"))
-        return self.log.warning(message)
+        return self._write(logging.WARNING, "WARNING", message)
 
     def notice(self, message, *args, **kwargs):
-        self.log.setLevel(logging.INFO)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "NOTICE"))
-        return self.log.info(message)
+        return self._write(logging.INFO, "NOTICE", message)
 
     def info(self, message, *args, **kwargs):
-        self.log.setLevel(logging.INFO)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "INFO"))
-        return self.log.info(message)
+        return self._write(logging.INFO, "INFO", message)
 
     def debug(self, message, *args, **kwargs):
-        self.log.setLevel(logging.DEBUG)
-        self.change_format("{} - {} - %(message)s".format(self.get_time().to_datetime_string(), "DEBUG"))
-        return self.log.debug(message)
+        return self._write(logging.DEBUG, "DEBUG", message)

fastapi_startkit/tests/logging/test_logging.py

@@ -1,5 +1,6 @@
 import io
 import logging
+import os
 import unittest
 from contextlib import redirect_stdout
 from unittest.mock import MagicMock, patch
@@ -242,53 +243,101 @@ def test_each_level_writes_to_stdout(self):
 
 class SingleDriverTest(unittest.TestCase):
     def setUp(self):
-        # The single/syslog drivers log through the "root"-named logger; disable
-        # propagation so records do not reach the installed LoggingHandler.
-        self.root_named = logging.getLogger("root")
-        self._propagate = self.root_named.propagate
-        self.root_named.propagate = False
-        self._existing_handlers = list(self.root_named.handlers)
+        import tempfile
+
+        self.directory = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self.directory.name, "app.log")
+        self.drivers: list[LogSingleDriver] = []
 
     def tearDown(self):
-        for handler in list(self.root_named.handlers):
-            if handler not in self._existing_handlers:
-                self.root_named.removeHandler(handler)
-                handler.close()
-        self.root_named.propagate = self._propagate
+        for driver in self.drivers:
+            driver.close()
+        self.directory.cleanup()
+
+    def _driver(self, path=None):
+        driver = LogSingleDriver(path=path or self.path, max_level="debug")
+        self.drivers.append(driver)
+        return driver
+
+    def _contents(self):
+        with open(self.path) as handle:
+            return handle.read()
+
+    @staticmethod
+    def _open_fd_count():
+        return len(os.listdir("/dev/fd"))
 
     def test_writes_formatted_levels_to_the_file(self):
-        import tempfile
-        import os
+        driver = self._driver()
+        for level in ("emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"):
+            getattr(driver, level)(f"{level}-message")
 
-        fd, path = tempfile.mkstemp(suffix=".log")
-        os.close(fd)
+        lines = self._contents().splitlines()
+        self.assertEqual(len(lines), 8)
+        for line, label in zip(
+            lines, ("EMERGENCY", "ALERT", "CRITICAL", "ERROR", "WARNING", "NOTICE", "INFO", "DEBUG")
+        ):
+            self.assertIn(f" - {label} - {label.lower()}-message", line)
+
+    def test_repeated_logging_keeps_handler_and_fd_counts_stable(self):
+        driver = self._driver()
+        driver.info("warm up")
+        handlers_before = len(driver.log.handlers)
+        fds_before = self._open_fd_count()
+
+        for i in range(200):
+            driver.error(f"boom {i}")
+            driver.debug(f"trace {i}")
+
+        self.assertEqual(handlers_before, 1)
+        self.assertEqual(len(driver.log.handlers), handlers_before)
+        self.assertEqual(self._open_fd_count(), fds_before)
+        self.assertEqual(len(self._contents().splitlines()), 401)
+
+    def test_drivers_for_the_same_file_share_one_handler(self):
+        first = self._driver()
+        second = self._driver()
+
+        self.assertIs(first.handler, second.handler)
+        self.assertEqual(len(second.log.handlers), 1)
+
+    def test_change_format_swaps_formatter_on_the_existing_handler(self):
+        driver = self._driver()
+        handler = driver.handler
+
+        driver.change_format("custom: %(message)s")
+        driver.log.info("hello")
+
+        self.assertEqual(driver.log.handlers, [handler])
+        self.assertEqual(self._contents(), "custom: hello\n")
+
+    def test_logging_leaves_root_logger_handlers_and_level_untouched(self):
+        root = logging.getLogger()
+        sentinel = logging.NullHandler()
+        root.addHandler(sentinel)
+        handlers_before = list(root.handlers)
+        level_before = root.level
         try:
-            driver = LogSingleDriver(path=path, max_level="debug")
+            driver = self._driver()
             driver.error("boom")
-            driver.info("ping")
-            with open(path) as handle:
-                contents = handle.read()
-            self.assertIn("ERROR", contents)
-            self.assertIn("boom", contents)
-            self.assertIn("INFO", contents)
-            self.assertIn("ping", contents)
+            driver.debug("trace")
+
+            self.assertEqual(root.handlers, handlers_before)
+            self.assertTrue(any(isinstance(h, LoggingHandler) for h in root.handlers))
+            self.assertEqual(root.level, level_before)
+            self.assertFalse(driver.log.propagate)
         finally:
-            os.remove(path)
+            root.removeHandler(sentinel)
 
-    def test_change_format_replaces_handlers(self):
-        import tempfile
-        import os
+    def test_close_releases_the_file_handler(self):
+        driver = self._driver()
+        driver.info("hello")
+        fds_open = self._open_fd_count()
 
-        fd, path = tempfile.mkstemp(suffix=".log")
-        os.close(fd)
-        try:
-            driver = LogSingleDriver(path=path, max_level="debug")
-            before = len(driver.log.handlers)
-            driver.change_format("%(message)s")
-            self.assertLessEqual(len(driver.log.handlers), before)
-            self.assertTrue(len(driver.log.handlers) >= 1)
-        finally:
-            os.remove(path)
+        driver.close()
+
+        self.assertEqual(driver.log.handlers, [])
+        self.assertEqual(self._open_fd_count(), fds_open - 1)
 
 
 class SlackDriverTest(unittest.TestCase):
@@ -364,24 +413,17 @@ class TimezoneAwareLogFileTest(unittest.TestCase):
 
     def setUp(self):
         self.config = get_app().make("config")
-        self.root_named = logging.getLogger("root")
-        self._propagate = self.root_named.propagate
-        self.root_named.propagate = False
-        self._existing_handlers = list(self.root_named.handlers)
 
     def tearDown(self):
-        for handler in list(self.root_named.handlers):
-            if handler not in self._existing_handlers:
-                self.root_named.removeHandler(handler)
-                handler.close()
-        self.root_named.propagate = self._propagate
         # Restore to the code's default so other tests' get_time() keeps working.
         self.config.set("logging.channels.timezone", "UTC")
 
     def _daily_path(self, directory, tz, fixed_instant):
         self.config.set("logging.channels.timezone", tz)
         with patch("pendulum.now", return_value=fixed_instant):
-            return DailyChannel(driver="daily", path=directory).driver.path
+            driver = DailyChannel(driver="daily", path=directory).driver
+        driver.close()
+        return driver.path
 
     def test_daily_file_date_follows_configured_timezone(self):
         import os