3 files changed, 113 insertions, 14 deletions

diff --git a/src/mailman_pgp/pgp/tests/base.py b/src/mailman_pgp/pgp/tests/base.py
index 8a5473d..5f132e7 100644
--- a/src/mailman_pgp/pgp/tests/base.py
+++ b/src/mailman_pgp/pgp/tests/base.py
@@ -19,18 +19,19 @@
 from email import message_from_file
 from unittest import TestCase
 
+from mailman.email.message import Message
 from pgpy import PGPKey
 from pkg_resources import resource_filename
 
 
 def load_message(path):
     with open(resource_filename('mailman_pgp.pgp.tests', path)) as f:
-        return message_from_file(f)
+        return message_from_file(f, Message)
 
 
 def load_key(path):
     key, _ = PGPKey.from_file(
-        resource_filename('mailman_pgp.pgp.tests', path))
+            resource_filename('mailman_pgp.pgp.tests', path))
     return key
 
 
diff --git a/src/mailman_pgp/rules/signature.py b/src/mailman_pgp/rules/signature.py
index 7ddc947..311ebfb 100644
--- a/src/mailman_pgp/rules/signature.py
+++ b/src/mailman_pgp/rules/signature.py
@@ -83,8 +83,7 @@ class Signature:
         user_manager = getUtility(IUserManager)
         sender = msg.sender
         address = user_manager.get_address(sender)
-        pgp_address = PGPAddress.query().filter_by(
-                email=address.email).first()
+        pgp_address = PGPAddress.for_address(address)
         if pgp_address is None:
             raise ValueError('PGP enabled address not found.')
 
diff --git a/src/mailman_pgp/rules/tests/test_signature.py b/src/mailman_pgp/rules/tests/test_signature.py
index 084f599..dbd832e 100644
--- a/src/mailman_pgp/rules/tests/test_signature.py
+++ b/src/mailman_pgp/rules/tests/test_signature.py
@@ -18,9 +18,17 @@ from unittest import TestCase
 
 from mailman.app.lifecycle import create_list
 from mailman.config import config
-from mailman.testing.helpers import specialized_message_from_string as mfs
+from mailman.interfaces.action import Action
+from mailman.interfaces.member import MemberRole
+from mailman.interfaces.usermanager import IUserManager
+from mailman.testing.helpers import (specialized_message_from_string as mfs,
+                                     set_preferred)
+from zope.component import getUtility
 
-from mailman_pgp.database import mm_transaction
+from mailman_pgp.database import mm_transaction, transaction
+from mailman_pgp.model.address import PGPAddress
+from mailman_pgp.model.list import PGPMailingList
+from mailman_pgp.pgp.tests.base import load_message, load_key
 from mailman_pgp.rules.signature import Signature
 from mailman_pgp.testing.layers import PGPConfigLayer
 
@@ -29,26 +37,117 @@ class TestSignature(TestCase):
     layer = PGPConfigLayer
 
     def setUp(self):
+        self.rule = Signature()
+
+        user_manager = getUtility(IUserManager)
         with mm_transaction():
             self.mlist = create_list('nobody@example.com',
                                      style_name='pgp-default')
+            self.sender = user_manager.create_user('RSA-1024b@example.org',
+                                                   display_name='RSA 1024b example')
+            set_preferred(self.sender)
+            self.mlist.subscribe(self.sender, MemberRole.member)
+
+        self.pgp_list = PGPMailingList.for_list(self.mlist)
+
+        sender_key = load_key('data/rsa_1024.pub.asc')
+        with transaction() as t:
+            self.pgp_sender = PGPAddress(self.sender.preferred_address)
+            self.pgp_sender.key = sender_key
+            t.add(self.pgp_sender)
+
+        self.msg_clear = load_message('data/clear.eml')
+        self.msg_inline_signed = load_message('data/inline_signed.eml')
+        self.msg_mime_signed = load_message('data/mime_signed.eml')
+        self.msg_inline_signed_invalid = load_message(
+                'data/inline_cleartext_signed_invalid.eml')
+        self.msg_mime_signed_invalid = load_message(
+                'data/mime_signed_invalid.eml')
 
     def test_has_rule(self):
         self.assertIn(Signature.name, config.rules.keys())
 
     def test_no_pgp_list(self):
         with mm_transaction():
