1 files changed, 123 insertions, 1 deletions
diff --git a/src/mailman/app/tests/test_subscriptions.py b/src/mailman/app/tests/test_subscriptions.py
index 8ba5f52ff..b19dc5398 100644
--- a/src/mailman/app/tests/test_subscriptions.py
+++ b/src/mailman/app/tests/test_subscriptions.py
@@ -18,7 +18,8 @@
 """Tests for the subscription service."""
 
 __all__ = [
-    'TestJoin'
+    'TestJoin',
+    'TestSubscriptionWorkflow',
     ]
 
 
@@ -26,11 +27,17 @@ import uuid
 import unittest
 
 from mailman.app.lifecycle import create_list
+from mailman.app.subscriptions import Workflow, SubscriptionWorkflow
 from mailman.interfaces.address import InvalidEmailAddressError
 from mailman.interfaces.member import MemberRole, MissingPreferredAddressError
+from mailman.interfaces.requests import IListRequests, RequestType
 from mailman.interfaces.subscriptions import (
     MissingUserError, ISubscriptionService)
 from mailman.testing.layers import ConfigLayer
+from mailman.interfaces.mailinglist import SubscriptionPolicy
+from mailman.interfaces.usermanager import IUserManager
+from mailman.utilities.datetime import now
+from mock import Mock
 from zope.component import getUtility
 
 
@@ -65,3 +72,118 @@ class TestJoin(unittest.TestCase):
                           self._service.join,
                           'test.example.com', anne.user.user_id,
                           role=MemberRole.owner)
+
+
+
+class TestWorkflow(unittest.TestCase):
+    layer = ConfigLayer
+
+    def setUp(self):
+        self.workflow = Workflow()
+        self.workflow._test_attribute = "test-value"
+        self.workflow._step_test = Mock()
+        self.workflow._next.append("test")
+
+    def test_iter_steps(self):
+        next(self.workflow)
+        self.assertTrue(self.workflow._step_test.called)
+        self.assertEqual(len(self.workflow._next), 0)
+        try:
+            next(self.workflow)
+        except StopIteration:
+            pass
+        else:
+            self.fail()
+
+    def test_save_restore(self):
+        self.workflow.save_state()
+        # Now create a new instance and restore
+        new_workflow = Workflow()
+        self.assertEqual(len(new_workflow._next), 0)
+        self.assertFalse(hasattr(new_workflow, "_test_attribute"))
+        new_workflow.restore_state()
+        self.assertEqual(len(new_workflow._next), 1)
+        self.assertEqual(new_workflow._next[0], "test")
+        self.assertEqual(self.workflow._test_attribute, "test-value")
+
+    def test_save_restore_no_next_step(self):
+        self.workflow._next.clear()
+        self.workflow.save_state()
+        # Now create a new instance and restore
+        new_workflow = Workflow()
+        new_workflow._next.append("test")
+        new_workflow.restore_state()
+        self.assertEqual(len(new_workflow._next), 0)
+
+
+
+class TestSubscriptionWorkflow(unittest.TestCase):
+    layer = ConfigLayer
+
+    def setUp(self):
+        self._mlist = create_list('test@example.com')
+        self._anne = 'anne@example.com'
+        self._user_manager = getUtility(IUserManager)
+
+    def test_preverified_address_joins_open_list(self):
+        # The mailing list has an open subscription policy, so the subscriber
+        # becomes a member with no human intervention.
+        self._mlist.subscription_policy = SubscriptionPolicy.open
+        anne = self._user_manager.create_address(self._anne, 'Anne Person')
+        self.assertIsNone(anne.verified_on)
+        self.assertIsNone(anne.user)
+        self.assertIsNone(self._mlist.subscribers.get_member(self._anne))
+        workflow = SubscriptionWorkflow(
+            self._mlist, anne,
+            pre_verified=True, pre_confirmed=False, pre_approved=False)
+        # Run the state machine to the end.  The result is that her address
+        # will be verified, linked to a user, and subscribed to the mailing
+        # list.
+        list(workflow)
+        self.assertIsNotNone(anne.verified_on)
+        self.assertIsNotNone(anne.user)
+        self.assertIsNotNone(self._mlist.subscribers.get_member(self._anne))
+
+    def test_verified_address_joins_moderated_list(self):
+        # The mailing list is moderated but the subscriber is not a verified
+        # address and the subscription request is not pre-verified.
+        # A confirmation email must be sent, it will serve as the verification
+        # email too.
+        anne = self._user_manager.create_address(self._anne, 'Anne Person')
+        request_db = IListRequests(self._mlist)
+        def _do_check():
+            anne.verified_on = now()
+            self.assertIsNone(self._mlist.subscribers.get_member(self._anne))
+            workflow = SubscriptionWorkflow(
+                self._mlist, anne,
+                pre_verified=False, pre_confirmed=True, pre_approved=False)
+            # Run the state machine to the end.
+            list(workflow)
+            # Look in the requests db
+            requests = list(request_db.of_type(RequestType.subscription))
+            self.assertEqual(len(requests), 1)
+            self.assertEqual(requests[0].key, anne.email)
+            request_db.delete_request(requests[0].id)
+        self._mlist.subscription_policy = SubscriptionPolicy.moderate
+        _do_check()
+        self._mlist.subscription_policy = \
+            SubscriptionPolicy.confirm_then_moderate
+        _do_check()
+
+    def test_confirmation_required(self):
+        # Tests subscriptions where user confirmation is required
+        self._mlist.subscription_policy = \
+            SubscriptionPolicy.confirm_then_moderate
+        anne = self._user_manager.create_address(self._anne, 'Anne Person')
+        self.assertIsNone(self._mlist.subscribers.get_member(self._anne))
+        workflow = SubscriptionWorkflow(
+            self._mlist, anne,
+            pre_verified=True, pre_confirmed=False, pre_approved=True)
+        # Run the state machine to the end.
+        list(workflow)
+        # A confirmation request must be pending
+        # TODO: test it
+        # Now restore and re-run the state machine as if we got the confirmation
+        workflow.restore_state()
+        list(workflow)
+        self.assertIsNotNone(self._mlist.subscribers.get_member(self._anne))