Remove purge log's randomized behavior (#7307)

* Remove purge logs randomized behavior

Remove purge logs from being fired with a percent chance, use a deterministic approach instead.

* Remove unnecessary list actions

Remove unnecessary list pop's

* Remove unnecessary list slicing

Remove unnecessary list slicing

* Added check for log_dir

Added check for log_dir

* Moved log_dir test before config call

Moved log_dir test before config call

* Create unittest for logger

Create unittest for logger

* Create log for unittest

Create log for unittest

* Create log for unittest

Create log for unittest

* Create log for unittest

Create log for unittest

* Update logger tests to use pytest

Update logger tests to use pytest

Co-authored-by: matham <moiein2000@gmail.com>

* Readability update

Make list comprehension more readable

Co-authored-by: matham <moiein2000@gmail.com>

* Readability update

Increase readability of comment

Co-authored-by: matham <moiein2000@gmail.com>

* Test logs will be created dynamically

Test logs will be created dynamically

* Test logs will be created dynamically

Test logs will be created dynamically

* Test logs will be created dynamically

Test logs will be created dynamically

* Create test_logs folder for logging tests

Create test_logs folder for logging tests

* Fix tests.

* Not all OS prevent open files from being deleted.

* Add missing_ok

Set missing_ok to True. If a file gets deleted by the user which has been queued to be deleted by kivy, the unlink command will throw an unnecessary Exception.

* Sometimes open file is counted in remaining.

* Make timestamp sorting more robust.

* Catch exception that missing_ok ignores.

Co-authored-by: matham <moiein2000@gmail.com>

kivy/logger.py

@@ -91,6 +91,7 @@ import sys
 import copy
 from random import randint
 from functools import partial
+import pathlib
 
 import kivy
 
@@ -150,47 +151,40 @@ class FileHandler(logging.Handler):
     encoding = 'utf-8'
 
     def purge_logs(self):
-        '''Purge log is called randomly to prevent the log directory from being
+        """Purge logs which exceed the maximum amount of log files,
-        filled by lots and lots of log files.
+        starting with the oldest creation timestamp (or edit-timestamp on Linux)
-        You've a chance of 1 in 20 that purge log will be fired.
+        """
-        '''
+
-        if randint(0, 20) != 0:
-            return
         if not self.log_dir:
             return
 
         from kivy.config import Config
-        maxfiles = Config.getint('kivy', 'log_maxfiles')
+        maxfiles = Config.getint("kivy", "log_maxfiles")
 
-        if maxfiles < 0:
+        # Get path to log directory
+        log_dir = pathlib.Path(self.log_dir)
+
+        if maxfiles < 0:  # No log file limit set
             return
 
-        Logger.info('Logger: Purge log fired. Analysing...')
+        Logger.info("Logger: Purge log fired. Processing...")
-        join = os.path.join
-        unlink = os.unlink
 
-        # search all log files
+        # Get all files from log directory and corresponding creation timestamps
-        lst = [join(self.log_dir, x) for x in os.listdir(self.log_dir)]
+        files = [(item, item.stat().st_ctime)
-        if len(lst) > maxfiles:
+                 for item in log_dir.iterdir() if item.is_file()]
-            # get creation time on every files
+        # Sort files by ascending timestamp
-            lst = [{'fn': x, 'ctime': os.path.getctime(x)} for x in lst]
+        files.sort(key=lambda x: x[1])
 
-            # sort by date
+        for file, _ in files[:(-maxfiles or len(files))]:
-            lst = sorted(lst, key=lambda x: x['ctime'])
+            # More log files than allowed maximum,
-
+            # delete files, starting with oldest creation timestamp
-            # get the oldest (keep last maxfiles)
+            # (or edit-timestamp on Linux)
-            lst = lst[:-maxfiles] if maxfiles else lst
-            Logger.info('Logger: Purge %d log files' % len(lst))
-
-            # now, unlink every file in the list
-            for filename in lst:
             try:
-                    unlink(filename['fn'])
+                file.unlink()
-                except PermissionError as e:
+            except (PermissionError, FileNotFoundError) as e:
-                    Logger.info('Logger: Skipped file {0}, {1}'.
+                Logger.info(f"Logger: Skipped file {file}, {repr(e)}")
-                                format(filename['fn'], e))
 
-        Logger.info('Logger: Purge finished!')
+        Logger.info("Logger: Purge finished!")
 
     def _configure(self, *largs, **kwargs):
         from time import strftime
@@ -220,8 +214,9 @@ class FileHandler(logging.Handler):
 
         if FileHandler.filename == filename and FileHandler.fd is not None:
             return
+
         FileHandler.filename = filename
-        if FileHandler.fd is not None:
+        if FileHandler.fd not in (None, False):
             FileHandler.fd.close()
         FileHandler.fd = open(filename, 'w', encoding=FileHandler.encoding)
         Logger.info('Logger: Record log in %s' % filename)
@@ -256,6 +251,8 @@ class FileHandler(logging.Handler):
                 Config.add_callback(self._configure, 'kivy', 'log_name')
             except Exception:
                 # deactivate filehandler...
+                if FileHandler.fd not in (None, False):
+                    FileHandler.fd.close()
                 FileHandler.fd = False
                 Logger.exception('Error while activating FileHandler logger')
                 return

kivy/tests/test_logger.py

@@ -0,0 +1,60 @@
+"""
+Logger tests
+============
+"""
+
+import pytest
+import pathlib
+import time
+
+
+@pytest.fixture
+def file_handler():
+    # restores handler to original state
+    from kivy.config import Config
+
+    log_dir = Config.get("kivy", "log_dir")
+    log_maxfiles = Config.get("kivy", "log_maxfiles")
+
+    try:
+        yield None
+    finally:
+        Config.set("kivy", "log_dir", log_dir)
+        Config.set("kivy", "log_maxfiles", log_maxfiles)
+
+
+@pytest.mark.parametrize('n', [0, 1, 5])
+def test_purge_logs(tmp_path, file_handler, n):
+    from kivy.config import Config
+    from kivy.logger import FileHandler
+
+    Config.set("kivy", "log_dir", str(tmp_path))
+    Config.set("kivy", "log_maxfiles", n)
+
+    # create the default file first so it gets deleted so names match
+    handler = FileHandler()
+    handler._configure()
+    open_file = pathlib.Path(handler.filename).name
+    # wait a little so the timestamps are different for different files
+    time.sleep(.05)
+
+    names = [f'log_{i}.txt' for i in range(n + 2)]
+    for name in names:
+        p = tmp_path / name
+        p.write_text('some data')
+        time.sleep(.05)
+
+    handler.purge_logs()
+
+    # files that should have remained after purge
+    expected_names = list(reversed(names))[:n]
+    files = {f.name for f in tmp_path.iterdir()}
+    if open_file in files:
+        # one of the remaining files is the current open log, remove it
+        files.remove(open_file)
+        if len(expected_names) == len(files) + 1:
+            # the open log may or may not have been counted in the remaining
+            # files, remove one from expected to match removed open file
+            expected_names = expected_names[:-1]
+
+    assert set(expected_names) == files