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() 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..cf8c3e3 100644 --- a/tests/test_bench.py +++ b/tests/test_bench.py @@ -4,25 +4,28 @@ import logging from io import StringIO -from typing import Generator, Tuple +from typing import Callable, 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]: +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." @@ -31,19 +34,14 @@ 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, "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(stream, expected) def test_enabled(construct: Tuple[StringIO, EpiLog]) -> None: @@ -51,35 +49,31 @@ 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(): +def test_empty_convert_units() -> None: """Test branch point in Units class to calculate unit conversion method.""" instance = Units() value: float = 1.2 @@ -103,9 +97,9 @@ 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 = NS_UNITS.convert_units(value) + result: Tuple[float, str] = NS_UNITS.convert_units(value) assert result == expected, "Unexpected conversion." @@ -122,7 +116,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..9312bc9 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -5,13 +5,15 @@ 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 + @pytest.mark.parametrize("names", [("a", "b", "c"), ("b", "c", "d", "e")]) @pytest.mark.parametrize("fn_name", ["get_logger", "dispatch"]) @@ -23,7 +25,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 +41,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 +57,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 +72,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 +99,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 +113,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 +155,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 +226,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 +269,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 +286,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 +298,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 +321,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."