Repository: Delgan/loguru
Pull request #196 (2019-12-30 17:13:33)

Retention fix
Addresses issues in #195 

- Ensures directories are never deleted when handling retention
- Ensures log extension is included in filenames when handling retention

gazpachoking: Oops, should have run the tests locally before submitting. Requiring the dots was the one thing I knew could be made more specific, and is what's causing the failure right now. I was trying to avoid changing too much, I think to make it better we need multiple glob patterns.
Delgan: Thanks for the PR!
I guess it's just a small edge case, I fix is probably possible while keeping just one glob pattern. I will take a look.
Delgan: @gazpachoking Ok, you was totally right, multiple glob patterns are required. 

I took the liberty of making changes to your branch as a result of this PR, I hope you don't mind. 

While refactoring the code, I decided to combine the glob pattern with a regex check which I think is more permissive: we want to match `"file.2020-01-01_01-01-01_0000001.log"` (automatic renaming on rotation), but we want to exclude both `"file.foo.log"` and `"foo.file.log"` which is complicated without regex.

At the same time, I realized it was wrong to call `os.path.splitext()` after  `glob.escape()` as it would result in something like `("file\\", ".log")`, so I modified the corresponding code. 

Also, I tried to fix a 100%-theoretical bug that could occur if `"."` was present in the `{time}` spec format (leading to wrong `splitext()`).

This results in code much more convoluted unfortunately, but this should greatly reduce false positives during `retention` process. Does it look good to you?

gazpachoking: It is a bit hard to read and understand the glob and regex bit, but the expanded test suite is clear, and makes me confident it's doing what it's supposed to now. 👍 

Patch:
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 2d5231e..89fb236 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -1,3 +1,9 @@
+`Unreleased`
+============
+
+- Prevent unrelated files and directories to be incorrectly collected thus causing errors during the `retention` process (`#195 <https://github.com/Delgan/loguru/pull/195>`_, thanks `@gazpachoking <https://github.com/gazpachoking>`_).
+
+
 `0.4.0`_ (2019-12-02)
 =====================
 
diff --git a/loguru/__init__.pyi b/loguru/__init__.pyi
index d60a585..3f81803 100644
--- a/loguru/__init__.pyi
+++ b/loguru/__init__.pyi
@@ -303,7 +303,7 @@ class Logger:
         extra: Optional[dict] = ...,
         patcher: Optional[PatcherFunction] = ...,
         activation: Optional[Sequence[ActivationConfig]] = ...
