implement tests for rotating loggers

zariiii9003 · zariiii9003 · commit 9a3a41eadf46 · 2020-08-31T01:10:09.000+02:00
diff --git a/can/io/logger.py b/can/io/logger.py
@@ -119,12 +119,19 @@ class BaseRotatingCanLogger(Listener, ABC):
     namer: Optional[Callable] = None
     rotator: Optional[Callable] = None
     rollover_count: int = 0
-    writer: FileIOMessageWriter
+    _writer: Optional[FileIOMessageWriter] = None
 
     def __init__(self, *args, **kwargs):
         self.writer_args = args
         self.writer_kwargs = kwargs
 
+    @property
+    def writer(self) -> FileIOMessageWriter:
+        if not self._writer:
+            raise ValueError("Attempt to access writer failed.")
+
+        return self._writer
+
     def rotation_filename(self, default_name: StringPathLike):
         """Modify the filename of a log file when rotating.
 
@@ -176,7 +183,7 @@ def on_message_received(self, msg: Message):
 
         self.writer.on_message_received(msg)
 
-    def get_new_writer(self, filename: StringPathLike) -> FileIOMessageWriter:
+    def get_new_writer(self, filename: StringPathLike):
         """Instantiate a new writer.
 
         :param filename:
@@ -193,7 +200,10 @@ def get_new_writer(self, filename: StringPathLike) -> FileIOMessageWriter:
                 f'Log format with suffix"{suffix}" is '
                 f"not supported by {self.__class__.__name__}."
             )
