logged-errors: remove old errors (#56476)

tests/test_publisher.py

@@ -1,3 +1,4 @@
+import datetime
 import io
 import json
 import os
@@ -169,6 +170,16 @@ def test_register_cronjobs():
     # noqa pylint: disable=not-an-iterable
     assert 'clean_sessions' in [x.function.__name__ for x in pub.cronjobs]
     # noqa pylint: disable=not-an-iterable
+    assert 'clean_nonces' in [x.function.__name__ for x in pub.cronjobs]
+    # noqa pylint: disable=not-an-iterable
+    assert 'clean_afterjobs' in [x.function.__name__ for x in pub.cronjobs]
+    # noqa pylint: disable=not-an-iterable
+    assert 'clean_tempfiles' in [x.function.__name__ for x in pub.cronjobs]
+    # noqa pylint: disable=not-an-iterable
+    assert 'clean_thumbnails' in [x.function.__name__ for x in pub.cronjobs]
+    # noqa pylint: disable=not-an-iterable
+    assert 'clean_loggederrors' in [x.function.__name__ for x in pub.cronjobs]
+    # noqa pylint: disable=not-an-iterable
     assert 'evaluate_jumps' in [x.name for x in pub.cronjobs]
 
 
@@ -349,6 +360,45 @@ def test_clean_thumbnails():
     assert os.listdir(dirname) == ['a']
 
 
+def test_clean_loggederrors_no_sql():
+    pub = create_temporary_pub()
+
+    # do nothing, no error
+    pub.clean_loggederrors()
+
+
+def test_clean_loggederrors():
+    pub = create_temporary_pub(sql_mode=True)
+
+    error1 = pub.loggederror_class()
+    error1.first_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(seconds=1)
+    error1.latest_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(seconds=1)
+    error1.store()
+
+    error2 = pub.loggederror_class()
+    error2.first_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(days=30, seconds=1)
+    error2.latest_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(seconds=1)
+    error2.store()
+
+    error3 = pub.loggederror_class()
+    error3.first_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(days=30, seconds=1)
+    error3.latest_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(days=30, seconds=1)
+    error3.store()
+
+    error4 = pub.loggederror_class()
+    error4.first_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(days=30, seconds=1)
+    error4.latest_occurence_timestamp = datetime.datetime.now() - datetime.timedelta(days=29, seconds=1)
+    error4.store()
+
+    pub.clean_loggederrors()
+
+    # error3 was deleted
+    assert pub.loggederror_class.count() == 3
+    assert pub.loggederror_class.get(error1.id)
+    assert pub.loggederror_class.get(error2.id)
+    assert pub.loggederror_class.get(error4.id)
+
+
 def test_get_site_language():
     pub = create_temporary_pub()
 

wcs/qommon/publisher.py

@@ -44,6 +44,8 @@ from django.utils import translation
 from django.utils.encoding import force_bytes, force_text
 from quixote.publish import Publisher, get_publisher, get_request, get_response
 
+from wcs.qommon.storage import Less
+
 from . import _, errors, force_str, logger, storage, template
 from .cron import CronJob
 from .http_request import HTTPRequest
@@ -666,6 +668,19 @@ class QommonPublisher(Publisher):
         one_month_ago = now - 30 * 86400
         self._clean_files(one_month_ago, os.path.join(self.app_dir, 'thumbs'))
 
+    def clean_loggederrors(self):
+        if not self.loggederror_class:
+            return
+
+        clauses = [
+            Less(
+                'latest_occurence_timestamp',
+                (datetime.datetime.now() - datetime.timedelta(days=30)).timetuple(),
+            )
+        ]
+        for error in self.loggederror_class.select(clause=clauses):
+            self.loggederror_class.remove_object(error.id)
+
     @classmethod
     def register_cronjobs(cls):
         cls.register_cronjob(CronJob(cls.clean_sessions, minutes=range(0, 60, 5), name='clean_sessions'))
@@ -673,6 +688,7 @@ class QommonPublisher(Publisher):
         cls.register_cronjob(CronJob(cls.clean_afterjobs, minutes=[0], name='clean_afterjobs'))
         cls.register_cronjob(CronJob(cls.clean_tempfiles, minutes=[0], name='clean_tempfiles'))
         cls.register_cronjob(CronJob(cls.clean_thumbnails, minutes=[0], name='clean_thumbnails'))
+        cls.register_cronjob(CronJob(cls.clean_loggederrors, hours=[3], name='clean_loggederrors'))
 
     _initialized = False