diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bf718da..d2f3b430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ #### Improvements - feat: Added `LogEntry.remote_port` field. ([#671](https://github.com/jazzband/django-auditlog/pull/671)) +- feat: Added `truncate` option to `auditlogflush` management command. ([#681](https://github.com/jazzband/django-auditlog/pull/681)) - Drop Python 3.8 support. ([#678](https://github.com/jazzband/django-auditlog/pull/678)) - Confirm Django 5.1 support and drop Django 3.2 support. ([#677](https://github.com/jazzband/django-auditlog/pull/677)) diff --git a/auditlog/management/commands/auditlogflush.py b/auditlog/management/commands/auditlogflush.py index e57ef2c6..7a42aa67 100644 --- a/auditlog/management/commands/auditlogflush.py +++ b/auditlog/management/commands/auditlogflush.py @@ -1,6 +1,7 @@ import datetime from django.core.management.base import BaseCommand +from django.db import connection from auditlog.models import LogEntry @@ -25,11 +26,24 @@ def add_arguments(self, parser): dest="before_date", type=datetime.date.fromisoformat, ) + parser.add_argument( + "-t", + "--truncate", + action="store_true", + default=None, + help="Truncate log entry table.", + dest="truncate", + ) def handle(self, *args, **options): answer = options["yes"] + truncate = options["truncate"] before = options["before_date"] - + if truncate and before: + self.stdout.write( + "Truncate deletes all log entries and can not be passed with before-date." + ) + return if answer is None: warning_message = ( "This action will clear all log entries from the database." @@ -42,11 +56,39 @@ def handle(self, *args, **options): ) answer = response == "y" - if answer: + if not answer: + self.stdout.write("Aborted.") + return + + if not truncate: entries = LogEntry.objects.all() if before is not None: entries = entries.filter(timestamp__date__lt=before) count, _ = entries.delete() self.stdout.write("Deleted %d objects." % count) else: - self.stdout.write("Aborted.") + database_vendor = connection.vendor + database_display_name = connection.display_name + table_name = LogEntry._meta.db_table + if not TruncateQuery.support_truncate_statement(database_vendor): + self.stdout.write( + "Database %s does not support truncate statement." + % database_display_name + ) + return + with connection.cursor() as cursor: + query = TruncateQuery.to_sql(table_name) + cursor.execute(query) + self.stdout.write("Truncated log entry table.") + + +class TruncateQuery: + SUPPORTED_VENDORS = ("postgresql", "mysql", "sqlite", "oracle", "microsoft") + + @classmethod + def support_truncate_statement(cls, database_vendor) -> bool: + return database_vendor in cls.SUPPORTED_VENDORS + + @staticmethod + def to_sql(table_name) -> str: + return f"TRUNCATE TABLE {table_name};" diff --git a/auditlog_tests/test_commands.py b/auditlog_tests/test_commands.py index 349d5d91..a976759f 100644 --- a/auditlog_tests/test_commands.py +++ b/auditlog_tests/test_commands.py @@ -6,7 +6,7 @@ import freezegun from django.core.management import call_command -from django.test import TestCase +from django.test import TestCase, TransactionTestCase from auditlog_tests.models import SimpleModel @@ -110,3 +110,82 @@ def test_before_date(self): out, "Deleted 1 objects.", msg="Output shows deleted 1 object." ) self.assertEqual(err, "", msg="No stderr") + + +class AuditlogFlushWithTruncateTest(TransactionTestCase): + def setUp(self): + input_patcher = mock.patch("builtins.input") + self.mock_input = input_patcher.start() + self.addCleanup(input_patcher.stop) + + def make_object(self): + return SimpleModel.objects.create(text="I am a simple model.") + + def call_command(self, *args, **kwargs): + outbuf = StringIO() + errbuf = StringIO() + call_command("auditlogflush", *args, stdout=outbuf, stderr=errbuf, **kwargs) + return outbuf.getvalue().strip(), errbuf.getvalue().strip() + + def test_flush_with_both_truncate_and_before_date_options(self): + obj = self.make_object() + self.assertEqual(obj.history.count(), 1, msg="There is one log entry.") + out, err = self.call_command("--truncate", "--before-date=2000-01-01") + + self.assertEqual(obj.history.count(), 1, msg="There is still one log entry.") + self.assertEqual( + out, + "Truncate deletes all log entries and can not be passed with before-date.", + msg="Output shows error", + ) + self.assertEqual(err, "", msg="No stderr") + + def test_flush_with_truncate_and_yes(self): + obj = self.make_object() + self.assertEqual(obj.history.count(), 1, msg="There is one log entry.") + out, err = self.call_command("--truncate", "--y") + + self.assertEqual(obj.history.count(), 0, msg="There is no log entry.") + self.assertEqual( + out, + "Truncated log entry table.", + msg="Output shows table gets truncate", + ) + self.assertEqual(err, "", msg="No stderr") + + def test_flush_with_truncate_with_input_yes(self): + obj = self.make_object() + self.assertEqual(obj.history.count(), 1, msg="There is one log entry.") + self.mock_input.return_value = "Y\n" + out, err = self.call_command("--truncate") + + self.assertEqual(obj.history.count(), 0, msg="There is no log entry.") + self.assertEqual( + out, + "This action will clear all log entries from the database.\nTruncated log entry table.", + msg="Output shows warning and table gets truncate", + ) + self.assertEqual(err, "", msg="No stderr") + + @mock.patch( + "django.db.connection.vendor", + new_callable=mock.PropertyMock(return_value="unknown"), + ) + @mock.patch( + "django.db.connection.display_name", + new_callable=mock.PropertyMock(return_value="Unknown"), + ) + def test_flush_with_truncate_for_unsupported_database_vendor( + self, mocked_vendor, mocked_db_name + ): + obj = self.make_object() + self.assertEqual(obj.history.count(), 1, msg="There is one log entry.") + out, err = self.call_command("--truncate", "--y") + + self.assertEqual(obj.history.count(), 1, msg="There is still one log entry.") + self.assertEqual( + out, + "Database Unknown does not support truncate statement.", + msg="Output shows error", + ) + self.assertEqual(err, "", msg="No stderr")