-    ): ...
+    ) -> List[int]: ...
     # @overload should be used to differentiate bytes and str once python/mypy#7781 is fixed
     @staticmethod
     def parse(
diff --git a/loguru/_file_sink.py b/loguru/_file_sink.py
index 34b94e5..fc2ad82 100644
--- a/loguru/_file_sink.py
+++ b/loguru/_file_sink.py
@@ -4,6 +4,7 @@ import glob
 import locale
 import numbers
 import os
+import re
 import shutil
 import string
 from functools import partial
@@ -155,7 +156,7 @@ class FileSink:
         self._kwargs = {**kwargs, "mode": mode, "buffering": buffering, "encoding": self.encoding}
         self._path = str(path)
 
-        self._glob_pattern = self._make_glob_pattern(self._path)
+        self._glob_pattern, self._regex_pattern = self._make_file_patterns(self._path)
         self._rotation_function = self._make_rotation_function(rotation)
         self._retention_function = self._make_retention_function(retention)
         self._compression_function = self._make_compression_function(compression)
@@ -192,15 +193,30 @@ class FileSink:
         self._file = open(new_path, **self._kwargs)
 
     @staticmethod
-    def _make_glob_pattern(path):
-        tokens = string.Formatter().parse(path)
-        parts = (glob.escape(text) + "*" * (name is not None) for text, name, *_ in tokens)
-        root, ext = os.path.splitext("".join(parts))
-        if ext:
-            pattern = root + ".*"
-        else:
-            pattern = root + "*"
-        return pattern
+    def _make_file_patterns(path):
+        def escape(string_, escape_function, rep):
+            tokens = string.Formatter().parse(string_)
+            parts = (escape_function(text) + rep * (name is not None) for text, name, *_ in tokens)
+            return "".join(parts)
+
+        # Cleanup tokens to prevent false-positive in "splitext"
+        pattern = escape(path, lambda x: x, "{}")
+
+        root, ext = os.path.splitext(pattern)
+        basename = os.path.basename(root)
+
+        # Used to quickly list files matching the log filename base (ignoring renaming / extension)
+        glob_pattern = escape(root, glob.escape, "*") + "*"
+
+        # Used to double-check the listed files, ensuring they entirely match a log filename pattern
+        regex_pattern = re.compile(
+            escape(basename, re.escape, r"[\s\S]*")
+            + r"(\.[0-9]+-[0-9]+-[0-9]+_[0-9]+-[0-9]+-[0-9]+_[0-9]+(\.[0-9]+)?)?"
+            + escape(ext, re.escape, r"[\s\S]*")
+            + r"(\.[\s\S]*)?$"
+        )
+
+        return glob_pattern, regex_pattern
 
     @staticmethod
     def _make_rotation_function(rotation):
@@ -341,7 +357,11 @@ class FileSink:
                 self._compression_function(self._file_path)
 
             if self._retention_function is not None:
-                logs = glob.glob(self._glob_pattern)
+                logs = [
+                    file
+                    for file in glob.glob(self._glob_pattern)
+                    if self._regex_pattern.match(os.path.basename(file)) and os.path.isfile(file)
+                ]
                 self._retention_function(logs)
 
         self._file = None


Test patch:
diff --git a/tests/test_filesink_delay.py b/tests/test_filesink_delay.py
index c74605a..bc43dea 100644
--- a/tests/test_filesink_delay.py
+++ b/tests/test_filesink_delay.py
@@ -37,7 +37,7 @@ def test_compression_early_remove(tmpdir):
 
 def test_retention(tmpdir):
     for i in range(5):
-        tmpdir.join("test.%d.log" % i).write("test")
+        tmpdir.join("test.2020-01-01_01-01-%d_000001.log" % i).write("test")
 
     i = logger.add(str(tmpdir.join("test.log")), retention=0, delay=True)
     logger.debug("a")
@@ -48,7 +48,7 @@ def test_retention(tmpdir):
 
 def test_retention_early_remove(tmpdir):
     for i in range(5):
-        tmpdir.join("test.%d.log" % i).write("test")
+        tmpdir.join("test.2020-01-01_01-01-%d_000001.log" % i).write("test")
 
     i = logger.add(str(tmpdir.join("test.log")), retention=0, delay=True)
     logger.remove(i)
diff --git a/tests/test_filesink_retention.py b/tests/test_filesink_retention.py
index fbf4ed7..49f2931 100644
--- a/tests/test_filesink_retention.py
+++ b/tests/test_filesink_retention.py
@@ -36,7 +36,7 @@ def test_retention_count(tmpdir, retention):
     file = tmpdir.join("test.log")
 
     for i in range(retention):
-        tmpdir.join("test.%d.log" % i).write("test")
+        tmpdir.join("test.2011-01-01_01-01-%d_000001.log" % i).write("test")
 
     i = logger.add(str(file), retention=retention)
     logger.debug("test")
@@ -63,24 +63,61 @@ def test_retention_function(tmpdir):
 
 
 def test_managed_files(tmpdir):
-    others = ["test.log", "test.log.1", "test.log.1.gz", "test.log.rar", "test.1.log"]
+    others = [
+        "test.log",
+        "test.log.1",
+        "test.log.1.gz",
+        "test.log.rar",
+        "test.log",
+        "test.2019-11-12_03-22-07_018985.log",
+        "test.2019-11-12_03-22-07_018985.log.tar.gz",
+        "test.2019-11-12_03-22-07_018985.2.log",
+        "test.2019-11-12_03-22-07_018985.2.log.tar.gz",
+    ]
 
     for other in others:
         tmpdir.join(other).write(other)
 
-    i = logger.add(str(tmpdir.join("test.log")), retention=0)
+    i = logger.add(str(tmpdir.join("test.log")), retention=0, catch=False)
     logger.remove(i)
 
     assert len(tmpdir.listdir()) == 0
 
 
 def test_not_managed_files(tmpdir):
-    others = ["test_.log", "_test.log", "tes.log", "te.st.log", "testlog", "test"]
+    others = {
+        "test_.log",
+        "_test.log",
+        "tes.log",
+        "te.st.log",
+        "testlog",
+        "test",
+        "test.tar.gz",
+        "test.logs",
+        "test.foo",
+        "foo.test.log",
+        "test.foo.log",
+        "test.123.log",
+        "test.2019-11-12_03-22-07_018985.abc.log",
+        "test.2019-11-12_03-22-07_018985.123.abc.log",
+    }
 
     for other in others:
         tmpdir.join(other).write(other)
 
-    i = logger.add(str(tmpdir.join("test.log")), retention=0)
+    i = logger.add(str(tmpdir.join("test.log")), retention=0, catch=False)
+    logger.remove(i)
+
+    assert set(f.basename for f in tmpdir.listdir()) == others
+
+
+def test_directories_ignored(tmpdir):
+    others = ["test.log.2", "test.123.log", "test.log.tar.gz", "test.archive"]
+
+    for other in others:
+        tmpdir.join(other).mkdir()
+
+    i = logger.add(str(tmpdir.join("test.log")), retention=0, catch=False)
     logger.remove(i)
 
     assert len(tmpdir.listdir()) == len(others)
@@ -112,6 +149,25 @@ def test_manage_formatted_files(monkeypatch_date, tmpdir):
     assert f3.check(exists=0)
 
 
+def test_date_with_dot_after_extension(monkeypatch_date, tmpdir):
+    file = tmpdir.join("file.{time:YYYY.MM}_log")
+
+    i = logger.add(str(tmpdir.join("file*.log")), retention=0, catch=False)
+    logger.remove(i)
+
+    assert file.check(exists=0)
+
+
+def test_symbol_in_filename(tmpdir):
+    file = tmpdir.join("file123.log")
+    file.write("")
+
+    i = logger.add(str(tmpdir.join("file*.log")), retention=0, catch=False)
+    logger.remove(i)
+
+    assert file.check(exists=1)
+
+
 def test_manage_file_without_extension(tmpdir):
     file = tmpdir.join("file")