diff options
Diffstat (limited to '')
| -rw-r--r-- | src/mailman/workflows/tests/test_unsubscriptions.py (renamed from src/mailman/app/tests/test_unsubscriptions.py) | 152 |
1 files changed, 75 insertions, 77 deletions
diff --git a/src/mailman/app/tests/test_unsubscriptions.py b/src/mailman/workflows/tests/test_unsubscriptions.py index 4ca657190..2e210e90b 100644 --- a/src/mailman/app/tests/test_unsubscriptions.py +++ b/src/mailman/workflows/tests/test_unsubscriptions.py @@ -21,14 +21,15 @@ import unittest from contextlib import suppress from mailman.app.lifecycle import create_list -from mailman.app.subscriptions import UnSubscriptionWorkflow -from mailman.interfaces.mailinglist import SubscriptionPolicy from mailman.interfaces.pending import IPendings from mailman.interfaces.subscriptions import TokenOwner from mailman.interfaces.usermanager import IUserManager from mailman.testing.helpers import LogFileMark, get_queue_messages from mailman.testing.layers import ConfigLayer from mailman.utilities.datetime import now +from mailman.workflows.unsubscription import ( + ConfirmModerationUnsubscriptionPolicy, ConfirmUnsubscriptionPolicy, + ModerationUnsubscriptionPolicy, OpenUnsubscriptionPolicy) from unittest.mock import patch from zope.component import getUtility @@ -40,7 +41,7 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def setUp(self): self._mlist = create_list('test@example.com') self._mlist.admin_immed_notify = False - self._mlist.unsubscription_policy = SubscriptionPolicy.open + self._mlist.unsubscription_policy = OpenUnsubscriptionPolicy self._mlist.send_welcome_message = False self._anne = 'anne@example.com' self._user_manager = getUtility(IUserManager) @@ -58,7 +59,7 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def test_start_state(self): # Test the workflow starts with no tokens or members. - workflow = UnSubscriptionWorkflow(self._mlist) + workflow = self._mlist.unsubscription_policy(self._mlist) self.assertEqual(workflow.token_owner, TokenOwner.no_one) self.assertIsNone(workflow.token) self.assertIsNone(workflow.member) @@ -67,8 +68,8 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # Test there is a Pendable object associated with a held # unsubscription request and it has some valid data associated with # it. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) with suppress(StopIteration): workflow.run_thru('send_confirmation') self.assertIsNotNone(workflow.token) @@ -84,23 +85,23 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def test_user_or_address_required(self): # The `subscriber` attribute must be a user or address that is provided # to the workflow. - workflow = UnSubscriptionWorkflow(self._mlist) + workflow = OpenUnsubscriptionPolicy(self._mlist) self.assertRaises(AssertionError, list, workflow) def test_user_is_subscribed_to_unsubscribe(self): # A user must be subscribed to a list when trying to unsubscribe. addr = self._user_manager.create_address('aperson@example.org') addr.verfied_on = now() - workflow = UnSubscriptionWorkflow(self._mlist, addr) + workflow = self._mlist.unsubscription_policy(self._mlist, addr) self.assertRaises(AssertionError, workflow.run_thru, 'subscription_checks') def test_confirmation_checks_open_list(self): # An unsubscription from an open list does not need to be confirmed or # moderated. - self._mlist.unsubscription_policy = SubscriptionPolicy.open - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) - workflow.run_thru('confirmation_checks') + self._mlist.unsubscription_policy = OpenUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) + workflow.run_thru('subscription_checks') with patch.object(workflow, '_step_do_unsubscription') as step: next(workflow) step.assert_called_once_with() @@ -108,10 +109,9 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def test_confirmation_checks_no_user_confirmation_needed(self): # An unsubscription from a list which does not need user confirmation # skips to the moderation checks. - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow(self._mlist, self.anne, - pre_confirmed=True) - workflow.run_thru('confirmation_checks') + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) + workflow.run_thru('subscription_checks') with patch.object(workflow, '_step_moderation_checks') as step: next(workflow) step.assert_called_once_with() @@ -120,8 +120,8 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # The unsubscription policy requires user-confirmation, but their # unsubscription is pre-confirmed. Since moderation is not reuqired, # the user will be immediately unsubscribed. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow( + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy( self._mlist, self.anne, pre_confirmed=True) workflow.run_thru('confirmation_checks') with patch.object(workflow, '_step_do_unsubscription') as step: @@ -133,19 +133,19 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # unsubscription is pre-confirmed. Since moderation is required, that # check will be performed. self._mlist.unsubscription_policy = ( - SubscriptionPolicy.confirm_then_moderate) - workflow = UnSubscriptionWorkflow( + ConfirmModerationUnsubscriptionPolicy) + workflow = self._mlist.unsubscription_policy( self._mlist, self.anne, pre_confirmed=True) workflow.run_thru('confirmation_checks') - with patch.object(workflow, '_step_do_unsubscription') as step: + with patch.object(workflow, '_step_moderation_checks') as step: next(workflow) step.assert_called_once_with() def test_send_confirmation_checks_confirm_list(self): # The unsubscription policy requires user confirmation and the # unsubscription is not pre-confirmed. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) workflow.run_thru('confirmation_checks') with patch.object(workflow, '_step_send_confirmation') as step: next(workflow) @@ -153,17 +153,17 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def test_moderation_checks_moderated_list(self): # The unsubscription policy requires moderation. - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) - workflow.run_thru('confirmation_checks') + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) + workflow.run_thru('subscription_checks') with patch.object(workflow, '_step_moderation_checks') as step: next(workflow) step.assert_called_once_with() def test_moderation_checks_approval_required(self): # The moderator must approve the subscription request. - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) workflow.run_thru('moderation_checks') with patch.object(workflow, '_step_get_moderator_approval') as step: next(workflow) @@ -172,8 +172,8 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def test_do_unsusbcription(self): # An open unsubscription policy means the user gets unsubscribed to # the mailing list without any further confirmations or approvals. - self._mlist.unsubscription_policy = SubscriptionPolicy.open - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = OpenUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) list(workflow) member = self._mlist.regular_members.get_member(self._anne) self.assertIsNone(member) @@ -182,9 +182,9 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # A moderation-requiring subscription policy plus a pre-approved # address means the user gets unsubscribed from the mailing list # without any further confirmation or approvals. - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow(self._mlist, self.anne, - pre_approved=True) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne, + pre_approved=True) list(workflow) # Anne is now unsubscribed form the mailing list. member = self._mlist.regular_members.get_member(self._anne) @@ -198,10 +198,10 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # address means the user gets unsubscribed to the mailing list without # any further confirmations or approvals. self._mlist.unsubscription_policy = ( - SubscriptionPolicy.confirm_then_moderate) - workflow = UnSubscriptionWorkflow(self._mlist, self.anne, - pre_approved=True, - pre_confirmed=True) + ConfirmModerationUnsubscriptionPolicy) + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne, + pre_approved=True, + pre_confirmed=True) list(workflow) member = self._mlist.regular_members.get_member(self._anne) self.assertIsNone(member) @@ -212,10 +212,8 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): def test_do_unsubscription_cleanups(self): # Once the user is unsubscribed, the token and its associated pending # database record will be removed from the database. - self._mlist.unsubscription_policy = SubscriptionPolicy.open - workflow = UnSubscriptionWorkflow(self._mlist, self.anne, - pre_approved=True, - pre_confirmed=True) + self._mlist.unsubscription_policy = OpenUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) # Run the workflow. list(workflow) # Anne is now unsubscribed from the list. @@ -229,9 +227,9 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # The workflow runs until moderator approval is required, at which # point the workflow is saved. Once the moderator approves, the # workflow resumes and the user is unsubscribed. - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow( - self._mlist, self.anne, pre_confirmed=True) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy( + self._mlist, self.anne) # Run the entire workflow. list(workflow) # The user is currently subscribed to the mailing list. @@ -244,7 +242,7 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # Create a new workflow with the previous workflow's save token, and # restore its state. This models an approved un-sunscription request # and should result in the user getting subscribed. - approved_workflow = UnSubscriptionWorkflow(self._mlist) + approved_workflow = self._mlist.unsubscription_policy(self._mlist) approved_workflow.token = workflow.token approved_workflow.restore() list(approved_workflow) @@ -260,9 +258,9 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # When the unsubscription is held for moderator approval, a message is # logged. mark = LogFileMark('mailman.subscribe') - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow( - self._mlist, self.anne, pre_confirmed=True) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy( + self._mlist, self.anne) # Run the entire workflow. list(workflow) self.assertIn( @@ -277,9 +275,9 @@ class TestUnSubscriptionWorkflow(unittest.TestCase): # When the unsubscription is held for moderator approval, and the list # is so configured, a notification is sent to the list moderators. self._mlist.admin_immed_notify = True - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow( - self._mlist, self.anne, pre_confirmed=True) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy( + self._mlist, self.anne) # Consume the entire state machine. list(workflow) items = get_queue_messages('virgin', expected_count=1) @@ -305,9 +303,9 @@ request approval: # the list is so configured, a notification is sent to the list # moderators. self._mlist.admin_immed_notify = False - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow( - self._mlist, self.anne, pre_confirmed=True) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy( + self._mlist, self.anne) # Consume the entire state machine. list(workflow) get_queue_messages('virgin', expected_count=0) @@ -318,9 +316,9 @@ request approval: def test_send_confirmation(self): # A confirmation message gets sent when the unsubscription must be # confirmed. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy # Run the workflow to model the confirmation step. - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) list(workflow) items = get_queue_messages('virgin', expected_count=1) message = items[0].msg @@ -336,8 +334,8 @@ request approval: def test_do_confirmation_unsubscribes_user(self): # Unsubscriptions to the mailing list must be confirmed. Once that's # done, the user's address is unsubscribed. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) list(workflow) # Anne is a member. member = self._mlist.regular_members.get_member(self._anne) @@ -347,7 +345,7 @@ request approval: self.assertIsNotNone(workflow.token) self.assertEqual(workflow.token_owner, TokenOwner.subscriber) # Confirm. - confirm_workflow = UnSubscriptionWorkflow(self._mlist) + confirm_workflow = self._mlist.unsubscription_policy(self._mlist) confirm_workflow.token = workflow.token confirm_workflow.restore() list(confirm_workflow) @@ -363,8 +361,8 @@ request approval: # done, the address is unsubscribed. address = self.anne.register('anne.person@example.com') self._mlist.subscribe(address) - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow(self._mlist, address) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, address) list(workflow) # Bart is a member. member = self._mlist.regular_members.get_member( @@ -375,7 +373,7 @@ request approval: self.assertIsNotNone(workflow.token) self.assertEqual(workflow.token_owner, TokenOwner.subscriber) # Confirm. - confirm_workflow = UnSubscriptionWorkflow(self._mlist) + confirm_workflow = self._mlist.unsubscription_policy(self._mlist) confirm_workflow.token = workflow.token confirm_workflow.restore() list(confirm_workflow) @@ -390,8 +388,8 @@ request approval: def test_do_confirmation_nonmember(self): # Attempt to confirm the unsubscription of a member who has already # been unsubscribed. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) list(workflow) # Anne is a member. member = self._mlist.regular_members.get_member(self._anne) @@ -403,7 +401,7 @@ request approval: # Unsubscribe Anne out of band. member.unsubscribe() # Confirm. - confirm_workflow = UnSubscriptionWorkflow(self._mlist) + confirm_workflow = self._mlist.unsubscription_policy(self._mlist) confirm_workflow.token = workflow.token confirm_workflow.restore() list(confirm_workflow) @@ -414,8 +412,8 @@ request approval: def test_do_confirmation_nonmember_final_step(self): # Attempt to confirm the unsubscription of a member who has already # been unsubscribed. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) list(workflow) # Anne is a member. member = self._mlist.regular_members.get_member(self._anne) @@ -425,7 +423,7 @@ request approval: self.assertIsNotNone(workflow.token) self.assertEqual(workflow.token_owner, TokenOwner.subscriber) # Confirm. - confirm_workflow = UnSubscriptionWorkflow(self._mlist) + confirm_workflow = self._mlist.unsubscription_policy(self._mlist) confirm_workflow.token = workflow.token confirm_workflow.restore() confirm_workflow.run_until('do_unsubscription') @@ -443,8 +441,8 @@ request approval: # the user confirming their subscription, and then the moderator # approving it, that different tokens are used in these two cases. self._mlist.unsubscription_policy = ( - SubscriptionPolicy.confirm_then_moderate) - workflow = UnSubscriptionWorkflow(self._mlist, self.anne) + ConfirmModerationUnsubscriptionPolicy) + workflow = self._mlist.unsubscription_policy(self._mlist, self.anne) # Run the state machine up to the first confirmation, and cache the # confirmation token. list(workflow) @@ -457,7 +455,7 @@ request approval: self.assertIsNotNone(workflow.token) self.assertEqual(workflow.token_owner, TokenOwner.subscriber) # The old token will not work for moderator approval. - moderator_workflow = UnSubscriptionWorkflow(self._mlist) + moderator_workflow = self._mlist.unsubscription_policy(self._mlist) moderator_workflow.token = token moderator_workflow.restore() list(moderator_workflow) @@ -468,7 +466,7 @@ request approval: # that there's a new token for the next steps. self.assertNotEqual(token, moderator_workflow.token) # The old token won't work. - final_workflow = UnSubscriptionWorkflow(self._mlist) + final_workflow = self._mlist.unsubscription_policy(self._mlist) final_workflow.token = token self.assertRaises(LookupError, final_workflow.restore) # Running this workflow will fail. @@ -492,9 +490,9 @@ request approval: def test_confirmation_needed_and_pre_confirmed(self): # The subscription policy is 'confirm' but the subscription is # pre-confirmed so the moderation checks can be skipped. - self._mlist.unsubscription_policy = SubscriptionPolicy.confirm - workflow = UnSubscriptionWorkflow( - self._mlist, self.anne, pre_confirmed=True, pre_approved=True) + self._mlist.unsubscription_policy = ConfirmUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy( + self._mlist, self.anne, pre_confirmed=True) list(workflow) # Anne was unsubscribed. self.assertIsNone(workflow.token) @@ -504,11 +502,11 @@ request approval: def test_confirmation_needed_moderator_address(self): address = self.anne.register('anne.person@example.com') self._mlist.subscribe(address) - self._mlist.unsubscription_policy = SubscriptionPolicy.moderate - workflow = UnSubscriptionWorkflow(self._mlist, address) + self._mlist.unsubscription_policy = ModerationUnsubscriptionPolicy + workflow = self._mlist.unsubscription_policy(self._mlist, address) # Get moderator approval. list(workflow) - approved_workflow = UnSubscriptionWorkflow(self._mlist) + approved_workflow = self._mlist.unsubscription_policy(self._mlist) approved_workflow.token = workflow.token approved_workflow.restore() list(approved_workflow) |