Fixes #68: fix filename in Timed mode when hitting size-based

rollover more than once per the timed interval. Thanks @moynihan

This also ensures the unit tests detect this condition.

Author: Preston-Landers

src/concurrent_log_handler/__init__.py

@@ -819,12 +819,13 @@ def doRollover(self):  # noqa: C901, PLR0912
 
         gzip_ext = ".gz" if self.clh.use_gzip else ""
 
-        counter = 0
-        while os.path.exists(dfn + gzip_ext):
-            counter += 1
-            ending = f".{counter - 1}{gzip_ext}"
-            if dfn.endswith(ending):
-                dfn = dfn[: -len(ending)]
+        counter = 1
+        if os.path.exists(dfn + gzip_ext):
+            while os.path.exists(f"{dfn}.{counter}{gzip_ext}"):
+                ending = f".{counter - 1}{gzip_ext}"
+                if dfn.endswith(ending):
+                    dfn = dfn[: -len(ending)]
+                counter += 1
             dfn = f"{dfn}.{counter}"
 
         # if os.path.exists(dfn):
@@ -840,8 +841,7 @@ def doRollover(self):  # noqa: C901, PLR0912
             # Thanks to @moynihan
             for file in self.getFilesToDelete():
                 os.remove(file)
-        # if not self.delay:
-        #     self.stream = self._open()
+
         newRolloverAt = self.computeRollover(currentTime)
         while newRolloverAt <= currentTime:
             newRolloverAt = newRolloverAt + self.interval

tests/stresstest.py

@@ -19,6 +19,7 @@
 import multiprocessing
 import os
 import random
+import re
 import string
 import time
 from dataclasses import dataclass, field
@@ -254,6 +255,19 @@ def run_stress_test(test_opts: TestOptions) -> int:
         f"{rollover_counter.get_value()} - min was {test_opts.min_rollovers})"
     )
 
+    log_path = os.path.join(test_opts.log_dir, test_opts.log_file)
+    all_log_files = glob.glob(f"{log_path}*")
+
+    gzip_ext = "[.]gz" if test_opts.log_opts["use_gzip"] else ""
+
+    # Issue #68 - check for incorrect naming of files when using TimedRotatingFileHandler
+    # and we had to rollover more often than the normal `when` interval due to size limits
+    # If this happens more than once per interval, the names would be like "logfile.2020-01-01.1.2.3.4.5.gz"
+    for log_file in all_log_files:
+        if re.search(r"\.\d+\.\d+" + gzip_ext, str(log_file)):
+            print(f"Error: Incorrect naming of log file: {log_file}")
+            return 1
+
     # If backupCount is less than 10, assume we want specifically to
     # test backupCount. This means, in most cases, we will have deleted
     # some logs files and we should not expect to find all log files.
@@ -264,13 +278,11 @@ def run_stress_test(test_opts: TestOptions) -> int:
     backup_count = test_opts.log_opts.get("backupCount", 0)
     if backup_count > 0 and backup_count < MAGIC_BACKUP_COUNT:
         expect_all_lines = False
-        log_path = os.path.join(test_opts.log_dir, test_opts.log_file)
-        log_files = glob.glob(f"{log_path}*")
         # Check that backupCount was not exceeded.
         # The +1 is because we're counting 'backups' plus the main log file.
-        if len(log_files) != backup_count + 1:
+        if len(all_log_files) != backup_count + 1:
             print(
-                f"Error: {len(log_files)} log files were created but "
+                f"Error: {len(all_log_files)} log files were created but "
                 f"we expected {backup_count + 1}. Could indicate a failure "
                 f"to rotate properly or to delete excessive backups (`backupCount`)."
             )

tests/test_stresstest.py

@@ -102,7 +102,24 @@
             }
         ),
     ),
-    "backupCount=5, use_timed=True, maxBytes=1KiB, interval=2, log_calls=1_000, use_gzip=True": TestOptions(
+    # This checks the issue in Issue #68 - in Timed mode, when we have to rollover more
+    # often than the interval due to size limits, the naming of the files is incorrect.
+    # The check for the incorrect names is done in `run_stress_test()`. The following case can
+    # also check it.
+    "use_timed=True, maxBytes=512B, interval=3, log_calls=2_000, use_gzip=True, num_processes=3": TestOptions(
+        use_timed=True,
+        log_calls=2_000,
+        min_rollovers=5,
+        num_processes=3,
+        log_opts=TestOptions.default_timed_log_opts(
+            {
+                "maxBytes": 512,
+                "interval": 3,
+                "use_gzip": True,
+            }
+        ),
+    ),
+    "backupCount=5, use_timed=True, maxBytes=1KiB, interval=5, log_calls=1_000, use_gzip=True, debug=True": TestOptions(
         use_timed=True,
         log_calls=1_000,
         min_rollovers=5,
@@ -156,7 +173,14 @@
 }
 
 
-@pytest.mark.parametrize("label, test_opts", TEST_CASES.items())
+use_timed_only = False
+
+test_cases = TEST_CASES
+if use_timed_only:
+    test_cases = {label: case for label, case in TEST_CASES.items() if case.use_timed}
+
+
+@pytest.mark.parametrize("label, test_opts", test_cases.items())
 def test_run_stress_test(label: str, test_opts: TestOptions):  # noqa: ARG001
     """Run the stress test with the given options and verify the result."""
     assert run_stress_test(test_opts) == (1 if test_opts.induce_failure else 0)