Email history enhancement (#10114)

* add warning that the log is useless by default

* Add setting to enhance email log again

* add missing test for #10109

* add test for delete protections

* add error code

* Update email.md

* Update email.md

src/backend/InvenTree/InvenTree/tasks.py

@@ -10,7 +10,7 @@ from typing import Callable, Optional
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
-from django.core.exceptions import AppRegistryNotReady
+from django.core.exceptions import AppRegistryNotReady, ValidationError
 from django.core.management import call_command
 from django.db import DEFAULT_DB_ALIAS, connections
 from django.db.migrations.executor import MigrationExecutor
@@ -479,10 +479,16 @@ def delete_old_emails():
         emails = EmailMessage.objects.filter(timestamp__lte=threshold)
 
         if emails.count() > 0:
-            logger.info('Deleted %s old email messages', emails.count())
-            emails.delete()
+            try:
+                emails.delete()
+                logger.info('Deleted %s old email messages', emails.count())
+            except ValidationError:
+                logger.info(
+                    'Did not delete %s old email messages because of a validation error',
+                    emails.count(),
+                )
 
-    except AppRegistryNotReady:
+    except AppRegistryNotReady:  # pragma: no cover
         logger.info("Could not perform 'delete_old_emails' - App registry not ready")
 
 

src/backend/InvenTree/InvenTree/test_tasks.py

@@ -239,3 +239,61 @@ class InvenTreeTaskTests(PluginRegistryMixin, TestCase):
             msg.message,
             "Background worker task 'InvenTree.tasks.failed_task' failed after 10 attempts",
         )
+
+    def test_delete_old_emails(self):
+        """Test the delete_old_emails task."""
+        from common.models import EmailMessage
+
+        # Create an email message
+        self.create_mails()
+
+        # Run the task
+        InvenTreeSetting.set_setting('INVENTREE_DELETE_EMAIL_DAYS', 31)
+        InvenTree.tasks.offload_task(InvenTree.tasks.delete_old_emails, force_sync=True)
+
+        # Check that the email message has been deleted
+        emails = EmailMessage.objects.all()
+        self.assertEqual(len(emails), 1)
+        self.assertEqual(emails[0].subject, 'Test Email 2')
+
+        # Set the setting higher than the threshold
+        InvenTreeSetting.set_setting('INVENTREE_DELETE_EMAIL_DAYS', 30)
+
+        # Run the task again
+        InvenTree.tasks.offload_task(InvenTree.tasks.delete_old_emails, force_sync=True)
+        emails = EmailMessage.objects.all()
+        self.assertEqual(len(emails), 0)
+
+        # Re-Add messages and enable a proper log
+        self.create_mails()
+
+        # Set the setting lower than the threshold
+        InvenTreeSetting.set_setting('INVENTREE_DELETE_EMAIL_DAYS', 7)
+        InvenTreeSetting.set_setting('INVENTREE_PROTECT_EMAIL_LOG', True)
+
+        # Run the task again
+        InvenTree.tasks.offload_task(InvenTree.tasks.delete_old_emails, force_sync=True)
+
+        # Check that the email message has not been deleted
+        emails = EmailMessage.objects.all()
+        self.assertEqual(len(emails), 2)
+
+    def create_mails(self):
+        """Create some email messages for testing."""
+        from common.models import EmailMessage
+
+        start_mails = [
+            ['Test Email 1', 'This is a test email.', 'abc@example.org', threshold_low],
+            [
+                'Test Email 2',
+                'This is another test email.',
+                'def@example.org',
+                threshold,
+            ],
+        ]
+        for subject, body, to, timestamp in start_mails:
+            msg = EmailMessage.objects.create(
+                subject=subject, body=body, to=to, priority=1
+            )
+            msg.timestamp = timestamp
+            msg.save()

src/backend/InvenTree/common/models.py

@@ -54,7 +54,7 @@ import InvenTree.ready
 import InvenTree.tasks
 import users.models
 from common.setting.type import InvenTreeSettingsKeyType, SettingsKeyType
