ruff

Author: emmuhamm

tests/logging/test_handlers.py

@@ -1,38 +1,3 @@
-
-
-
-
-#! What do we want to test here? 
-
-#! The processing of the environment variables, partiuclarly the protobuf, the random one, and also what happens when you try to initialise it without the right environment
-
-# this should test  _convert_str_to_handlertype with throttle, protobufstream, and random
-# throttle should return handlertype.throttle, none , protobufstream should return handlertype.protobufstream, and a valid protobufconf with the same values as expected
-
-# See if its possible to test _make_ers_handler_conf
-# This should just take a default string comma and then return the correct features
-
-# Definitely test the creation of the full thing, with the four environments
-
-
-#! Also test the from string in HandlerTypes
-# Should get this testesd for the enums but also the error itself
-
-
-#! We need to test the filters.. 
-# Figure out how to test logging filters..
-
-
-# Need tests for the throttling for sure
-
-# Need a test for the base handlerfilter:
-# using one handlertype
-# using multiple handler types
-
-
-###########
-
-
 """Comprehensive tests for the logging filters in handlers.py.
 
 Tests cover:
@@ -47,7 +12,7 @@
 import logging
 import time
 from threading import Thread
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -79,9 +44,9 @@ def clean_logger():
 
 
 @pytest.fixture
-def log_record():
+def log_record() -> logging.LogRecord:
     """Provide a basic log record for testing."""
-    record = logging.LogRecord(
+    return logging.LogRecord(
         name="test.module",
         level=logging.ERROR,
         pathname="/path/to/test.py",
@@ -90,7 +55,6 @@ def log_record():
         args=(),
         exc_info=None,
     )
-    return record
 
 
 @pytest.fixture
@@ -130,7 +94,9 @@ def mock_ers_handlers():
 class TestBaseHandlerFilter:
     """Tests for BaseHandlerFilter.get_allowed() logic."""
 
-    def test_non_ers_uses_record_handlers_attribute(self, log_record):
+    def test_non_ers_uses_record_handlers_attribute(
+        self, log_record: logging.LogRecord
+    ):
         """Test get_allowed() uses 'handlers' attribute from record for non-ERS."""
         log_record.handlers = [HandlerType.Rich, HandlerType.File]
         filter_obj = BaseHandlerFilter()
@@ -139,7 +105,9 @@ def test_non_ers_uses_record_handlers_attribute(self, log_record):
 
         assert allowed == [HandlerType.Rich, HandlerType.File]
 
-    def test_non_ers_defaults_to_base_handlers(self, log_record):
+    def test_non_ers_defaults_to_base_handlers(
+        self, log_record: logging.LogRecord
+    ):
         """Test get_allowed() falls back to default handlers when attribute missing."""
         # log_record has no 'handlers' attribute
         filter_obj = BaseHandlerFilter()
@@ -151,7 +119,9 @@ def test_non_ers_defaults_to_base_handlers(self, log_record):
         expected_handlers = {HandlerType.Stream, HandlerType.Rich, HandlerType.File}
         assert expected_handlers.issubset(set(allowed))
 
-    def test_ers_path_valid_configuration(self, ers_log_record, mock_ers_handlers):
+    def test_ers_path_valid_configuration(
+        self, ers_log_record: logging.LogRecord, mock_ers_handlers: dict
+    ):
         """Test get_allowed() extracts ERS handlers correctly with valid config."""
         ers_log_record.ers_handlers = mock_ers_handlers
         filter_obj = BaseHandlerFilter()
@@ -160,7 +130,9 @@ def test_ers_path_valid_configuration(self, ers_log_record, mock_ers_handlers):
 
         assert allowed == [HandlerType.Throttle, HandlerType.Protobufstream]
 
-    def test_ers_path_no_matching_level_variable(self, ers_log_record, mock_ers_handlers):
+    def test_ers_path_no_matching_level_variable(
+        self, ers_log_record: logging.LogRecord, mock_ers_handlers: dict
+    ):
         """Test get_allowed() returns None when log level has no ERS mapping."""
         # Set a log level that does not have an ERS equivalent
         ers_log_record.levelno = 25  # Between INFO and WARNING
@@ -194,7 +166,9 @@ def test_list_handler_ids_converted_to_set(self):
         assert isinstance(filter_obj.handler_ids, set)
         assert filter_obj.handler_ids == {HandlerType.Rich, HandlerType.File}
 
-    def test_filter_returns_true_when_handler_in_allowed(self, log_record):
+    def test_filter_returns_true_when_handler_in_allowed(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() returns True when handler_id is in allowed list."""
         log_record.handlers = [HandlerType.Rich, HandlerType.File, HandlerType.Stream]
         filter_obj = HandleIDFilter(HandlerType.Rich)