-        return writer_class(filename, *self.writer_args, **self.writer_kwargs)
+        else:
+            self._writer = writer_class(
+                filename, *self.writer_args, **self.writer_kwargs
+            )
 
     def stop(self):
         """Stop handling new messages.
@@ -227,7 +237,7 @@ class SizedRotatingCanLogger(BaseRotatingCanLogger):
 
     Example::
 
-        from can import Notifier, RotatingFileLogger
+        from can import Notifier, SizedRotatingCanLogger
         from can.interfaces.vector import VectorBus
 
         bus = VectorBus(channel=[0], app_name="CANape", fd=True)
@@ -248,7 +258,6 @@ class SizedRotatingCanLogger(BaseRotatingCanLogger):
       * .txt :class:`can.Printer`
 
     The log files may be incomplete until `stop()` is called due to buffering.
-
     """
 
     def __init__(
@@ -261,17 +270,16 @@ def __init__(
         :param max_bytes:
             The size threshold at which a new log file shall be created. If set to 0, no
             rollover will be performed.
-
         """
         super(SizedRotatingCanLogger, self).__init__(*args, **kwargs)
 
         self.base_filename = os.path.abspath(base_filename)
         self.max_bytes = max_bytes
 
-        self.writer = self.get_new_writer(self.base_filename)
+        self.get_new_writer(self.base_filename)
 
     def should_rollover(self, msg: Message) -> bool:
-        if self.max_bytes == 0:
+        if self.max_bytes <= 0:
             return False
 
         if self.writer.file.tell() >= self.max_bytes:
@@ -287,7 +295,7 @@ def do_rollover(self):
         dfn = self.rotation_filename(self._default_name())
         self.rotate(sfn, dfn)
 
-        self.writer = self.get_new_writer(self.base_filename)
+        self.get_new_writer(self.base_filename)
 
     def _default_name(self) -> StringPathLike:
         """Generate the default rotation filename."""
diff --git a/can/logger.py b/can/logger.py
@@ -153,7 +153,9 @@ def main():
     print(f"Can Logger (Started on {datetime.now()})")
 
     if results.file_size:
-        logger = SizedRotatingCanLogger(base_filename=results.log_file, max_bytes=results.file_size)
+        logger = SizedRotatingCanLogger(
+            base_filename=results.log_file, max_bytes=results.file_size
+        )
     else:
         logger = Logger(filename=results.log_file)
 
diff --git a/test/test_rotating_loggers.py b/test/test_rotating_loggers.py
@@ -0,0 +1,268 @@
+#!/usr/bin/env python
+# coding: utf-8
+
+"""
+Test rotating loggers
+"""
+import os
+from pathlib import Path
+import tempfile
+from unittest.mock import Mock
+
+import pytest
+
+import can
+from .data.example_data import generate_message
+
+
+class TestBaseRotatingCanLogger:
+    @staticmethod
+    def _get_instance(*args, **kwargs):
+        class SubClass(can.io.BaseRotatingCanLogger):
+            """Subclass that implements abstract methods for testing."""
+
+            def should_rollover(self, msg):
+                ...
+
+            def do_rollover(self):
+                ...
+
+        return SubClass(*args, **kwargs)
+
+    def test_import(self):
+        assert hasattr(can.io, "BaseRotatingCanLogger")
+
+    def test_attributes(self):
+        assert issubclass(can.io.BaseRotatingCanLogger, can.Listener)
+        assert hasattr(can.io.BaseRotatingCanLogger, "supported_writers")
+        assert hasattr(can.io.BaseRotatingCanLogger, "namer")
+        assert hasattr(can.io.BaseRotatingCanLogger, "rotator")
+        assert hasattr(can.io.BaseRotatingCanLogger, "rollover_count")
+        assert hasattr(can.io.BaseRotatingCanLogger, "writer")
+        assert hasattr(can.io.BaseRotatingCanLogger, "rotation_filename")
+        assert hasattr(can.io.BaseRotatingCanLogger, "rotate")
+        assert hasattr(can.io.BaseRotatingCanLogger, "on_message_received")
+        assert hasattr(can.io.BaseRotatingCanLogger, "get_new_writer")
+        assert hasattr(can.io.BaseRotatingCanLogger, "stop")
+        assert hasattr(can.io.BaseRotatingCanLogger, "should_rollover")
+        assert hasattr(can.io.BaseRotatingCanLogger, "do_rollover")
+
+    def test_supported_writers(self):
+        supported_writers = can.io.BaseRotatingCanLogger.supported_writers
+        assert supported_writers[".asc"] == can.ASCWriter
+        assert supported_writers[".blf"] == can.BLFWriter
+        assert supported_writers[".csv"] == can.CSVWriter
+        assert supported_writers[".log"] == can.CanutilsLogWriter
+        assert supported_writers[".txt"] == can.Printer
+
+    def test_get_new_writer(self):
+        logger_instance = self._get_instance()
+
+        # access to non existing writer shall raise ValueError
+        with pytest.raises(ValueError):
+            _ = logger_instance.writer
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.ASC"))
+            assert isinstance(logger_instance.writer, can.ASCWriter)
+            logger_instance.stop()
+
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.BLF"))
+            assert isinstance(logger_instance.writer, can.BLFWriter)
+            logger_instance.stop()
+
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.CSV"))
+            assert isinstance(logger_instance.writer, can.CSVWriter)
+            logger_instance.stop()
+
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.LOG"))
+            assert isinstance(logger_instance.writer, can.CanutilsLogWriter)
+            logger_instance.stop()
+
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.TXT"))
+            assert isinstance(logger_instance.writer, can.Printer)
+            logger_instance.stop()
+
+    def test_rotation_filename(self):
+        logger_instance = self._get_instance()
+
+        default_name = "default"
+        assert logger_instance.rotation_filename(default_name) == "default"
+
+        logger_instance.namer = lambda x: x + "_by_namer"
+        assert logger_instance.rotation_filename(default_name) == "default_by_namer"
+
+    def test_rotate(self):
+        logger_instance = self._get_instance()
+
+        # test without rotator
+        with tempfile.TemporaryDirectory() as temp_dir:
+            source = os.path.join(temp_dir, "source.txt")
+            dest = os.path.join(temp_dir, "dest.txt")
+
+            assert os.path.exists(source) is False
+            assert os.path.exists(dest) is False
+
+            logger_instance.get_new_writer(source)
+            logger_instance.stop()
+
+            assert os.path.exists(source) is True
+            assert os.path.exists(dest) is False
+
+            logger_instance.rotate(source, dest)
+
+            assert os.path.exists(source) is False
+            assert os.path.exists(dest) is True
+
+        # test with rotator
+        rotator_func = Mock()
+        logger_instance.rotator = rotator_func
+        with tempfile.TemporaryDirectory() as temp_dir:
+            source = os.path.join(temp_dir, "source.txt")
+            dest = os.path.join(temp_dir, "dest.txt")
+
+            assert os.path.exists(source) is False
+            assert os.path.exists(dest) is False
+
+            logger_instance.get_new_writer(source)
+            logger_instance.stop()
+
+            assert os.path.exists(source) is True
+            assert os.path.exists(dest) is False
+
+            logger_instance.rotate(source, dest)
+            rotator_func.assert_called_with(source, dest)
+
+            # assert that no rotation was performed since rotator_func
+            # does not do anything
+            assert os.path.exists(source) is True
+            assert os.path.exists(dest) is False
+
+    def test_stop(self):
+        """Test if stop() method of writer is called."""
+        logger_instance = self._get_instance()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.ASC"))
+
+            # replace stop method of writer with Mock
+            org_stop = logger_instance.writer.stop
+            mock_stop = Mock()
+            logger_instance.writer.stop = mock_stop
+
+            logger_instance.stop()
+            mock_stop.assert_called()
+
+            # close file.ASC to enable cleanup of temp_dir
+            org_stop()
+
+    def test_on_message_received(self):
+        logger_instance = self._get_instance()
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger_instance.get_new_writer(os.path.join(temp_dir, "file.ASC"))
+
+            """Test without rollover"""
+            should_rollover = Mock(return_value=False)
+            do_rollover = Mock()
+            writers_on_message_received = Mock()
+
+            logger_instance.should_rollover = should_rollover
+            logger_instance.do_rollover = do_rollover
+            logger_instance.writer.on_message_received = writers_on_message_received
+
+            msg = generate_message(0x123)
+            logger_instance.on_message_received(msg)
+
+            should_rollover.assert_called_with(msg)
+            do_rollover.assert_not_called()
+            writers_on_message_received.assert_called_with(msg)
+
+            """Test with rollover"""
+            should_rollover = Mock(return_value=True)
+            do_rollover = Mock()
+            writers_on_message_received = Mock()
+
+            logger_instance.should_rollover = should_rollover
+            logger_instance.do_rollover = do_rollover
+            logger_instance.writer.on_message_received = writers_on_message_received
+
+            msg = generate_message(0x123)
+            logger_instance.on_message_received(msg)
+
+            should_rollover.assert_called_with(msg)
+            do_rollover.assert_called()
+            writers_on_message_received.assert_called_with(msg)
+
+            # stop writer to enable cleanup of temp_dir
+            logger_instance.stop()
+
+
+class TestSizedRotatingCanLogger:
+    def test_import(self):
+        assert hasattr(can.io, "SizedRotatingCanLogger")
+        assert hasattr(can, "SizedRotatingCanLogger")
+
+    def test_attributes(self):
+        assert issubclass(can.SizedRotatingCanLogger, can.io.BaseRotatingCanLogger)
+        assert hasattr(can.SizedRotatingCanLogger, "supported_writers")
+        assert hasattr(can.SizedRotatingCanLogger, "namer")
+        assert hasattr(can.SizedRotatingCanLogger, "rotator")
+        assert hasattr(can.SizedRotatingCanLogger, "should_rollover")
+        assert hasattr(can.SizedRotatingCanLogger, "do_rollover")
+
+    def test_create_instance(self):
+        base_filename = "mylogfile.ASC"
+        max_bytes = 512
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger_instance = can.SizedRotatingCanLogger(
+                base_filename=os.path.join(temp_dir, base_filename), max_bytes=max_bytes
+            )
+            assert Path(logger_instance.base_filename).name == base_filename
+            assert logger_instance.max_bytes == max_bytes
+            assert logger_instance.rollover_count == 0
+            assert isinstance(logger_instance.writer, can.ASCWriter)
+
+            logger_instance.stop()
+
+    def test_should_rollover(self):
+        base_filename = "mylogfile.ASC"
+        max_bytes = 512
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger_instance = can.SizedRotatingCanLogger(
+                base_filename=os.path.join(temp_dir, base_filename), max_bytes=max_bytes
+            )
+            msg = generate_message(0x123)
+            do_rollover = Mock()
+            logger_instance.do_rollover = do_rollover
+
+            logger_instance.writer.file.tell = Mock(return_value=511)
+            assert logger_instance.should_rollover(msg) is False
+            logger_instance.on_message_received(msg)
+            do_rollover.assert_not_called()
+
+            logger_instance.writer.file.tell = Mock(return_value=512)
+            assert logger_instance.should_rollover(msg) is True
+            logger_instance.on_message_received(msg)
+            do_rollover.assert_called()
+
+            logger_instance.stop()
+
+    def test_logfile_size(self):
+        base_filename = "mylogfile.ASC"
+        max_bytes = 1024
+        msg = generate_message(0x123)
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger_instance = can.SizedRotatingCanLogger(
+                base_filename=os.path.join(temp_dir, base_filename), max_bytes=max_bytes
+            )
+            for _ in range(128):
+                logger_instance.on_message_received(msg)
+
+            for file_path in os.listdir(temp_dir):
+                assert os.path.getsize(os.path.join(temp_dir, file_path)) <= 1100
+
+            logger_instance.stop()
