1 files changed, 88 insertions, 3 deletions
diff --git a/src/mailman/chains/tests/test_headers.py b/src/mailman/chains/tests/test_headers.py
index a00f9c588..312f1eb54 100644
--- a/src/mailman/chains/tests/test_headers.py
+++ b/src/mailman/chains/tests/test_headers.py
@@ -25,12 +25,16 @@ __all__ = [
 import unittest
 
 from mailman.app.lifecycle import create_list
-from mailman.chains.headers import HeaderMatchRule
+from mailman.chains.headers import HeaderMatchRule, make_link
 from mailman.config import config
+from mailman.core.chains import process
 from mailman.email.message import Message
-from mailman.interfaces.chain import LinkAction
+from mailman.interfaces.chain import LinkAction, HoldEvent
+from mailman.interfaces.mailinglist import IHeaderMatchSet
+from mailman.testing.helpers import (
+    LogFileMark, configuration, event_subscribers,
+    specialized_message_from_string as mfs)
 from mailman.testing.layers import ConfigLayer
-from mailman.testing.helpers import LogFileMark, configuration
 
 
 
@@ -42,6 +46,24 @@ class TestHeaderChain(unittest.TestCase):
     def setUp(self):
         self._mlist = create_list('test@example.com')
 
+    def test_make_link(self):
+        # Test that make_link() with no given chain creates a Link with a
+        # deferred link action.
+        link = make_link('Subject', '[tT]esting')
+        self.assertEqual(link.rule.header, 'Subject')
+        self.assertEqual(link.rule.pattern, '[tT]esting')
+        self.assertEqual(link.action, LinkAction.defer)
+        self.assertIsNone(link.chain)
+
+    def test_make_link_with_chain(self):
+        # Test that make_link() with a given chain creates a Link with a jump
+        # action to the chain.
+        link = make_link('Subject', '[tT]esting', 'accept')
+        self.assertEqual(link.rule.header, 'Subject')
+        self.assertEqual(link.rule.pattern, '[tT]esting')
+        self.assertEqual(link.action, LinkAction.jump)
+        self.assertEqual(link.chain, config.chains['accept'])
+
     @configuration('antispam', header_checks="""
     Foo: a+
     Bar: bb?
@@ -119,3 +141,66 @@ class TestHeaderChain(unittest.TestCase):
                               HeaderMatchRule, 'x-spam-score', '.*')
         finally:
             config.rules = saved_rules
+
+    def test_list_rule(self):
+        # Test that the header-match chain has the header checks from the
+        # mailing-list configuration.
+        chain = config.chains['header-match']
+        header_matches = IHeaderMatchSet(self._mlist)
+        header_matches.add('Foo', 'a+')
+        links = [link for link in chain.get_links(self._mlist, Message(), {})
+                 if link.rule.name != 'any']
+        self.assertEqual(len(links), 1)
+        self.assertEqual(links[0].action, LinkAction.defer)
+        self.assertEqual(links[0].rule.header, 'foo')
+        self.assertEqual(links[0].rule.pattern, 'a+')
+
+    def test_list_complex_rule(self):
+        # Test that the mailing-list header-match complex rules are read
+        # properly.
+        chain = config.chains['header-match']
+        header_matches = IHeaderMatchSet(self._mlist)
+        header_matches.add('Foo', 'a+', 'reject')
+        header_matches.add('Bar', 'b+', 'discard')
+        header_matches.add('Baz', 'z+', 'accept')
+        links = [link for link in chain.get_links(self._mlist, Message(), {})
+                 if link.rule.name != 'any']
+        self.assertEqual(len(links), 3)
+        self.assertEqual([
+            (link.rule.header, link.rule.pattern, link.action, link.chain.name)
+            for link in links
+            ],
+            [('foo', 'a+', LinkAction.jump, 'reject'),
+             ('bar', 'b+', LinkAction.jump, 'discard'),
+             ('baz', 'z+', LinkAction.jump, 'accept'),
+            ])
+
+    @configuration('antispam', header_checks="""
+    Foo: foo
+    """, jump_chain='hold')
+    def test_priority_site_over_list(self):
+        # Test that the site-wide checks take precedence over the list-specific
+        # checks.
+        msg = mfs("""\
+From: anne@example.com
+To: test@example.com
+Subject: A message
+Message-ID: <ant>
+Foo: foo
+MIME-Version: 1.0
+
+A message body.
+""")
+        msgdata = {}
+        header_matches = IHeaderMatchSet(self._mlist)
+        header_matches.add('Foo', 'foo', 'accept')
+        # This event subscriber records the event that occurs when the message
+        # is processed by the owner chain.
+        events = []
+        with event_subscribers(events.append):
+            process(self._mlist, msg, msgdata, start_chain='header-match')
+        self.assertEqual(len(events), 1)
+        event = events[0]
+        # Site-wide wants to hold the message, the list wants to accept it.
+        self.assertTrue(isinstance(event, HoldEvent))
+        self.assertEqual(event.chain, config.chains['hold'])