@@ -203,7 +177,9 @@ def test_filter_returns_true_when_handler_in_allowed(self, log_record):
 
         assert result is True
 
-    def test_filter_returns_false_when_handler_not_in_allowed(self, log_record):
+    def test_filter_returns_false_when_handler_not_in_allowed(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() returns False when handler_id not in allowed."""
         log_record.handlers = [HandlerType.File, HandlerType.Stream]
         filter_obj = HandleIDFilter(HandlerType.Rich)
@@ -212,7 +188,9 @@ def test_filter_returns_false_when_handler_not_in_allowed(self, log_record):
 
         assert result is False
 
-    def test_filter_returns_false_when_get_allowed_returns_none(self, log_record):
+    def test_filter_returns_false_when_get_allowed_returns_none(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() returns False when get_allowed() returns None."""
         filter_obj = HandleIDFilter(HandlerType.Rich)
         filter_obj.get_allowed = MagicMock(return_value=None)
@@ -221,7 +199,9 @@ def test_filter_returns_false_when_get_allowed_returns_none(self, log_record):
 
         assert result is False
 
-    def test_filter_with_multiple_handler_ids(self, log_record):
+    def test_filter_with_multiple_handler_ids(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() with multiple handler_ids checks intersection."""
         log_record.handlers = [HandlerType.Rich, HandlerType.File]
         filter_obj = HandleIDFilter([HandlerType.Rich, HandlerType.Stream])
@@ -231,7 +211,9 @@ def test_filter_with_multiple_handler_ids(self, log_record):
         # Should return True because Rich is in both sets
         assert result is True
 
-    def test_filter_no_intersection_with_multiple_ids(self, log_record):
+    def test_filter_no_intersection_with_multiple_ids(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() returns False when no intersection with multiple ids."""
         log_record.handlers = [HandlerType.File]
         filter_obj = HandleIDFilter([HandlerType.Rich, HandlerType.Stream])
@@ -249,7 +231,9 @@ def test_filter_no_intersection_with_multiple_ids(self, log_record):
 class TestThrottleFilter:
     """Tests for ThrottleFilter throttling and suppression logic."""
 
-    def test_initial_phase_lets_through_first_n_messages(self, log_record):
+    def test_initial_phase_lets_through_first_n_messages(
+        self, log_record: logging.LogRecord
+    ):
         """Test that first N messages pass through without suppression."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=3, time_limit=10)
@@ -259,7 +243,9 @@ def test_initial_phase_lets_through_first_n_messages(self, log_record):
         assert filter_obj.filter(log_record) is True
         assert filter_obj.filter(log_record) is True
 
-    def test_after_initial_threshold_suppresses(self, log_record):
+    def test_after_initial_threshold_suppresses(
+        self, log_record: logging.LogRecord
+    ):
         """Test that messages are suppressed after initial_threshold."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=2, time_limit=10)
@@ -271,7 +257,9 @@ def test_after_initial_threshold_suppresses(self, log_record):
         # 3rd should be suppressed
         assert filter_obj.filter(log_record) is False
 
-    def test_escalating_threshold_doubles_on_report(self, log_record):
+    def test_escalating_threshold_doubles_on_report(
+        self, log_record: logging.LogRecord
+    ):
         """Test that threshold escalates (10->100->1000) when reporting."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=1, time_limit=100)
@@ -288,7 +276,9 @@ def test_escalating_threshold_doubles_on_report(self, log_record):
 
         assert issue_record.threshold == 100  # Escalated from 10
 
-    def test_time_window_reset_resets_counters(self, log_record, monkeypatch):
+    def test_time_window_reset_resets_counters(
+        self, log_record: logging.LogRecord, monkeypatch: pytest.MonkeyPatch
+    ):
         """Test that state resets after time_limit expires."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=1, time_limit=1)
@@ -302,7 +292,9 @@ def test_time_window_reset_resets_counters(self, log_record, monkeypatch):
         # Time advances beyond time_limit with no suppression, reset should allow pass
         assert filter_obj.filter(log_record) is True
 
-    def test_suppressed_counter_increments(self, log_record):
+    def test_suppressed_counter_increments(
+        self, log_record: logging.LogRecord
+    ):
         """Test that suppressed_counter increments for each suppressed message."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=0, time_limit=100)
@@ -317,7 +309,9 @@ def test_suppressed_counter_increments(self, log_record):
             if i > 0:
                 assert issue_record.suppressed_counter >= 0
 
-    def test_throttle_suppression_flag_bypasses_filter(self, log_record):
+    def test_throttle_suppression_flag_bypasses_filter(
+        self, log_record: logging.LogRecord
+    ):
         """Test that _throttle_suppression flag allows suppression messages through."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=0, time_limit=100)
@@ -329,7 +323,9 @@ def test_throttle_suppression_flag_bypasses_filter(self, log_record):
         log_record._throttle_suppression = True
         assert filter_obj.filter(log_record) is True
 
-    def test_get_allowed_returns_none_skips_throttle(self, log_record):
+    def test_get_allowed_returns_none_skips_throttle(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() returns True if get_allowed() returns None."""
         filter_obj = ThrottleFilter()
         filter_obj.get_allowed = MagicMock(return_value=None)
@@ -338,7 +334,9 @@ def test_get_allowed_returns_none_skips_throttle(self, log_record):
         result = filter_obj.filter(log_record)
         assert result is False
 
-    def test_throttle_not_in_allowed_returns_true(self, log_record):
+    def test_throttle_not_in_allowed_returns_true(
+        self, log_record: logging.LogRecord
+    ):
         """Test filter() returns True if Throttle not in allowed handlers."""
         log_record.handlers = [HandlerType.Rich, HandlerType.File]
         filter_obj = ThrottleFilter(initial_threshold=0, time_limit=10)
@@ -358,7 +356,9 @@ def test_timestamp_formatting(self):
         assert formatted.count("-") == 2  # Two dashes for date
         assert formatted.count(":") == 2  # Two colons for time
 
-    def test_different_issues_tracked_separately(self, log_record):
+    def test_different_issues_tracked_separately(
+        self, log_record: logging.LogRecord
+    ):
         """Test that different file:line combinations track state separately."""
         filter_obj = ThrottleFilter(initial_threshold=2, time_limit=10)
 
@@ -390,13 +390,15 @@ def test_different_issues_tracked_separately(self, log_record):
         # Issue 2: suppressed (independent)
         assert filter_obj.filter(record2) is False
 
-    def test_thread_safety_concurrent_issues(self, log_record):
+    def test_thread_safety_concurrent_issues(
+        self, log_record: logging.LogRecord
+    ):
         """Test ThrottleFilter is thread-safe with concurrent logging."""
         filter_obj = ThrottleFilter(initial_threshold=5, time_limit=10)
         log_record.handlers = [HandlerType.Throttle]
         results = []
 
-        def log_messages(record, num_messages):
+        def log_messages(record: logging.LogRecord, num_messages: int) -> None:
             """Log from a thread."""
             for _ in range(num_messages):
                 result = filter_obj.filter(record)
@@ -465,7 +467,7 @@ def test_reset_clears_all_state(self):
 class TestFiltersIntegration:
     """Integration tests with real logger setup."""
 
-    def test_logger_with_handle_id_filter(self, clean_logger):
+    def test_logger_with_handle_id_filter(self, clean_logger: logging.Logger):
         """Test logger with HandleIDFilter allows only specific handlers."""
         stream = io.StringIO()
         handler = logging.StreamHandler(stream)
@@ -490,7 +492,7 @@ def test_logger_with_handle_id_filter(self, clean_logger):
         # Message should appear because Stream is in allowed
         assert "Test message" in stream.getvalue()
 
-    def test_logger_with_throttle_filter(self, clean_logger):
+    def test_logger_with_throttle_filter(self, clean_logger: logging.Logger):
         """Test logger correctly suppresses messages with ThrottleFilter."""
         stream = io.StringIO()
         handler = logging.StreamHandler(stream)
@@ -521,7 +523,7 @@ def test_logger_with_throttle_filter(self, clean_logger):
         # First 2 should appear, then suppression message
         assert output.count("Repeated message") >= 2
 
-    def test_chained_filters(self, clean_logger):
+    def test_chained_filters(self, clean_logger: logging.Logger):
         """Test stacking HandleIDFilter and ThrottleFilter."""
         stream = io.StringIO()
         handler = logging.StreamHandler(stream)
@@ -561,7 +563,7 @@ def test_chained_filters(self, clean_logger):
 class TestEdgeCases:
     """Tests for edge cases and boundary conditions."""
 
-    def test_empty_handlers_list(self, log_record):
+    def test_empty_handlers_list(self, log_record: logging.LogRecord):
         """Test filter behavior with empty handlers list."""
         log_record.handlers = []
         filter_obj = HandleIDFilter(HandlerType.Rich)
@@ -570,7 +572,7 @@ def test_empty_handlers_list(self, log_record):
 
         assert result is False
 
-    def test_none_handlers_attribute(self, log_record):
+    def test_none_handlers_attribute(self, log_record: logging.LogRecord):
         """Test filter when record.handlers is None."""
         log_record.handlers = None
         filter_obj = HandleIDFilter(HandlerType.Rich)
@@ -579,15 +581,17 @@ def test_none_handlers_attribute(self, log_record):
         result = filter_obj.filter(log_record)
         assert result is False
 
-    def test_throttle_with_zero_initial_threshold(self, log_record):
+    def test_throttle_with_zero_initial_threshold(
+        self, log_record: logging.LogRecord
+    ):
         """Test ThrottleFilter with initial_threshold=0."""
         log_record.handlers = [HandlerType.Throttle]
         filter_obj = ThrottleFilter(initial_threshold=0, time_limit=10)
 
         # All messages should be suppressed after first
         assert filter_obj.filter(log_record) is False
 
-    def test_issue_record_key_format(self, log_record):
+    def test_issue_record_key_format(self, log_record: logging.LogRecord):
         """Test that issue_record key is formatted correctly."""
         filter_obj = ThrottleFilter()
 
@@ -596,7 +600,9 @@ def test_issue_record_key_format(self, log_record):
 
         assert isinstance(record, IssueRecord)
 
-    def test_multiple_handler_types_intersection(self, log_record):
+    def test_multiple_handler_types_intersection(
+        self, log_record: logging.LogRecord
+    ):
         """Test set intersection with multiple handler types."""
         log_record.handlers = [
             HandlerType.Rich,
@@ -609,7 +615,9 @@ def test_multiple_handler_types_intersection(self, log_record):
         result = filter_obj.filter(log_record)
         assert result is True
 
-    def test_protobuf_conf_in_ers_handlers(self, ers_log_record, mock_ers_handlers):
+    def test_protobuf_conf_in_ers_handlers(
+        self, ers_log_record: logging.LogRecord, mock_ers_handlers: dict
+    ):
         """Test that ProtobufConf is properly included in ERS configuration."""
         ers_log_record.ers_handlers = mock_ers_handlers
         filter_obj = BaseHandlerFilter()
@@ -618,7 +626,9 @@ def test_protobuf_conf_in_ers_handlers(self, ers_log_record, mock_ers_handlers):
 
         assert HandlerType.Protobufstream in allowed
 
-    def test_suppression_message_includes_count(self, log_record, clean_logger):
+    def test_suppression_message_includes_count(
+        self, clean_logger: logging.Logger
+    ):
         """Test that suppression message includes suppressed count."""
         stream = io.StringIO()
         handler = logging.StreamHandler(stream)