-            ordinary_list = create_list('ordinary@example.com')
+            ordinary_list = create_list('test@example.com')
         msg = mfs("""\
 From: anne@example.com
-To: ordinary@example.com
-Subject: A Message with non-ascii body
-Message-ID: <ant>
-MIME-Version: 1.0
+To: test@example.com
 
-A message body.
 """)
 
-        rule = config.rules[Signature.name]
         with self.assertRaises(ValueError):
-            rule.check(ordinary_list, msg, {})
+            self.rule.check(ordinary_list, msg, {})
+
+    def test_no_address(self):
+        with transaction():
+            self.pgp_list.unsigned_msg_action = Action.defer
+        msg = mfs("""\
+From: anne@example.com
+To: nobody@example.com
+
+""")
+        with self.assertRaises(ValueError):
+            self.rule.check(self.mlist, msg, {})
+
+    def test_no_key(self):
+        pass
+
+    def assertAction(self, msgdata, action, reasons):
+        self.assertEqual(msgdata['moderation_action'], action.name)
+        self.assertListEqual(msgdata['moderation_reasons'], reasons)
+
+    def test_unsigned_action(self):
+        with transaction():
+            self.pgp_list.unsigned_msg_action = Action.hold
+            self.pgp_list.inline_pgp_action = Action.defer
+            self.pgp_list.expired_sig_action = Action.defer
+            self.pgp_list.invalid_sig_action = Action.defer
+            self.pgp_list.revoked_sig_action = Action.defer
+
+        msgdata = {}
+        matches = self.rule.check(self.mlist, self.msg_clear, msgdata)
+        self.assertTrue(matches)
+        self.assertAction(msgdata, Action.hold, ['The message is unsigned.'])
+
+        matches = self.rule.check(self.mlist, self.msg_inline_signed, msgdata)
+        self.assertFalse(matches)
+
+        matches = self.rule.check(self.mlist, self.msg_mime_signed, msgdata)
+        self.assertFalse(matches)
+
+    def test_inline_pgp_action(self):
+        with transaction():
+            self.pgp_list.unsigned_msg_action = Action.defer
+            self.pgp_list.inline_pgp_action = Action.hold
+            self.pgp_list.expired_sig_action = Action.defer
+            self.pgp_list.invalid_sig_action = Action.defer
+            self.pgp_list.revoked_sig_action = Action.defer
+
+        msgdata = {}
+        matches = self.rule.check(self.mlist, self.msg_inline_signed, msgdata)
+        self.assertTrue(matches)
+        self.assertAction(msgdata, Action.hold, ['Inline PGP is not allowed.'])
+
+        matches = self.rule.check(self.mlist, self.msg_mime_signed, msgdata)
+        self.assertFalse(matches)
+
+    def test_invalid_sig_action(self):
+        with transaction():
+            self.pgp_list.unsigned_msg_action = Action.defer
+            self.pgp_list.inline_pgp_action = Action.defer
+            self.pgp_list.expired_sig_action = Action.defer
+            self.pgp_list.invalid_sig_action = Action.hold
+            self.pgp_list.revoked_sig_action = Action.defer
+
+        msgdata = {}
+        matches = self.rule.check(self.mlist, self.msg_inline_signed_invalid,
+                                  msgdata)
+        self.assertTrue(matches)
+        self.assertAction(msgdata, Action.hold, ['Signature did not verify.'])
+
+        msgdata = {}
+        matches = self.rule.check(self.mlist, self.msg_mime_signed_invalid,
+                                  msgdata)
+        self.assertTrue(matches)
+        self.assertAction(msgdata, Action.hold, ['Signature did not verify.'])