# Copyright (C) 2014-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 <https://www.gnu.org/licenses/>.

"""Test the digest runner."""

import os
import re
import unittest

from email import message_from_binary_file, message_from_bytes
from email.iterators import _structure as structure
from email.mime.text import MIMEText
from importlib.resources import open_binary
from io import StringIO
from mailman.app.lifecycle import create_list
from mailman.config import config
from mailman.email.message import Message
from mailman.interfaces.member import DeliveryMode
from mailman.interfaces.template import ITemplateManager
from mailman.runners.digest import DigestRunner
from mailman.testing.helpers import (
    digest_mbox,
    get_queue_messages,
    LogFileMark,
    make_digest_messages,
    make_testable_runner,
    message_from_string,
    specialized_message_from_string as mfs,
    subscribe,
)
from mailman.testing.layers import ConfigLayer
from string import Template
from tempfile import TemporaryDirectory
from zope.component import getUtility


class TestDigest(unittest.TestCase):
    """Test the digest runner."""

    layer = ConfigLayer
    maxDiff = None

    def setUp(self):
        self._mlist = create_list('test@example.com')
        self._mlist.send_welcome_message = False
        self._mlist.digest_size_threshold = 1
        self._digestq = config.switchboards['digest']
        self._shuntq = config.switchboards['shunt']
        self._virginq = config.switchboards['virgin']
        self._runner = make_testable_runner(DigestRunner, 'digest')
        self._process = config.handlers['to-digest'].process

    def _check_virgin_queue(self):
        # There should be two messages in the virgin queue: the digest as
        # plain-text and as multipart.
        items = get_queue_messages('virgin', expected_count=2)
        self.assertEqual(
            sorted(item.msg.get_content_type() for item in items),
            ['multipart/mixed', 'text/plain'])
        for item in items:
            self.assertEqual(item.msg['subject'],
                             'Test Digest, Vol 1, Issue 1')
        return items

    def _get_plain_body(self, items):
        # Need to run after _check_virgin_queue which gets the items.
        for item in items:
            if item.msg.get_content_type() == 'text/plain':
                return item.msg.get_payload(decode=True)

    def test_simple_message(self):
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        make_digest_messages(self._mlist)
        self._check_virgin_queue()
        # The digest mbox and all intermediary mboxes must have been removed
        # (GL #259).
        self.assertEqual(os.listdir(self._mlist.data_path), [])

    def test_non_ascii_message(self):
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        msg = Message()
        msg['From'] = 'anne@example.org'
        msg['To'] = 'test@example.com'
        msg['Content-Type'] = 'multipart/mixed'
        msg.attach(MIMEText('message with non-ascii chars: \xc3\xa9',
                            'plain', 'utf-8'))
        mbox = digest_mbox(self._mlist)
        mbox.add(msg.as_string())
        # Use any error logs as the error message if the test fails.
        error_log = LogFileMark('mailman.error')
        make_digest_messages(self._mlist, msg)
        # The runner will send the file to the shunt queue on exception.
        self.assertEqual(len(self._shuntq.files), 0, error_log.read())
        self._check_virgin_queue()

    def test_utf7_message_with_inline_ascii_sig(self):
        # The test message is bigger than 1K.  Set the threshold bigger to
        # avoid double processing in make_digest_messages.
        self._mlist.digest_size_threshold = 5
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        with open_binary('mailman.runners.tests.data',
                         'ascii_in_utf7.eml') as fp:
            msg = message_from_binary_file(fp, Message)
        # Use any error logs as the error message if the test fails.
        error_log = LogFileMark('mailman.error')
        make_digest_messages(self._mlist, msg)
        # The runner will send the file to the shunt queue on exception.
        self.assertEqual(len(self._shuntq.files), 0, error_log.read())
        self._check_virgin_queue()

    def test_non_ascii_in_ascii_part(self):
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        msg = message_from_bytes(b"""\
From: anne@example.org
To: test@example.com
Subject: Non-ascii in ascii message
MIME-Version: 1.0
Content-Type: multipart/mixed; boundary="abcxyz"

--abcxyz
Content-Type: text/plain; charset=us-ascii
Content-Transfer-Encoding: 8bit

Don\xe2\x80\x99t try this at home.
--abcxyz--
""", Message)
        # Use any error logs as the error message if the test fails.
        error_log = LogFileMark('mailman.error')
        make_digest_messages(self._mlist, msg)
        # The runner will send the file to the shunt queue on exception.
        self.assertEqual(len(self._shuntq.files), 0, error_log.read())
        items = self._check_virgin_queue()
        self.assertEqual(self._get_plain_body(items), b"""\
Send Test mailing list submissions to
\ttest@example.com

To subscribe or unsubscribe via email, send a message with subject or
body 'help' to
\ttest-request@example.com

You can reach the person managing the list at
\ttest-owner@example.com

When replying, please edit your Subject line so it is more specific
than "Re: Contents of Test digest..."

Today\'s Topics:

   1. Non-ascii in ascii message (anne@example.org)


----------------------------------------------------------------------

Message: 1
From: anne@example.org
Subject: Non-ascii in ascii message
To: test@example.com
Content-Type: multipart/mixed; boundary="abcxyz"

Don\xef\xbf\xbd\xef\xbf\xbd\xef\xbf\xbdt try this at home.

------------------------------

Subject: Digest Footer

_______________________________________________
Test mailing list -- test@example.com
To unsubscribe send an email to test-leave@example.com


------------------------------

End of Test Digest, Vol 1, Issue 1
**********************************
""")

    def test_mime_digest_format(self):
        # Make sure that the format of the MIME digest is as expected.
        self._mlist.digest_size_threshold = 0.6
        self._mlist.volume = 1
        self._mlist.next_digest_number = 1
        self._mlist.send_welcome_message = False
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        # Fill the digest.
        process = config.handlers['to-digest'].process
        size = 0
        for i in range(1, 5):
            text = Template("""\
From: aperson@example.com
To: xtest@example.com
Subject: Test message $i
List-Post: <test@example.com>

Here is message $i
""").substitute(i=i)
            msg = message_from_string(text)
            process(self._mlist, msg, {})
            size += len(text)
            if size >= self._mlist.digest_size_threshold * 1024:
                break
        # Run the digest runner to create the MIME and RFC 1153 digests.
        runner = make_testable_runner(DigestRunner)
        runner.run()
        items = get_queue_messages('virgin', expected_count=2)
        # Find the MIME one.
        mime_digest = None
        for item in items:
            if item.msg.is_multipart():
                assert mime_digest is None, 'We got two MIME digests'
                mime_digest = item.msg
        fp = StringIO()
        # Verify the structure is what we expect.
        structure(mime_digest, fp)
        self.assertMultiLineEqual(fp.getvalue(), """\
multipart/mixed
    text/plain
    text/plain
    multipart/digest
        message/rfc822
            text/plain
        message/rfc822
            text/plain
        message/rfc822
            text/plain
        message/rfc822
            text/plain
    text/plain
""")
        # Verify the Message: headers
        for i in range(1, 5):
            body = mime_digest.get_payload(2).get_payload(i-1).get_payload(0)
            self.assertEqual(body['message'], str(i))

    def test_plain_digest_format(self):
        # Make sure that the format of the MIME digest is as expected.
        self._mlist.digest_size_threshold = 0.6
        self._mlist.volume = 1
        self._mlist.next_digest_number = 1
        self._mlist.send_welcome_message = False
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        # Fill the digest.
        process = config.handlers['to-digest'].process
        size = 0
        for i in range(1, 5):
            text = Template("""\
From: aperson@example.com
To: xtest@example.com
Subject: Test message $i
List-Post: <test@example.com>

Here is message $i
""").substitute(i=i)
            msg = message_from_string(text)
            process(self._mlist, msg, {})
            size += len(text)
            if size >= self._mlist.digest_size_threshold * 1024:
                break
        # Run the digest runner to create the MIME and RFC 1153 digests.
        runner = make_testable_runner(DigestRunner)
        runner.run()
        items = get_queue_messages('virgin', expected_count=2)
        # Find the plain one.
        plain_digest = None
        for item in items:
            if not item.msg.is_multipart():
                assert plain_digest is None, 'We got two plain digests'
                plain_digest = item.msg
        fp = StringIO()
        # Verify the structure is what we expect.
        structure(plain_digest, fp)
        self.assertMultiLineEqual(fp.getvalue(), """\
text/plain
""")
        # Verify the Message: headers
        body = plain_digest.get_payload(decode=True).decode('us-ascii')
        for i in range(1, 5):
            mo = re.search('^Message: {}$'.format(i), body, re.MULTILINE)
            self.assertIsNotNone(mo)

    def test_issue141(self):
        # Currently DigestMode.summary_digests are equivalent to mime_digests.
        # This also tests GL issue 234.
        self._mlist.send_welcome_message = False
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.summary_digests
        make_digest_messages(self._mlist)
        # There should be one message in the outgoing queue, destined for
        # Bart, formatted as a MIME digest.
        items = get_queue_messages('virgin', expected_count=1)
        # Bart is the only recipient.
        self.assertEqual(items[0].msgdata['recipients'],
                         set(['bperson@example.com']))
        # The message is a MIME digest, with the structure we expect.
        fp = StringIO()
        structure(items[0].msg, fp)
        self.assertMultiLineEqual(fp.getvalue(), """\
multipart/mixed
    text/plain
    text/plain
    multipart/digest
        message/rfc822
            text/plain
    text/plain
""")

    def test_issue141_one_last_digest(self):
        # Currently DigestMode.summary_digests are equivalent to mime_digests.
        # Also tests issue 234.
        self._mlist.send_welcome_message = False
        bart = subscribe(self._mlist, 'Bart')
        self._mlist.send_one_last_digest_to(
            bart.address, DeliveryMode.summary_digests)
        make_digest_messages(self._mlist)
        # There should be one message in the outgoing queue, destined for
        # Bart, formatted as a MIME digest.
        items = get_queue_messages('virgin', expected_count=1)
        # Bart is the only recipient.
        self.assertEqual(items[0].msgdata['recipients'],
                         set(['bperson@example.com']))
        # The message is a MIME digest, with the structure we expect.
        fp = StringIO()
        structure(items[0].msg, fp)
        self.assertMultiLineEqual(fp.getvalue(), """\
multipart/mixed
    text/plain
    text/plain
    multipart/digest
        message/rfc822
            text/plain
    text/plain
""")


