# Copyright (C) 2015-2025 by the Free Software Foundation, Inc. # # This file is part of GNU Mailman. # # GNU Mailman is free software: you can redistribute it and/or modify it under # the terms of the GNU General Public License as published by the Free # Software Foundation, either version 3 of the License, or (at your option) # any later version. # # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or # FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for # more details. # # You should have received a copy of the GNU General Public License along with # GNU Mailman. If not, see . """Digest helper tests.""" import os import unittest from datetime import timedelta from mailman.app.digests import ( bump_digest_number_and_volume, maybe_send_digest_now, ) from mailman.app.lifecycle import create_list from mailman.config import config from mailman.interfaces.digests import DigestFrequency from mailman.interfaces.member import DeliveryMode from mailman.runners.digest import DigestRunner from mailman.testing.helpers import ( get_queue_messages, make_testable_runner, specialized_message_from_string as mfs, subscribe, ) from mailman.testing.layers import ConfigLayer from mailman.utilities.datetime import factory, now as right_now class TestBumpDigest(unittest.TestCase): layer = ConfigLayer def setUp(self): self._mlist = create_list('ant@example.com') self._mlist.volume = 7 self._mlist.next_digest_number = 4 self.right_now = right_now() def test_bump_no_previous_digest(self): self._mlist.digest_last_sent_at = None bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 7) self.assertEqual(self._mlist.next_digest_number, 5) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_yearly(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( days=-370) self._mlist.digest_volume_frequency = DigestFrequency.yearly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 8) self.assertEqual(self._mlist.next_digest_number, 1) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_yearly_not_yet(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( days=-200) self._mlist.digest_volume_frequency = DigestFrequency.yearly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 7) self.assertEqual(self._mlist.next_digest_number, 5) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_monthly(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( days=-32) self._mlist.digest_volume_frequency = DigestFrequency.monthly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 8) self.assertEqual(self._mlist.next_digest_number, 1) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_monthly_not_yet(self): # The normal test date starts on the first day of the month, so let's # fast forward it a few days so we can set the digest last sent time # to earlier in the same month. self._mlist.digest_last_sent_at = self.right_now factory.fast_forward(days=26) self._mlist.digest_volume_frequency = DigestFrequency.monthly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 7) self.assertEqual(self._mlist.next_digest_number, 5) self.assertEqual(self._mlist.digest_last_sent_at, right_now()) def test_bump_quarterly(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( days=-93) self._mlist.digest_volume_frequency = DigestFrequency.quarterly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 8) self.assertEqual(self._mlist.next_digest_number, 1) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_quarterly_not_yet(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( days=-88) self._mlist.digest_volume_frequency = DigestFrequency.quarterly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 7) self.assertEqual(self._mlist.next_digest_number, 5) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_weekly(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( days=-8) self._mlist.digest_volume_frequency = DigestFrequency.weekly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 8) self.assertEqual(self._mlist.next_digest_number, 1) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_weekly_not_yet(self): # The normal test date starts on the first day of the week, so let's # fast forward it a few days so we can set the digest last sent time # to earlier in the same week. self._mlist.digest_last_sent_at = self.right_now factory.fast_forward(days=3) self._mlist.digest_volume_frequency = DigestFrequency.weekly bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 7) self.assertEqual(self._mlist.next_digest_number, 5) self.assertEqual(self._mlist.digest_last_sent_at, right_now()) def test_bump_daily(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( hours=-27) self._mlist.digest_volume_frequency = DigestFrequency.daily bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 8) self.assertEqual(self._mlist.next_digest_number, 1) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_daily_not_yet(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( hours=-5) self._mlist.digest_volume_frequency = DigestFrequency.daily bump_digest_number_and_volume(self._mlist) self.assertEqual(self._mlist.volume, 7) self.assertEqual(self._mlist.next_digest_number, 5) self.assertEqual(self._mlist.digest_last_sent_at, self.right_now) def test_bump_bad_frequency(self): self._mlist.digest_last_sent_at = self.right_now + timedelta( hours=-22) self._mlist.digest_volume_frequency = -10 self.assertRaises(AssertionError, bump_digest_number_and_volume, self._mlist) class TestMaybeSendDigest(unittest.TestCase): layer = ConfigLayer def setUp(self): self._mlist = create_list('ant@example.com') self._mlist.send_welcome_message = False self._mailbox_path = os.path.join(self._mlist.data_path, 'digest.mmdf') # The mailing list needs at least one digest recipient. member = subscribe(self._mlist, 'Anne') member.preferences.delivery_mode = DeliveryMode.plaintext_digests self._subject_number = 1 self._runner = make_testable_runner(DigestRunner, 'digest') def _to_digest(self, count=1): for i in range(count): msg = mfs("""\ To: ant@example.com From: anne@example.com Subject: message {} """.format(self._subject_number)) self._subject_number += 1 config.handlers['to-digest'].process(self._mlist, msg, {}) def test_send_digest_over_threshold(self): # Put a few messages in the digest. self._to_digest(3) # Set the size threshold low enough to trigger a send. self._mlist.digest_size_threshold = 0.1 maybe_send_digest_now(self._mlist) self._runner.run() # There are no digests in flight now, and a single digest message has # been sent. get_queue_messages('digest', expected_count=0) self.assertFalse(os.path.exists(self._mailbox_path)) items = get_queue_messages('virgin', expected_count=1) digest_contents = str(items[0].msg) self.assertIn('Subject: message 1', digest_contents) self.assertIn('Subject: message 2', digest_contents) def test_dont_send_digest_under_threshold(self): # Put a few messages in the digest. self._to_digest(3) # Set the size threshold high enough to not trigger a send. self._mlist.digest_size_threshold = 100 maybe_send_digest_now(self._mlist) self._runner.run() # A digest is still being collected, but none have been sent. get_queue_messages('digest', expected_count=0) self.assertGreater(os.path.getsize(self._mailbox_path), 0) self.assertLess(os.path.getsize(self._mailbox_path), 100 * 1024.0) get_queue_messages('virgin', expected_count=0) def test_dont_send_digest_threshold_zero(self): # Put a few messages in the digest. self._to_digest(3) # Set the size threshold to zero which should not trigger a send. self._mlist.digest_size_threshold = 0 maybe_send_digest_now(self._mlist) self._runner.run() # A digest is still being collected, but none have been sent. get_queue_messages('digest', expected_count=0) self.assertGreater(os.path.getsize(self._mailbox_path), 0) get_queue_messages('virgin', expected_count=0) def test_force_send_digest_under_threshold(self): # Put a few messages in the digest. self._to_digest(3) # Set the size threshold high enough to not trigger a send. self._mlist.digest_size_threshold = 100 # Force sending a digest anyway. maybe_send_digest_now(self._mlist, force=True) self._runner.run() # There are no digests in flight now, and a single digest message has # been sent. get_queue_messages('digest', expected_count=0) self.assertFalse(os.path.exists(self._mailbox_path)) items = get_queue_messages('virgin', expected_count=1) digest_contents = str(items[0].msg) self.assertIn('Subject: message 1', digest_contents) self.assertIn('Subject: message 2', digest_contents) def test_force_and_threshold_zero(self): # Put a few messages in the digest. self._to_digest(3) # Set the size threshold to zero (unlimited). self._mlist.digest_size_threshold = 0 # Force sending a digest anyway. maybe_send_digest_now(self._mlist, force=True) self._runner.run() # There are no digests in flight now, and a single digest message has # been sent. get_queue_messages('digest', expected_count=0) self.assertFalse(os.path.exists(self._mailbox_path)) items = get_queue_messages('virgin', expected_count=1) digest_contents = str(items[0].msg) self.assertIn('Subject: message 1', digest_contents) self.assertIn('Subject: message 2', digest_contents)