From a9c06f72ea80347fdc215b63b80e2adcc7ad5901 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 9 Jul 2026 11:35:07 -0700 Subject: [PATCH] test(logging): port logging suite to pytest style Rewrite the logging test module in idiomatic pytest while preserving all 52 tests, their assertions, and subsystem breadth: - unittest.TestCase classes -> plain classes with bare assert - setUpModule/tearDownModule -> module-scoped autouse booted_app fixture - per-class setUp/tearDown -> fixtures (isolated_named_root, snapshot_root) - tempfile.mkstemp/mkdtemp -> tmp_path - redirect_stdout(io.StringIO()) -> capsys - assertRaises -> pytest.raises; assertRegex -> re.search Coverage unchanged at 71.18% (>= 68% gate); full suite green. --- .../tests/logging/test_logging.py | 546 ++++++++---------- 1 file changed, 243 insertions(+), 303 deletions(-) diff --git a/fastapi_startkit/tests/logging/test_logging.py b/fastapi_startkit/tests/logging/test_logging.py index 3fa89272..3f22b07e 100644 --- a/fastapi_startkit/tests/logging/test_logging.py +++ b/fastapi_startkit/tests/logging/test_logging.py @@ -1,10 +1,11 @@ -import io import logging -import unittest -from contextlib import redirect_stdout +import re from unittest.mock import MagicMock, patch +import pytest + from fastapi_startkit.application import Application, app as get_app +from fastapi_startkit.container.container import Container from fastapi_startkit.logging.ChannelFactory import ChannelFactory from fastapi_startkit.logging.channels import ( DailyChannel, @@ -29,45 +30,54 @@ from fastapi_startkit.logging.managers import LoggingManager from fastapi_startkit.logging.providers.log_provider import LogProvider - -_saved_state = {} +ALL_LEVELS = ("emergency", "alert", "critical", "error", "warning", "notice", "info", "debug") -def setUpModule(): +@pytest.fixture(scope="module", autouse=True) +def booted_app(): """Boot a testing Application with the LogProvider so the logging config, the channel/driver factories, and the `logger` channel are wired up exactly as they are at runtime — the framework does the registration, not the test. Booting installs a global LoggingHandler on the real root logger and swaps - the global app singleton, so both are captured here and restored in - tearDownModule to stop this module's setup from leaking into other tests. + the global app singleton, so both are captured here and restored on teardown + to stop this module's setup from leaking into other tests. """ - from fastapi_startkit.container.container import Container - root = logging.getLogger() - _saved_state["handlers"] = list(root.handlers) - _saved_state["level"] = root.level + saved_handlers = list(root.handlers) + saved_level = root.level try: - _saved_state["app"] = Container.instance() + saved_app = Container.instance() except Exception: - _saved_state["app"] = None + saved_app = None Application(env="testing", providers=[LogProvider]) + yield - -def tearDownModule(): - """Restore the real root logger and the global app singleton.""" - from fastapi_startkit.container.container import Container - - root = logging.getLogger() for handler in list(root.handlers): - if handler not in _saved_state.get("handlers", []): + if handler not in saved_handlers: root.removeHandler(handler) - root.setLevel(_saved_state.get("level", logging.WARNING)) - + root.setLevel(saved_level) Logger.instance = None - if _saved_state.get("app") is not None: - Container.set_instance(_saved_state["app"]) + if saved_app is not None: + Container.set_instance(saved_app) + + +@pytest.fixture +def isolated_named_root(): + """Isolate the "root"-named logger the single/syslog drivers write through: + disable propagation so records do not reach the installed LoggingHandler, and + remove any handlers the test attaches so they do not leak into other tests.""" + root_named = logging.getLogger("root") + propagate = root_named.propagate + root_named.propagate = False + existing = list(root_named.handlers) + yield root_named + for handler in list(root_named.handlers): + if handler not in existing: + root_named.removeHandler(handler) + handler.close() + root_named.propagate = propagate class RecordingDriver(BaseDriver): @@ -93,60 +103,61 @@ def __getattr__(self, name): raise AttributeError(name) -class ChannelFactoryTest(unittest.TestCase): +class TestChannelFactory: def test_make_returns_registered_channel_classes(self): - self.assertIs(ChannelFactory.make("single"), SingleChannel) - self.assertIs(ChannelFactory.make("daily"), DailyChannel) - self.assertIs(ChannelFactory.make("terminal"), TerminalChannel) - self.assertIs(ChannelFactory.make("stack"), StackChannel) + assert ChannelFactory.make("single") is SingleChannel + assert ChannelFactory.make("daily") is DailyChannel + assert ChannelFactory.make("terminal") is TerminalChannel + assert ChannelFactory.make("stack") is StackChannel def test_make_returns_none_for_unknown_channel(self): - self.assertIsNone(ChannelFactory.make("does-not-exist")) + assert ChannelFactory.make("does-not-exist") is None def test_register_adds_new_channels(self): marker = object() try: ChannelFactory.register({"custom_test_channel": marker}) - self.assertIs(ChannelFactory.make("custom_test_channel"), marker) + assert ChannelFactory.make("custom_test_channel") is marker finally: ChannelFactory.channels.pop("custom_test_channel", None) -class DriverFactoryTest(unittest.TestCase): +class TestDriverFactory: def test_make_maps_names_to_driver_classes(self): - self.assertIs(DriverFactory.make("single"), LogSingleDriver) - self.assertIs(DriverFactory.make("daily"), LogSingleDriver) - self.assertIs(DriverFactory.make("slack"), LogSlackDriver) - self.assertIs(DriverFactory.make("syslog"), LogSyslogDriver) - self.assertIs(DriverFactory.make("terminal"), LogTerminalDriver) + assert DriverFactory.make("single") is LogSingleDriver + assert DriverFactory.make("daily") is LogSingleDriver + assert DriverFactory.make("slack") is LogSlackDriver + assert DriverFactory.make("syslog") is LogSyslogDriver + assert DriverFactory.make("terminal") is LogTerminalDriver def test_make_returns_none_for_unknown_driver(self): - self.assertIsNone(DriverFactory.make("unknown")) + assert DriverFactory.make("unknown") is None -class BaseDriverShouldRunTest(unittest.TestCase): - def setUp(self): - self.driver = BaseDriver() +class TestBaseDriverShouldRun: + @pytest.fixture + def driver(self): + return BaseDriver() - def test_returns_true_when_no_max_level(self): - self.assertTrue(self.driver.should_run("debug", None)) + def test_returns_true_when_no_max_level(self, driver): + assert driver.should_run("debug", None) - def test_more_severe_level_runs_under_a_less_severe_threshold(self): + def test_more_severe_level_runs_under_a_less_severe_threshold(self, driver): # error (index 3) is more severe than info (index 6) - self.assertTrue(self.driver.should_run("error", "info")) + assert driver.should_run("error", "info") - def test_less_severe_level_is_filtered_out(self): + def test_less_severe_level_is_filtered_out(self, driver): # debug (index 7) is less severe than info (index 6) - self.assertFalse(self.driver.should_run("debug", "info")) + assert not driver.should_run("debug", "info") - def test_equal_level_runs(self): - self.assertTrue(self.driver.should_run("info", "info")) + def test_equal_level_runs(self, driver): + assert driver.should_run("info", "info") - def test_get_time_returns_formatted_datetime(self): - self.assertIsInstance(self.driver.get_time().to_datetime_string(), str) + def test_get_time_returns_formatted_datetime(self, driver): + assert isinstance(driver.get_time().to_datetime_string(), str) -class BaseChannelTest(unittest.TestCase): +class TestBaseChannel: def _channel(self, should_run=True, max_level="debug"): channel = BaseChannel() channel.driver = RecordingDriver(should_run=should_run) @@ -155,27 +166,27 @@ def _channel(self, should_run=True, max_level="debug"): def test_level_methods_delegate_to_driver_when_allowed(self): channel = self._channel(should_run=True) - for level in ("emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"): + for level in ALL_LEVELS: result = getattr(channel, level)("message") - self.assertEqual(result, f"{level}:message") - self.assertEqual(len(channel.driver.calls), 8) + assert result == f"{level}:message" + assert len(channel.driver.calls) == 8 def test_level_methods_are_suppressed_when_not_allowed(self): channel = self._channel(should_run=False) - self.assertIsNone(channel.info("message")) - self.assertIsNone(channel.error("message")) - self.assertEqual(channel.driver.calls, []) + assert channel.info("message") is None + assert channel.error("message") is None + assert channel.driver.calls == [] def test_log_reports_whether_level_should_run(self): - self.assertTrue(self._channel(should_run=True).log("info", "m")) - self.assertFalse(self._channel(should_run=False).log("info", "m")) + assert self._channel(should_run=True).log("info", "m") + assert not self._channel(should_run=False).log("info", "m") def test_channel_builds_a_new_channel_instance(self): channel = self._channel() - self.assertIsInstance(channel.channel("terminal"), TerminalChannel) + assert isinstance(channel.channel("terminal"), TerminalChannel) -class MultiBaseChannelTest(unittest.TestCase): +class TestMultiBaseChannel: def _multi(self, should_run=True): inner = BaseChannel() inner.driver = RecordingDriver(should_run=should_run) @@ -186,109 +197,75 @@ def _multi(self, should_run=True): def test_broadcasts_each_level_to_all_channels(self): multi, inner = self._multi(should_run=True) - for level in ("emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"): + for level in ALL_LEVELS: getattr(multi, level)("message") recorded = [level for level, _ in inner.driver.calls] - self.assertEqual( - recorded, - ["emergency", "alert", "critical", "error", "warning", "notice", "info", "debug"], - ) + assert recorded == list(ALL_LEVELS) def test_skips_channels_that_should_not_run(self): multi, inner = self._multi(should_run=False) multi.info("message") multi.error("message") - self.assertEqual(inner.driver.calls, []) + assert inner.driver.calls == [] -class TerminalDriverTest(unittest.TestCase): - def setUp(self): - self.driver = LogTerminalDriver() +class TestTerminalDriver: + @pytest.fixture + def driver(self): + return LogTerminalDriver() - def test_get_format_includes_time_level_and_message(self): - formatted = self.driver.get_format("hello", "INFO") - self.assertIn("INFO", formatted) - self.assertIn("hello", formatted) - self.assertRegex(formatted, r"\d{4}-\d{2}-\d{2}") + def test_get_format_includes_time_level_and_message(self, driver): + formatted = driver.get_format("hello", "INFO") + assert "INFO" in formatted + assert "hello" in formatted + assert re.search(r"\d{4}-\d{2}-\d{2}", formatted) - def test_each_level_writes_to_stdout(self): + def test_each_level_writes_to_stdout(self, driver, capsys): for level, method in ( - ("EMERGENCY", self.driver.emergency), - ("ALERT", self.driver.alert), - ("CRITICAL", self.driver.critical), - ("ERROR", self.driver.error), - ("WARNING", self.driver.warning), - ("NOTICE", self.driver.notice), - ("INFO", self.driver.info), - ("DEBUG", self.driver.debug), + ("EMERGENCY", driver.emergency), + ("ALERT", driver.alert), + ("CRITICAL", driver.critical), + ("ERROR", driver.error), + ("WARNING", driver.warning), + ("NOTICE", driver.notice), + ("INFO", driver.info), + ("DEBUG", driver.debug), ): - buffer = io.StringIO() - with redirect_stdout(buffer): - method("payload") - output = buffer.getvalue() - self.assertIn(level, output) - self.assertIn("payload", output) - - -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) - - 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 + method("payload") + output = capsys.readouterr().out + assert level in output + assert "payload" in output - def test_writes_formatted_levels_to_the_file(self): - import tempfile - import os - fd, path = tempfile.mkstemp(suffix=".log") - os.close(fd) - try: - driver = LogSingleDriver(path=path, max_level="debug") - 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) - finally: - os.remove(path) - - def test_change_format_replaces_handlers(self): - import tempfile - import os - - 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) - - -class SlackDriverTest(unittest.TestCase): +class TestSingleDriver: + def test_writes_formatted_levels_to_the_file(self, tmp_path, isolated_named_root): + path = tmp_path / "single.log" + driver = LogSingleDriver(path=str(path), max_level="debug") + driver.error("boom") + driver.info("ping") + contents = path.read_text() + assert "ERROR" in contents + assert "boom" in contents + assert "INFO" in contents + assert "ping" in contents + + def test_change_format_replaces_handlers(self, tmp_path, isolated_named_root): + path = tmp_path / "single.log" + driver = LogSingleDriver(path=str(path), max_level="debug") + before = len(driver.log.handlers) + driver.change_format("%(message)s") + assert len(driver.log.handlers) <= before + assert len(driver.log.handlers) >= 1 + + +class TestSlackDriver: def _driver(self): return LogSlackDriver(token="tok", channel="#bot", username="bot", emoji=":x:") def test_get_format_includes_level_and_message(self): formatted = self._driver().get_format("hi", "ERROR") - self.assertIn("ERROR", formatted) - self.assertIn("hi", formatted) + assert "ERROR" in formatted + assert "hi" in formatted def test_level_methods_post_to_slack(self): driver = self._driver() @@ -301,12 +278,12 @@ def test_level_methods_post_to_slack(self): url, payload = requests_mock.post.call_args.args # Assert the concrete Slack endpoint, not driver.slack_url (that would be # circular — it is the value that produced the call). - self.assertEqual(url, "https://slack.com/api/chat.postMessage") - self.assertIn("boom", payload["text"]) - self.assertIn("ERROR", payload["text"]) - self.assertEqual(payload["token"], "tok") - self.assertEqual(payload["channel"], "C123") - self.assertEqual(payload["username"], "bot") + assert url == "https://slack.com/api/chat.postMessage" + assert "boom" in payload["text"] + assert "ERROR" in payload["text"] + assert payload["token"] == "tok" + assert payload["channel"] == "C123" + assert payload["username"] == "bot" def test_find_channel_resolves_channel_id(self): driver = self._driver() @@ -314,7 +291,7 @@ def test_find_channel_resolves_channel_id(self): response.json.return_value = {"channels": [{"name": "bot", "id": "C123"}]} with patch("fastapi_startkit.logging.drivers.LogSlackDriver.requests") as requests_mock: requests_mock.post.return_value = response - self.assertEqual(driver.find_channel("#bot"), "C123") + assert driver.find_channel("#bot") == "C123" def test_find_channel_raises_when_missing(self): driver = self._driver() @@ -322,11 +299,11 @@ def test_find_channel_raises_when_missing(self): response.json.return_value = {"channels": [{"name": "other", "id": "C999"}]} with patch("fastapi_startkit.logging.drivers.LogSlackDriver.requests") as requests_mock: requests_mock.post.return_value = response - with self.assertRaises(Exception): + with pytest.raises(Exception): driver.find_channel("#bot") -class SyslogDriverTest(unittest.TestCase): +class TestSyslogDriver: def test_levels_route_to_the_underlying_logger(self): with patch("fastapi_startkit.logging.drivers.LogSyslogDriver.logging.handlers.SysLogHandler"): driver = LogSyslogDriver(path="/dev/null") @@ -339,24 +316,15 @@ def test_levels_route_to_the_underlying_logger(self): driver.log.critical.assert_called_once_with("halt") -class TimezoneAwareLogFileTest(unittest.TestCase): +class TestTimezoneAwareLogFile: """The daily log file name comes from get_time(), which resolves `logging.channels.timezone` — so the file that gets generated on disk must follow the application's configured timezone, not the machine/UTC clock.""" - def setUp(self): + @pytest.fixture(autouse=True) + def _config(self, isolated_named_root): 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 + yield # Restore to the code's default so other tests' get_time() keeps working. self.config.set("logging.channels.timezone", "UTC") @@ -365,28 +333,25 @@ def _daily_path(self, directory, tz, fixed_instant): with patch("pendulum.now", return_value=fixed_instant): return DailyChannel(driver="daily", path=directory).driver.path - def test_daily_file_date_follows_configured_timezone(self): - import os - import tempfile - + def test_daily_file_date_follows_configured_timezone(self, tmp_path): import pendulum # 23:30 UTC sits on a day boundary: timezones ahead of UTC are already on # the next calendar day, timezones behind are still on the previous one. fixed = pendulum.datetime(2026, 7, 8, 23, 30, 0, tz="UTC") - directory = tempfile.mkdtemp() + directory = str(tmp_path) ahead = self._daily_path(directory, "Pacific/Kiritimati", fixed) # UTC+14 behind = self._daily_path(directory, "Pacific/Honolulu", fixed) # UTC-10 # File names differ purely because of the configured timezone. - self.assertTrue(ahead.endswith("2026-07-09.log"), ahead) - self.assertTrue(behind.endswith("2026-07-08.log"), behind) - self.assertNotEqual(ahead, behind) + assert ahead.endswith("2026-07-09.log"), ahead + assert behind.endswith("2026-07-08.log"), behind + assert ahead != behind # And both files are actually created on disk with the tz-derived names. - self.assertTrue(os.path.isfile(os.path.join(directory, "2026-07-09.log"))) - self.assertTrue(os.path.isfile(os.path.join(directory, "2026-07-08.log"))) + assert (tmp_path / "2026-07-09.log").is_file() + assert (tmp_path / "2026-07-08.log").is_file() def test_get_time_reflects_configured_timezone(self): import pendulum @@ -395,70 +360,60 @@ def test_get_time_reflects_configured_timezone(self): self.config.set("logging.channels.timezone", "Pacific/Kiritimati") with patch("pendulum.now", return_value=fixed): now = BaseDriver().get_time() - self.assertEqual(now.to_date_string(), "2026-07-09") - self.assertEqual(now.timezone_name, "Pacific/Kiritimati") + assert now.to_date_string() == "2026-07-09" + assert now.timezone_name == "Pacific/Kiritimati" -class ChannelConstructionTest(unittest.TestCase): +class TestChannelConstruction: + @staticmethod + def _close_file_handlers(driver): + for handler in list(driver.log.handlers): + if isinstance(handler, logging.FileHandler): + driver.log.removeHandler(handler) + handler.close() + def test_terminal_channel_uses_terminal_driver(self): channel = TerminalChannel() - self.assertEqual(channel.max_level, "info") - self.assertIsInstance(channel.driver, LogTerminalDriver) + assert channel.max_level == "info" + assert isinstance(channel.driver, LogTerminalDriver) - def test_single_channel_builds_single_driver(self): - import tempfile - import os - - directory = tempfile.mkdtemp() - path = os.path.join(directory, "single.log") + def test_single_channel_builds_single_driver(self, tmp_path): + channel = SingleChannel(driver="single", path=str(tmp_path / "single.log")) try: - channel = SingleChannel(driver="single", path=path) - self.assertIsInstance(channel.driver, LogSingleDriver) + assert isinstance(channel.driver, LogSingleDriver) finally: - for handler in list(channel.driver.log.handlers): - if isinstance(handler, logging.FileHandler): - channel.driver.log.removeHandler(handler) - handler.close() - - def test_daily_channel_writes_dated_file(self): - import tempfile + self._close_file_handlers(channel.driver) - directory = tempfile.mkdtemp() - channel = DailyChannel(driver="daily", path=directory) + def test_daily_channel_writes_dated_file(self, tmp_path): + channel = DailyChannel(driver="daily", path=str(tmp_path)) try: - self.assertIsInstance(channel.driver, LogSingleDriver) - self.assertTrue(channel.driver.path.endswith(".log")) + assert isinstance(channel.driver, LogSingleDriver) + assert channel.driver.path.endswith(".log") finally: - for handler in list(channel.driver.log.handlers): - if isinstance(handler, logging.FileHandler): - channel.driver.log.removeHandler(handler) - handler.close() + self._close_file_handlers(channel.driver) def test_stack_channel_collects_known_channels(self): channel = StackChannel(channels=["terminal"]) - self.assertEqual(len(channel.channels), 1) - self.assertIsInstance(channel.channels[0], TerminalChannel) + assert len(channel.channels) == 1 + assert isinstance(channel.channels[0], TerminalChannel) def test_stack_channel_ignores_unknown_channels(self): channel = StackChannel(channels=["nope-not-real"]) - self.assertEqual(channel.channels, []) + assert channel.channels == [] - def test_stack_channel_respects_level_thresholds(self): + def test_stack_channel_respects_level_thresholds(self, capsys): channel = StackChannel(channels=["terminal"]) - buffer = io.StringIO() - with redirect_stdout(buffer): - channel.info("shown") - self.assertIn("shown", buffer.getvalue()) - - buffer = io.StringIO() - with redirect_stdout(buffer): - channel.debug("hidden") + channel.info("shown") + assert "shown" in capsys.readouterr().out + + channel.debug("hidden") # terminal level is "info", so debug is filtered out - self.assertEqual(buffer.getvalue(), "") + assert capsys.readouterr().out == "" -class LoggerFacadeTest(unittest.TestCase): - def setUp(self): +class TestLoggerFacade: + @pytest.fixture(autouse=True) + def _bind_logger(self): self.app = get_app() self._had_logger = False try: @@ -469,8 +424,7 @@ def setUp(self): self.channel = MagicMock() self.app.bind("logger", self.channel) Logger.instance = None - - def tearDown(self): + yield Logger.instance = None if self._had_logger: self.app.bind("logger", self._previous) @@ -479,8 +433,8 @@ def test_level_methods_delegate_to_resolved_channel(self): Logger.info("hello") self.channel.info.assert_called_once() (message,), _ = self.channel.info.call_args - self.assertTrue(message.endswith("hello")) - self.assertIn(" - ", message) + assert message.endswith("hello") + assert " - " in message def test_error_and_debug_delegate(self): Logger.error("bad") @@ -500,27 +454,27 @@ def test_log_falls_back_to_error_for_unknown_level(self): def test_logger_info_returns_caller_location(self): info = Logger.logger_info() - self.assertIsInstance(info, str) - self.assertIn(" - ", info) - self.assertNotIn("logging/logger.py", info) + assert isinstance(info, str) + assert " - " in info + assert "logging/logger.py" not in info -class LoggingManagerTest(unittest.TestCase): - def setUp(self): - self.root = logging.getLogger() - self._handlers = list(self.root.handlers) - - def tearDown(self): - for handler in list(self.root.handlers): - if handler not in self._handlers: - self.root.removeHandler(handler) +class TestLoggingManager: + @pytest.fixture(autouse=True) + def _snapshot_root(self): + root = logging.getLogger() + handlers = list(root.handlers) + yield + for handler in list(root.handlers): + if handler not in handlers: + root.removeHandler(handler) def test_channel_delegates_to_channel_factory(self): sentinel = object() factory = MagicMock() factory.make.return_value = MagicMock(return_value=sentinel) manager = LoggingManager(channel_factory=factory, driver_factory=None) - self.assertIs(manager.channel("single"), sentinel) + assert manager.channel("single") is sentinel factory.make.assert_called_once_with("single") def test_configure_python_logging_installs_handler_once(self): @@ -528,18 +482,18 @@ def test_configure_python_logging_installs_handler_once(self): LoggingManager.configure_python_logging() LoggingManager.configure_python_logging() installed = [h for h in root.handlers if isinstance(h, LoggingHandler)] - self.assertEqual(len(installed), 1) - + assert len(installed) == 1 -class LoggingHandlerTest(unittest.TestCase): - def setUp(self): - self.root = logging.getLogger() - self._handlers = list(self.root.handlers) - def tearDown(self): - for handler in list(self.root.handlers): - if handler not in self._handlers: - self.root.removeHandler(handler) +class TestLoggingHandler: + @pytest.fixture(autouse=True) + def _snapshot_root(self): + root = logging.getLogger() + handlers = list(root.handlers) + yield + for handler in list(root.handlers): + if handler not in handlers: + root.removeHandler(handler) def test_emit_forwards_record_to_logger(self): handler = LoggingHandler() @@ -556,93 +510,79 @@ def test_emit_forwards_record_to_logger(self): handler.emit(record) log_mock.assert_called_once() args, _ = log_mock.call_args - self.assertEqual(args[0], "ERROR") - self.assertIn("kaboom", args[1]) + assert args[0] == "ERROR" + assert "kaboom" in args[1] def test_install_is_idempotent(self): root = logging.getLogger() LoggingHandler.install() LoggingHandler.install() - self.assertEqual(len([h for h in root.handlers if isinstance(h, LoggingHandler)]), 1) - - -class MakeDirectoryTest(unittest.TestCase): - def test_returns_false_for_existing_file(self): - import tempfile - import os - - fd, path = tempfile.mkstemp() - os.close(fd) - try: - self.assertFalse(make_directory(path)) - finally: - os.remove(path) + assert len([h for h in root.handlers if isinstance(h, LoggingHandler)]) == 1 - def test_creates_missing_parent_directories(self): - import tempfile - import os - base = tempfile.mkdtemp() - target = os.path.join(base, "nested", "deep", "file.log") - self.assertTrue(make_directory(target)) - self.assertTrue(os.path.isdir(os.path.join(base, "nested", "deep"))) +class TestMakeDirectory: + def test_returns_false_for_existing_file(self, tmp_path): + path = tmp_path / "file" + path.write_text("") + assert not make_directory(str(path)) - def test_returns_true_without_recreating_existing_parent(self): - import tempfile - import os + def test_creates_missing_parent_directories(self, tmp_path): + target = tmp_path / "nested" / "deep" / "file.log" + assert make_directory(str(target)) + assert (tmp_path / "nested" / "deep").is_dir() - base = tempfile.mkdtemp() - target = os.path.join(base, "file.log") - self.assertTrue(make_directory(target)) - self.assertFalse(os.path.isfile(target)) + def test_returns_true_without_recreating_existing_parent(self, tmp_path): + target = tmp_path / "file.log" + assert make_directory(str(target)) + assert not target.is_file() -class LoggingConfigTest(unittest.TestCase): +class TestLoggingConfig: def test_default_channels_are_present(self): config = LoggingConfig() - self.assertIn("stack", config.channels) - self.assertIn("daily", config.channels) - self.assertIn("terminal", config.channels) - self.assertIsInstance(config.default, str) + assert "stack" in config.channels + assert "daily" in config.channels + assert "terminal" in config.channels + assert isinstance(config.default, str) def test_channel_dataclass_defaults(self): - self.assertEqual(config_channels.SingleChannel().driver, "single") - self.assertEqual(config_channels.SingleChannel().level, "debug") - self.assertEqual(config_channels.StackChannel().channels, ["daily", "terminal"]) - self.assertEqual(config_channels.DailyChannel().driver, "daily") - self.assertEqual(config_channels.TerminalChannel().level, "info") - self.assertEqual(config_channels.SlackChannel().driver, "slack") - self.assertEqual(config_channels.SyslogChannel().path, "/var/run/syslog") + assert config_channels.SingleChannel().driver == "single" + assert config_channels.SingleChannel().level == "debug" + assert config_channels.StackChannel().channels == ["daily", "terminal"] + assert config_channels.DailyChannel().driver == "daily" + assert config_channels.TerminalChannel().level == "info" + assert config_channels.SlackChannel().driver == "slack" + assert config_channels.SyslogChannel().path == "/var/run/syslog" -class ListenerTest(unittest.TestCase): +class TestListener: def test_handle_logs_exception_details(self): logger = MagicMock() listener = LoggerExceptionListener(logger) - self.assertEqual(listener.listens, ["*"]) + assert listener.listens == ["*"] listener.handle(ValueError("nope"), "app.py", 42) logger.error.assert_called_once() (message,), _ = logger.error.call_args - self.assertIn("ValueError", message) - self.assertIn("app.py", message) - self.assertIn("42", message) + assert "ValueError" in message + assert "app.py" in message + assert "42" in message -class LogProviderTest(unittest.TestCase): +class TestLogProvider: def test_register_binds_factories_and_manager(self): application = get_app() LogProvider(application).register() # The container auto-resolves bound classes into instances on make(). - self.assertIsInstance(application.make("LogChannelFactory"), ChannelFactory) - self.assertIsInstance(application.make("LogDriverFactory"), DriverFactory) - self.assertIsInstance(application.make("LoggingManager"), LoggingManager) + assert isinstance(application.make("LogChannelFactory"), ChannelFactory) + assert isinstance(application.make("LogDriverFactory"), DriverFactory) + assert isinstance(application.make("LoggingManager"), LoggingManager) def test_register_merges_default_config(self): from fastapi_startkit.facades import Config application = get_app() LogProvider(application).register() - self.assertEqual(Config.get("logging.default"), "stack") + assert Config.get("logging.default") == "stack" def test_boot_binds_and_swaps_logger_channel(self): application = get_app() @@ -653,8 +593,8 @@ def test_boot_binds_and_swaps_logger_channel(self): config.set("logging.default", "terminal") try: provider.boot() - self.assertIsInstance(application.make("logger"), TerminalChannel) - self.assertIn("logging", application.published_resources) + assert isinstance(application.make("logger"), TerminalChannel) + assert "logging" in application.published_resources finally: config.set("logging.default", original_default) @@ -664,4 +604,4 @@ def test_boot_returns_early_when_no_default_channel(self): provider = LogProvider(mock_app) provider.boot() called_keys = [call.args[0] for call in mock_app.make.call_args_list if call.args] - self.assertNotIn("LoggingManager", called_keys) + assert "LoggingManager" not in called_keys