From 2390f437f47832262c1245e44ee75a565e47f8cb Mon Sep 17 00:00:00 2001 From: Spill-Tea Date: Mon, 13 Apr 2026 22:12:42 -0700 Subject: [PATCH 1/5] tests(tests): Improve static typing, and extract common utilities for unit tests. --- tests/conftest.py | 7 +++ tests/test_bench.py | 43 +++++++------- tests/test_manager.py | 126 ++++++++++++++++++++---------------------- 3 files changed, 88 insertions(+), 88 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index f24f5c0..eae3b1d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ import logging import os +from io import IOBase from typing import Callable, Generator, Optional import pytest @@ -55,3 +56,9 @@ def builder(*args, **kwargs) -> EpiLog: # Required for Intentionally Failed Tests if instance is not None: teardown_epilogs(instance) + + +def _assert_msg_in_output(stream: IOBase, msg: str) -> None: + stream.seek(0) + output: str = stream.read() + assert msg in output, "Message not found in output stream after logging." diff --git a/tests/test_bench.py b/tests/test_bench.py index 99bd477..09b4682 100644 --- a/tests/test_bench.py +++ b/tests/test_bench.py @@ -4,13 +4,15 @@ import logging from io import StringIO -from typing import Generator, Tuple +from typing import Dict, Generator, Tuple import pytest from EpiLog import EpiLog from EpiLog.benchmark import NS_UNITS, BenchMark, Units +from .conftest import _assert_msg_in_output + @pytest.fixture def construct(build_manager) -> Generator[Tuple[StringIO, EpiLog], None, None]: @@ -31,19 +33,20 @@ def test_empty_benchmark(construct: Tuple[StringIO, EpiLog]) -> None: stream, manager = construct manager.level = logging.INFO - log = manager.get_logger("test") - msg = "I'm positively bedeviled with meetings et cetera" - expected = f"INFO | {msg}" + log: logging.Logger = manager.get_logger("test") + msg: str = "I'm positively bedeviled with meetings et cetera" + expected: str = f"INFO | {msg}" with BenchMark(log, msg) as b: assert b.enabled, "Expected Logging to be Enabled on Benchmark Class." - stream.seek(0) - output = stream.read() + _assert_msg_in_output(stream, expected) + # stream.seek(0) + # output = stream.read() - assert msg in output, "Message not Found in output stream after logging." - assert expected in output, "Message Format not found in Output Stream." - assert "Traceback" not in output, "Error raised during use of context manager." + # assert msg in output, "Message not Found in output stream after logging." + # assert expected in output, "Message Format not found in Output Stream." + # assert "Traceback" not in output, "Error raised during use of context manager." def test_enabled(construct: Tuple[StringIO, EpiLog]) -> None: @@ -51,32 +54,28 @@ def test_enabled(construct: Tuple[StringIO, EpiLog]) -> None: stream, manager = construct manager.level = logging.CRITICAL - log = manager.get_logger("disabled") - msg = "That's exactly the kind of paranoia that makes me weary of spending time " - msg += "with you." + log: logging.Logger = manager.get_logger("disabled") + msg: str = "That's exactly the kind of paranoia that makes me weary of spending" + msg += " time with you." with BenchMark(log, msg, logging.DEBUG) as b: assert not b.enabled, "Expected Logging to be Disabled on Benchmark Class." - stream.seek(0) - output = stream.read() - assert msg not in output, "Message Found in output stream after disabled Benchmark." + with pytest.raises(AssertionError): + _assert_msg_in_output(stream, msg) def test_error(construct: Tuple[StringIO, EpiLog]) -> None: """Test that an error message is emitted when one occurs during context.""" stream, manager = construct manager.level = logging.DEBUG - log = manager.get_logger("errors") + log: logging.Logger = manager.get_logger("errors") with pytest.raises(AssertionError): with BenchMark(log, "msg", logging.DEBUG): assert False, "We are Intentionally raising an error" # noqa: B011 - stream.seek(0) - output = stream.read() - assert "ERROR" in output, "Expected Error Level Log Emitted" - assert "Traceback" in output, "Expected Log message to include traceback info." + _assert_msg_in_output(stream, "ERROR") def test_empty_convert_units(): @@ -105,7 +104,7 @@ def test_empty_convert_units(): ) def test_units_convert_units(value: int, expected: float): """Test that we correctly convert units to largest relevant bin.""" - result = NS_UNITS.convert_units(value) + result: Tuple[float, str] = NS_UNITS.convert_units(value) assert result == expected, "Unexpected conversion." @@ -122,7 +121,7 @@ def test_units_convert_units(value: int, expected: float): ) def test_breakdown(value: int, expected: dict) -> None: """Test Breakdown of ns into appropriate time bins.""" - result = NS_UNITS.breakdown_units(value) + result: Dict[str, int] = NS_UNITS.breakdown_units(value) container = dict((i.unit, 0) for i in NS_UNITS) container.update(expected) diff --git a/tests/test_manager.py b/tests/test_manager.py index e261d88..cd9da16 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -5,13 +5,21 @@ import logging import sys from io import IOBase, StringIO -from typing import Callable, Tuple, Union +from typing import Callable, Optional, Tuple, Union import pytest from EpiLog import EpiLog from EpiLog.manager import defaultFormat +from .conftest import _assert_msg_in_output + + +# def _assert_msg_in_output(stream: IOBase, msg: str) -> None: +# stream.seek(0) +# output: str = stream.read() +# assert msg in output, "Message not found in output stream after logging." + @pytest.mark.parametrize("names", [("a", "b", "c"), ("b", "c", "d", "e")]) @pytest.mark.parametrize("fn_name", ["get_logger", "dispatch"]) @@ -23,7 +31,7 @@ def test_get_logger( """Tests that we correctly construct a named logger.""" manager: EpiLog = build_manager() assert len(manager.loggers) == 0 - function = getattr(manager, fn_name) + function: Callable[[str], logging.Logger] = getattr(manager, fn_name) for n in names: function(n) assert len(manager.loggers) == len(set(names)) @@ -39,12 +47,13 @@ def test_get_logger_name_caches(build_manager: Callable[..., EpiLog]) -> None: log_b: logging.Logger = manager.get_logger(name) assert id(log_a) == id(log_b), "Expected same object." + assert len(manager.loggers) == 1, "Expected a single key entry in logger registry." def test_logging(build_manager: Callable[..., EpiLog]) -> None: """Makes certain no errors are raised while logging.""" manager: EpiLog = build_manager() - log = manager.get_logger("test") + log: logging.Logger = manager.get_logger("test") log.info("Bob boop beep") @@ -54,15 +63,11 @@ def test_stream(build_manager: Callable[..., EpiLog]) -> None: handler = logging.StreamHandler(stream) manager: EpiLog = build_manager(stream=handler) - log = manager.get_logger("test") - message = "You are blind to reality and for that I am most proud" + log: logging.Logger = manager.get_logger("test") + message: str = "You are blind to reality and for that I am most proud" log.info(message) - stream.seek(0) - output = stream.read() - stream.close() - - assert message in output, "Message not Found in output stream after logging" + _assert_msg_in_output(stream, message) def test_level_change(build_manager: Callable[..., EpiLog]) -> None: @@ -73,24 +78,19 @@ def test_level_change(build_manager: Callable[..., EpiLog]) -> None: manager: EpiLog = build_manager(level=logging.INFO, stream=handler) log: logging.Logger = manager.get_logger("test") - message = ( + message: str = ( "I would say that he's blessedly unburdened with the complications of a" " university education." ) log.debug(message) - stream.seek(0) - output = stream.read() - assert message not in output, "Message should not have been Received" + assert stream.tell() == 0, "Expected no output at current level." + with pytest.raises(AssertionError): + _assert_msg_in_output(stream, message) # Then repeat after Changing level manager.level = logging.DEBUG log.debug(message) - - stream.seek(0) - output = stream.read() - stream.close() - - assert message in output, "Message not Found in output stream after logging" + _assert_msg_in_output(stream, message) def test_format_change(build_manager: Callable[..., EpiLog]) -> None: @@ -105,11 +105,7 @@ def test_format_change(build_manager: Callable[..., EpiLog]) -> None: message = "I have no idea why all of this is happening or how to control it." log.info(message) - stream.seek(0) - output = stream.read() - stream.close() - - assert f"INFO | {message}\n" == output, "Expected Message format to match." + _assert_msg_in_output(stream, f"INFO | {message}\n") def test_stream_change(build_manager: Callable[..., EpiLog]) -> None: @@ -123,44 +119,37 @@ def test_stream_change(build_manager: Callable[..., EpiLog]) -> None: stream=handler_a, formatter=logging.Formatter("%(levelname)s | %(message)s"), ) - log = manager.get_logger("test") + log: logging.Logger = manager.get_logger("test") # Modify Stream manager.stream = handler_b assert manager.stream == handler_b, "Unexpected Stream." - message = ( + message: str = ( "You humans have so many emotions! You only need two: anger and confusion!" ) - expected = f"INFO | {message}\n" + expected: str = f"INFO | {message}\n" log.info(message) - # Test first stream - stream_a.seek(0) - output_a = stream_a.read() - stream_a.close() - assert ( - output_a != expected - ), "Did not Expect message to be written to previous stream." + # Test first stream (where we do not expect message output) + with pytest.raises(AssertionError): + _assert_msg_in_output(stream_a, message) # Test Second, Expected stream - stream_b.seek(0) - output_b = stream_b.read() - stream_b.close() - assert output_b == expected, "Expected message to be written to current stream." + _assert_msg_in_output(stream_b, expected) def test_stream_change_to_none(build_manager: Callable[..., EpiLog]) -> None: """Confirm that attempts to set the stream to none result in default stream.""" manager: EpiLog = build_manager() assert manager.stream is not None, "Expected Default Stream to instantiate." - previous_id = id(manager.stream) + previous_id: int = id(manager.stream) manager.stream = None # type: ignore[assignment] assert manager.stream is not None, "Expected Default Stream to kick in." - assert isinstance( - manager.stream, (logging.Handler, logging.Filterer) - ), "Unexpected stream type." + assert isinstance(manager.stream, (logging.Handler, logging.Filterer)), ( + "Unexpected stream type." + ) assert id(manager.stream) != previous_id, "Expected new stream instance." @@ -172,7 +161,10 @@ def test_stream_change_to_none(build_manager: Callable[..., EpiLog]) -> None: logging.Formatter("%(name)s | %(levelname)s | %(message)s"), ], ) -def test_format_instantiation(f, build_manager: Callable[..., EpiLog]) -> None: +def test_format_instantiation( + f: Optional[logging.Formatter], + build_manager: Callable[..., EpiLog], +) -> None: """Tests expected behaviors when instantiating with Formatters.""" manager: EpiLog = build_manager(formatter=f) if f is not None: @@ -240,42 +232,42 @@ def test_handlers_error(build_manager: Callable[..., EpiLog]) -> None: def test_get_log_by_name(build_manager: Callable[..., EpiLog]) -> None: """Tests the special dunder method __get_item__ works as expected.""" manager: EpiLog = build_manager() - name = "get_item" + name: str = "get_item" log: logging.Logger = manager.get_logger(name) assert manager[name] == log, "Expected to retrieve the same logger via get item." def _confirm_removal(cls: EpiLog, name: Union[str, logging.Logger]) -> None: - if isinstance(name, logging.Logger): - _name = name.name - else: - _name = name - + _name: str = name.name if isinstance(name, logging.Logger) else name assert _name in cls.loggers, "Expected logger to be registered" - assert ( - _name in logging.Logger.manager.loggerDict - ), "Expected logger to be in logging module registry" + assert _name in logging.Logger.manager.loggerDict, ( + "Expected logger to be in logging module registry" + ) cls.remove(name) assert _name not in cls.loggers, "Expected logger to be removed locally" - assert ( - _name not in logging.Logger.manager.loggerDict - ), "Expected logger to be removed from logging module registry" + assert _name not in logging.Logger.manager.loggerDict, ( + "Expected logger to be removed from logging module registry" + ) # Confirm current stream is not closed if hasattr(cls.stream, "_closed"): # only available >=python3.10 - assert not cls.stream._closed, "Expected current stream to remain open." + assert not cls.stream._closed, ( # pyright: ignore + "Expected current stream to remain open." + ) if hasattr(cls.stream, "stream"): - assert not cls.stream.stream.closed, "Expected current stream to remain open." + assert not cls.stream.stream.closed, ( # pyright: ignore + "Expected current stream to remain open." + ) def test_log_removal_by_name(build_manager: Callable[..., EpiLog]) -> None: """Test we can remove a logger by providing the name of the logger.""" manager: EpiLog = build_manager() - name = "removal_by_name" + name: str = "removal_by_name" manager.get_logger(name) _confirm_removal(manager, name) @@ -283,7 +275,7 @@ def test_log_removal_by_name(build_manager: Callable[..., EpiLog]) -> None: def test_log_removal_by_logger(build_manager: Callable[..., EpiLog]) -> None: """Test we can remove a logger by providing the logger itself.""" manager: EpiLog = build_manager() - name = "removal_by_logger" + name: str = "removal_by_logger" log: logging.Logger = manager.get_logger(name) _confirm_removal(manager, log) @@ -300,7 +292,9 @@ def _handle_second_removal( _confirm_removal(manager, log) if hasattr(second_handler, "_closed"): - assert second_handler._closed, "Expected additional Handler to be closed." + assert second_handler._closed, ( # pyright: ignore + "Expected additional Handler to be closed." + ) return second_handler @@ -310,13 +304,13 @@ def test_log_removal_with_additional_handler( ) -> None: """Test that additional handlers and their streams are closed on removal.""" manager: EpiLog = build_manager() - name = "removal_by_logger" + name: str = "removal_by_logger" log: logging.Logger = manager.get_logger(name) with StringIO() as stream: - second_handler = _handle_second_removal(stream, log, manager) + second: logging.StreamHandler = _handle_second_removal(stream, log, manager) assert stream.closed, "Expected stream to be closed." - assert second_handler.stream.closed, "Expected stream to be closed." + assert second.stream.closed, "Expected stream to be closed." @pytest.mark.parametrize( @@ -333,9 +327,9 @@ def test_log_removal_with_protected_streams( ) -> None: """Test additional handler is closed on removal, but sys streams remain open.""" manager: EpiLog = build_manager() - name = "removal_by_logger" + name: str = "removal_by_logger" log: logging.Logger = manager.get_logger(name) - second_handler = _handle_second_removal(stream, log, manager) + second_handler: logging.StreamHandler = _handle_second_removal(stream, log, manager) assert not stream.closed, "Expected protected stream to remain open." assert not second_handler.stream.closed, "Expected protected stream to remain open." From 9dc9012ae2f5688576d70dd341f02306e4a8bb72 Mon Sep 17 00:00:00 2001 From: Spill-Tea Date: Mon, 13 Apr 2026 22:13:00 -0700 Subject: [PATCH 2/5] typing(benchmark): Use Self typing, using typing_extensions for python versions < 3.11 --- requirements.txt | 1 + src/EpiLog/benchmark.py | 9 ++++++++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index e69de29..3d89d21 100644 --- a/requirements.txt +++ b/requirements.txt @@ -0,0 +1 @@ +typing-extensions; python_version < "3.11" \ No newline at end of file diff --git a/src/EpiLog/benchmark.py b/src/EpiLog/benchmark.py index 5c83d62..010976c 100644 --- a/src/EpiLog/benchmark.py +++ b/src/EpiLog/benchmark.py @@ -29,6 +29,13 @@ from typing import Dict, Iterator, Tuple, Union +# Python < 3.11 support +try: + from typing import Self +except ImportError: + from typing_extensions import Self + + @dataclass class Unit: """Unit comparing string id to a modifier of the next unit. @@ -145,7 +152,7 @@ def __init__( self.description = description self.t0 = 0 - def __enter__(self) -> "BenchMark": + def __enter__(self) -> Self: if self.enabled: self.t0 = perf_counter_ns() From 6a9ee84cc081e4dcfc083950d3307fc2891e1246 Mon Sep 17 00:00:00 2001 From: Spill-Tea Date: Mon, 13 Apr 2026 22:13:45 -0700 Subject: [PATCH 3/5] chore(test.bench): Remove commented code. --- tests/test_bench.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/tests/test_bench.py b/tests/test_bench.py index 09b4682..5247ab4 100644 --- a/tests/test_bench.py +++ b/tests/test_bench.py @@ -41,12 +41,6 @@ def test_empty_benchmark(construct: Tuple[StringIO, EpiLog]) -> None: assert b.enabled, "Expected Logging to be Enabled on Benchmark Class." _assert_msg_in_output(stream, expected) - # stream.seek(0) - # output = stream.read() - - # assert msg in output, "Message not Found in output stream after logging." - # assert expected in output, "Message Format not found in Output Stream." - # assert "Traceback" not in output, "Error raised during use of context manager." def test_enabled(construct: Tuple[StringIO, EpiLog]) -> None: From 330c056845281204921802adb2bacd283f92265c Mon Sep 17 00:00:00 2001 From: Spill-Tea Date: Mon, 13 Apr 2026 22:18:46 -0700 Subject: [PATCH 4/5] chore(test.bench): Minor additional static typing improvements. --- tests/test_bench.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/tests/test_bench.py b/tests/test_bench.py index 5247ab4..cf8c3e3 100644 --- a/tests/test_bench.py +++ b/tests/test_bench.py @@ -4,7 +4,7 @@ import logging from io import StringIO -from typing import Dict, Generator, Tuple +from typing import Callable, Dict, Generator, Tuple import pytest @@ -15,16 +15,17 @@ @pytest.fixture -def construct(build_manager) -> Generator[Tuple[StringIO, EpiLog], None, None]: +def construct( + build_manager: Callable[..., EpiLog], +) -> Generator[Tuple[StringIO, EpiLog], None, None]: """Construct EpiLog Log Manager with standard Fixings.""" - stream = StringIO() - handler = logging.StreamHandler(stream) - form = logging.Formatter("%(levelname)s | %(message)s") - manager = build_manager(stream=handler, formatter=form) + with StringIO() as stream: + handler = logging.StreamHandler(stream) + form = logging.Formatter("%(levelname)s | %(message)s") + manager: EpiLog = build_manager(stream=handler, formatter=form) - yield stream, manager + yield stream, manager - stream.close() assert stream.closed, "Expected Stream to be Closed." @@ -72,7 +73,7 @@ def test_error(construct: Tuple[StringIO, EpiLog]) -> None: _assert_msg_in_output(stream, "ERROR") -def test_empty_convert_units(): +def test_empty_convert_units() -> None: """Test branch point in Units class to calculate unit conversion method.""" instance = Units() value: float = 1.2 @@ -96,7 +97,7 @@ def test_empty_convert_units(): (604_800_000_000_000, (1.0, "weeks")), ], ) -def test_units_convert_units(value: int, expected: float): +def test_units_convert_units(value: int, expected: float) -> None: """Test that we correctly convert units to largest relevant bin.""" result: Tuple[float, str] = NS_UNITS.convert_units(value) assert result == expected, "Unexpected conversion." From 6b6a2107e895899edbceb2262ad9c1ae847622a7 Mon Sep 17 00:00:00 2001 From: Spill-Tea Date: Mon, 13 Apr 2026 22:23:54 -0700 Subject: [PATCH 5/5] chore(tests): minor linting. --- tests/test_manager.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/tests/test_manager.py b/tests/test_manager.py index cd9da16..9312bc9 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -15,12 +15,6 @@ from .conftest import _assert_msg_in_output -# def _assert_msg_in_output(stream: IOBase, msg: str) -> None: -# stream.seek(0) -# output: str = stream.read() -# assert msg in output, "Message not found in output stream after logging." - - @pytest.mark.parametrize("names", [("a", "b", "c"), ("b", "c", "d", "e")]) @pytest.mark.parametrize("fn_name", ["get_logger", "dispatch"]) def test_get_logger(