class TestI18nDigest(unittest.TestCase):
    layer = ConfigLayer
    maxDiff = None

    def setUp(self):
        config.push('french', """
        [mailman]
        default_language: fr
        """)
        self.addCleanup(config.pop, 'french')
        self._mlist = create_list('test@example.com')
        self._mlist.send_welcome_message = False
        self._mlist.preferred_language = 'fr'
        # Can't use 0 because of !1230.
        self._mlist.digest_size_threshold = 0.01
        self._process = config.handlers['to-digest'].process
        self._runner = make_testable_runner(DigestRunner)
        # Add a French version of the digest masthead.
        tempdir = TemporaryDirectory()
        self.addCleanup(tempdir.cleanup)
        french_path = os.path.join(tempdir.name, 'fr', 'masthead.txt')
        os.makedirs(os.path.dirname(french_path))
        with open(french_path, 'w', encoding='utf-8') as fp:
            print("""\
Envoyez vos messages pour la liste $display_name à
\t$got_list_email

Pour vous (dés)abonner par courriel, envoyez un message avec « help » dans
le corps ou dans le sujet à
\t$got_request_email

Vous pouvez contacter l'administrateur de la liste à l'adresse
\t$got_owner_email

Si vous répondez, n'oubliez pas de changer l'objet du message afin
qu'il soit plus spécifique que « Re: Contenu du groupe de $display_name...
""", file=fp)
        getUtility(ITemplateManager).set(
            'list:member:digest:masthead', self._mlist.list_id,
            'file:///{}/$language/masthead.txt'.format(tempdir.name))

    def test_multilingual_digest(self):
        # When messages come in with a content-type character set different
        # than that of the list's preferred language, recipients will get an
        # internationalized digest.
        #
        # Subscribe some users receiving digests.
        anne = subscribe(self._mlist, 'Anne')
        anne.preferences.delivery_mode = DeliveryMode.mime_digests
        bart = subscribe(self._mlist, 'Bart')
        bart.preferences.delivery_mode = DeliveryMode.plaintext_digests
        msg = mfs("""\
From: aperson@example.org
To: test@example.com
Subject: =?iso-2022-jp?b?GyRCMGxIVhsoQg==?=
MIME-Version: 1.0
Content-Type: text/plain; charset=iso-2022-jp
Content-Transfer-Encoding: 7bit

\x1b$B0lHV\x1b(B
""")
        self._process(self._mlist, msg, {})
        self._runner.run()
        # There are two digests in the virgin queue; one is the MIME digest
        # and the other is the RFC 1153 digest.
        items = get_queue_messages('virgin', expected_count=2)
        if items[0].msg.is_multipart():
            mime, rfc1153 = items[0].msg, items[1].msg
        else:
            rfc1153, mime = items[0].msg, items[1].msg
        # The MIME version contains a mix of French and Japanese.  The digest
        # chrome added by Mailman is in French.
        self.assertEqual(mime['subject'].encode(),
                         '=?iso-8859-1?q?Groupe_Test=2C_Vol_1=2C_Parution_1?=')
        self.assertEqual(str(mime['subject']),
                         'Groupe Test, Vol 1, Parution 1')
        # The first subpart contains the iso-8859-1 masthead.
        masthead = mime.get_payload(0).get_payload(decode=True).decode(
            'iso-8859-1')
        self.assertMultiLineEqual(masthead.splitlines()[0],
                                  'Envoyez vos messages pour la liste Test à')
        # The second subpart contains the utf-8 table of contents.
        self.assertEqual(mime.get_payload(1)['content-description'],
                         "Today's Topics (1 messages)")
        toc = mime.get_payload(1).get_payload(decode=True).decode('utf-8')
        self.assertMultiLineEqual(toc.splitlines()[0], 'Thèmes du jour :')
        # The third subpart is a multipart/digest part and its first subpart
        # contains the posted message in Japanese.
        self.assertEqual(mime.get_payload(2).get_content_type(),
                         'multipart/digest')
        self.assertEqual(mime.get_payload(2).get_payload(0).get_content_type(),
                         'message/rfc822')
        post = mime.get_payload(2).get_payload(0).get_payload(0)
        self.assertEqual(post['subject'], '=?iso-2022-jp?b?GyRCMGxIVhsoQg==?=')
        # Compare the bytes so that this module doesn't contain string
        # literals in multiple incompatible character sets.
        self.assertEqual(post.get_payload(decode=True), b'\x1b$B0lHV\x1b(B\n')
        # The RFC 1153 digest will have the same subject, but its payload will
        # be recast into utf-8.
        self.assertEqual(str(rfc1153['subject']),
                         'Groupe Test, Vol 1, Parution 1')
        self.assertEqual(rfc1153.get_charset(), 'utf-8')
        lines = rfc1153.get_payload(decode=True).decode('utf-8').splitlines()
        self.assertEqual(lines[0], 'Envoyez vos messages pour la liste Test à')