audit: add clean job to remove old audit entries (#51974)

tests/test_audit.py

@@ -0,0 +1,48 @@
+import pytest
+from django.utils.timezone import now
+
+from wcs.audit import Audit
+from wcs.formdef import FormDef
+from wcs.qommon import audit
+from wcs.qommon.http_request import HTTPRequest
+
+from .utilities import clean_temporary_pub, create_temporary_pub
+
+
+@pytest.fixture
+def pub(request):
+    pub = create_temporary_pub()
+    req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
+    pub.set_app_dir(req)
+    pub.load_site_options()
+    return pub
+
+
+def teardown_module(module):
+    clean_temporary_pub()
+
+
+def test_audit_clean_job(pub, freezer):
+    Audit.wipe()
+    FormDef.wipe()
+    pub.user_class.wipe()
+
+    user = pub.user_class()
+    user.name = 'Test'
+    user.store()
+
+    formdef = FormDef()
+    formdef.name = 'form title'
+    formdef.store()
+
+    current_timestamp = now()
+    freezer.move_to('2018-12-01T00:00:00')
+    audit('listing', obj=formdef, user=user.id)
+    audit('listing', obj=formdef, user=user.id)
+
+    freezer.move_to(current_timestamp)
+    audit('listing', obj=formdef, user=user.id)
+
+    assert Audit.count() == 3
+    Audit.clean()
+    assert Audit.count() == 1

wcs/audit.py

@@ -14,6 +14,8 @@
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, see <http://www.gnu.org/licenses/>.
 
+import datetime
+
 from django.utils.timezone import now
 from quixote import get_publisher, get_request
 
@@ -89,3 +91,8 @@ class Audit(sql.Audit):
         if self.extra_data and self.extra_data.get('extra_label'):
             parts.append(self.extra_data.get('extra_label'))
         return ' - '.join(parts)
+
+    @classmethod
+    def clean(cls, **kwargs):
+        audit_retention_days = get_publisher().get_site_option('audit-retention-days') or 730
+        Audit.wipe(clause=[sql.Less('timestamp', now() - datetime.timedelta(days=int(audit_retention_days)))])

wcs/publisher.py

@@ -153,6 +153,12 @@ class WcsPublisher(QommonPublisher):
         cls.register_cronjob(
             CronJob(cls.clean_deleted_users, name='clean_deleted_users', hours=[3], minutes=[0])
         )
+
+        # once a day clean old audit entries
+        from .audit import Audit
+
+        cls.register_cronjob(CronJob(Audit.clean, name='clean_audit', hours=[3], minutes=[0]))
+
         # other jobs
         data_sources.register_cronjob()
         formdef.register_cronjobs()