# Copyright (C) 2012-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 pipeline runner."""

import unittest

from mailman.app.lifecycle import create_list
from mailman.config import config
from mailman.interfaces.handler import IHandler
from mailman.interfaces.pipeline import IPipeline
from mailman.runners.pipeline import PipelineRunner
from mailman.testing.helpers import (
    configuration,
    get_queue_messages,
    make_testable_runner,
    specialized_message_from_string as mfs,
)
from mailman.testing.layers import ConfigLayer
from zope.interface import implementer


@implementer(IHandler)
class MyTestHandler:
    """See `IHandler`."""

    name = 'test handler'
    description = 'A test handler'

    def __init__(self, marker, test):
        self._marker = marker
        self._test = test

    def process(self, mlist, msg, msgdata):
        self._test.mark(self._marker)


@implementer(IPipeline)
class MyTestPipeline:
    name = 'test'
    description = 'a test pipeline'

    def __init__(self, marker, test):
        self._marker = marker
        self._test = test

    def __iter__(self):
        yield MyTestHandler(self._marker, self._test)


class TestPipelineRunner(unittest.TestCase):
    """Test the pipeline runner."""

    layer = ConfigLayer

    def setUp(self):
        self._mlist = create_list('test@example.com')
        self._mlist.posting_pipeline = 'test posting'
        self._mlist.owner_pipeline = 'test owner'
        config.pipelines['test posting'] = MyTestPipeline('posting', self)
        config.pipelines['test owner'] = MyTestPipeline('owner', self)
        self._pipeline = make_testable_runner(PipelineRunner, 'pipeline')
        self._markers = []
        self._msg = mfs("""\
From: anne@example.com
To: test@example.com

""")

    def tearDown(self):
        del config.pipelines['test posting']
        del config.pipelines['test owner']

    def mark(self, marker):
        # Record a marker seen by a handler.
        self._markers.append(marker)

    def test_posting(self):
        # A message accepted for posting gets processed through the posting
        # pipeline.
        msgdata = dict(listid='test.example.com')
        config.switchboards['pipeline'].enqueue(self._msg, msgdata)
        self._pipeline.run()
        self.assertEqual(len(self._markers), 1)
        self.assertEqual(self._markers[0], 'posting')

    def test_owner(self):
        # A message accepted for posting to a list's owners gets processed
        # through the owner pipeline.
        msgdata = dict(listid='test.example.com',
                       to_owner=True)
        config.switchboards['pipeline'].enqueue(self._msg, msgdata)
        self._pipeline.run()
        self.assertEqual(len(self._markers), 1)
        self.assertEqual(self._markers[0], 'owner')

    def test_cleanse_owner(self):
        # The owner pipeline honors remove_dkim_headers.
        msgdata = dict(listid='test.example.com',
                       to_owner=True)
        self._msg['DKIM-Signature'] = 'just testing'
        self._mlist.owner_pipeline = 'default-owner-pipeline'
        config.switchboards['pipeline'].enqueue(self._msg, msgdata)
        with configuration('mta', remove_dkim_headers='yes'):
            self._pipeline.run()
        item = get_queue_messages('out', expected_count=1)[0]
        self.assertIsNone(item.msg.get('DKIM-Signature'))