-from common.settings import global_setting_overrides
+from common.settings import get_global_setting, global_setting_overrides
 from generic.enums import StringEnum
 from generic.states import ColorEnum
 from generic.states.custom import state_color_mappings
@@ -2522,6 +2522,28 @@ class Priority(models.IntegerChoices):
 HEADER_PRIORITY = 'X-Priority'
 HEADER_MSG_ID = 'Message-ID'
 
+del_error_msg = _(
+    'INVE-E8: Email log deletion is protected. Set INVENTREE_PROTECT_EMAIL_LOG to False to allow deletion.'
+)
+
+
+class NoDeleteQuerySet(models.query.QuerySet):
+    """Custom QuerySet to prevent deletion of EmailLog entries."""
+
+    def delete(self):
+        """Override delete method to prevent deletion of EmailLog entries."""
+        if get_global_setting('INVENTREE_PROTECT_EMAIL_LOG'):
+            raise ValidationError(del_error_msg)
+        super().delete()
+
+
+class NoDeleteManager(models.Manager):
+    """Custom Manager to use NoDeleteQuerySet."""
+
+    def get_queryset(self):
+        """Return a NoDeleteQuerySet."""
+        return NoDeleteQuerySet(self.model, using=self._db)
+
 
 class EmailMessage(models.Model):
     """Model for storing email messages sent or received by the system.
@@ -2663,6 +2685,14 @@ class EmailMessage(models.Model):
 
         return ret
 
+    objects = NoDeleteManager()
+
+    def delete(self, *kwargs):
+        """Delete entry - if not protected."""
+        if get_global_setting('INVENTREE_PROTECT_EMAIL_LOG'):
+            raise ValidationError(del_error_msg)
+        return super().delete(*kwargs)
+
 
 class EmailThread(InvenTree.models.InvenTreeMetadataModel):
     """Model for storing email threads."""

src/backend/InvenTree/common/setting/system.py

@@ -340,6 +340,12 @@ SYSTEM_SETTINGS: dict[str, InvenTreeSettingsKeyType] = {
         'units': _('days'),
         'validator': [int, MinValueValidator(7)],
     },
+    'INVENTREE_PROTECT_EMAIL_LOG': {
+        'name': _('Protect Email Log'),
+        'description': _('Prevent deletion of email log entries'),
+        'default': False,
+        'validator': bool,
+    },
     'BARCODE_ENABLE': {
         'name': _('Barcode Support'),
         'description': _('Enable barcode scanner support in the web interface'),

src/backend/InvenTree/common/test_emails.py

@@ -10,6 +10,7 @@ from anymail.inbound import AnymailInboundMessage
 from anymail.signals import AnymailInboundEvent, AnymailTrackingEvent, inbound, tracking
 
 from common.models import EmailMessage, Priority
+from common.settings import set_global_setting
 from InvenTree.helpers_email import send_email
 from InvenTree.unit_test import InvenTreeAPITestCase
 
@@ -129,6 +130,31 @@ class EmailTests(InvenTreeAPITestCase):
         self.assertEqual(msg.status, EmailMessage.EmailStatus.FAILED)
         self.assertEqual(msg.error_message, 'Test error sending email')
 
+    def test_email_model_delete(self):
+        """Test that the email model does not allow deletion if disabled."""
+        set_global_setting('INVENTREE_PROTECT_EMAIL_LOG', True)
+        EmailMessage.objects.create(
+            subject='test sub', body='test msg', to='abc@example.org', priority=3
+        )
+
+        with self.assertRaises(ValidationError):
+            EmailMessage.objects.all().delete()
+
+        msg = EmailMessage.objects.create(
+            subject='test sub', body='test msg', to='abc@example.org', priority=3
+        )
+
+        with self.assertRaises(ValidationError):
+            msg.delete()
+
+        # Should still work without the protection
+        self.assertEqual(EmailMessage.objects.count(), 2)
+        set_global_setting('INVENTREE_PROTECT_EMAIL_LOG', False)
+        msg.delete()
+
+        # Check that the message was deleted
+        self.assertEqual(EmailMessage.objects.count(), 1)
+
 
 class EmailEventsTests(TestCase):
     """Unit tests for anymail events."""