aboutsummaryrefslogtreecommitdiff
path: root/src/mailman_pgp/workflows
diff options
context:
space:
mode:
Diffstat (limited to 'src/mailman_pgp/workflows')
-rw-r--r--src/mailman_pgp/workflows/base.py3
-rw-r--r--src/mailman_pgp/workflows/tests/test_base.py130
-rw-r--r--src/mailman_pgp/workflows/tests/test_subscription.py2
3 files changed, 74 insertions, 61 deletions
diff --git a/src/mailman_pgp/workflows/base.py b/src/mailman_pgp/workflows/base.py
index d05781d..014f2dd 100644
--- a/src/mailman_pgp/workflows/base.py
+++ b/src/mailman_pgp/workflows/base.py
@@ -55,9 +55,8 @@ class PGPMixin:
class SetPubkeyMixin:
- def __init__(self, pubkey=None, pre_confirmed=False):
+ def __init__(self, pubkey=None):
self.pubkey = pubkey
- self.pubkey_confirmed = pre_confirmed
@property
def pubkey_key(self):
diff --git a/src/mailman_pgp/workflows/tests/test_base.py b/src/mailman_pgp/workflows/tests/test_base.py
index af00d01..c59cfe0 100644
--- a/src/mailman_pgp/workflows/tests/test_base.py
+++ b/src/mailman_pgp/workflows/tests/test_base.py
@@ -24,8 +24,12 @@ from unittest.mock import patch
from mailman.app.lifecycle import create_list
from mailman.interfaces.pending import IPendings
from mailman.interfaces.usermanager import IUserManager
+from mailman.interfaces.workflows import IWorkflow
from mailman.testing.helpers import get_queue_messages
+from mailman.workflows.common import SubscriptionBase
+from public import public
from zope.component import getUtility
+from zope.interface import implementer
from mailman_pgp.database import mm_transaction, transaction
from mailman_pgp.model.address import PGPAddress
@@ -33,11 +37,11 @@ from mailman_pgp.model.list import PGPMailingList
from mailman_pgp.pgp.tests.base import load_key
from mailman_pgp.pgp.wrapper import PGPWrapper
from mailman_pgp.testing.layers import PGPConfigLayer
-from mailman_pgp.workflows.base import KEY_REQUEST
-from mailman_pgp.workflows.subscription import ConfirmSubscriptionPolicy
+from mailman_pgp.workflows.base import (KEY_REQUEST, PGPMixin, SetPubkeyMixin,
+ ConfirmPubkeyMixin)
-class PubkeyMixinSetup():
+class PubkeyMixinTestSetup():
def setUp(self):
with mm_transaction():
self.mlist = create_list('test@example.com',
@@ -54,34 +58,62 @@ class PubkeyMixinSetup():
self.sender = self.um.create_address('rsa-1024b@example.org')
-class TestPGPMixin(PubkeyMixinSetup, unittest.TestCase):
+@implementer(IWorkflow)
+class PGPTestWorkflow(SubscriptionBase, PGPMixin, SetPubkeyMixin,
+ ConfirmPubkeyMixin):
+ name = 'test-workflow'
+ description = ''
+ initial_state = 'prepare'
+ save_attributes = (
+ 'pubkey_key',
+ 'pubkey_confirmed',
+ 'address_key',
+ 'subscriber_key',
+ 'user_key',
+ 'token_owner_key'
+ )
+
+ def __init__(self, mlist, subscriber=None, *, pubkey=None,
+ pubkey_pre_confirmed=False):
+ SubscriptionBase.__init__(self, mlist, subscriber)
+ SetPubkeyMixin.__init__(self, pubkey=pubkey)
+ ConfirmPubkeyMixin.__init__(self, pre_confirmed=pubkey_pre_confirmed)
+ PGPMixin.__init__(self)
+
+ def _step_prepare(self):
+ self.push('do_subscription')
+ self.push('pubkey_confirmation')
+ self.push('pubkey_checks')
+ self.push('pgp_prepare')
+ self.push('sanity_checks')
+
+
+@public
+class TestPGPMixin(PubkeyMixinTestSetup, unittest.TestCase):
layer = PGPConfigLayer
def test_create_address(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
workflow.run_thru('pgp_prepare')
pgp_address = PGPAddress.for_address(self.sender)
self.assertIsNotNone(pgp_address)
def test_address_existing(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
with transaction() as t:
pgp_address = PGPAddress(self.sender)
t.add(pgp_address)
workflow.run_thru('pgp_prepare')
+ still = PGPAddress.for_address(self.sender)
+ self.assertIsNotNone(still)
-class TestSetPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
+@public
+class TestSetPubkeyMixin(PubkeyMixinTestSetup, unittest.TestCase):
layer = PGPConfigLayer
def test_key_request_sent(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
list(workflow)
items = get_queue_messages('virgin', expected_count=1)
message = items[0].msg
@@ -91,24 +123,20 @@ class TestSetPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
self.assertEqual(message.get_payload(), KEY_REQUEST)
def test_receive_key(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
list(workflow)
with transaction():
pgp_address = PGPAddress.for_address(self.sender)
pgp_address.key = self.sender_key.pubkey
- receive_workflow = ConfirmSubscriptionPolicy(self.mlist)
+ receive_workflow = PGPTestWorkflow(self.mlist)
receive_workflow.token = workflow.token
receive_workflow.restore()
receive_workflow.run_thru('receive_key')
def test_set_pubkey(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True,
- pubkey=self.sender_key.pubkey)
+ workflow = PGPTestWorkflow(self.mlist, self.sender,
+ pubkey=self.sender_key.pubkey)
workflow.run_thru('pubkey_checks')
pgp_address = PGPAddress.for_address(self.sender)
self.assertIsNotNone(pgp_address)
@@ -117,9 +145,7 @@ class TestSetPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
self.sender_key.fingerprint)
def test_pubkey_set(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
with transaction() as t:
pgp_address = PGPAddress(self.sender)
pgp_address.key = self.sender_key.pubkey
@@ -129,26 +155,23 @@ class TestSetPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
self.sender_key.fingerprint)
-class TestConfirmPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
+@public
+class TestConfirmPubkeyMixin(PubkeyMixinTestSetup, unittest.TestCase):
layer = PGPConfigLayer
def test_key_request_pubkey_set(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True,
- pubkey=self.sender_key.pubkey,
- pubkey_pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender,
+ pubkey=self.sender_key.pubkey,
+ pubkey_pre_confirmed=True)
workflow.run_thru('pubkey_confirmation')
with patch.object(workflow, '_step_do_subscription') as step:
next(workflow)
step.assert_called_once_with()
def test_send_key_confirm_request(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True,
- pubkey=self.sender_key.pubkey,
- pubkey_pre_confirmed=False)
+ workflow = PGPTestWorkflow(self.mlist, self.sender,
+ pubkey=self.sender_key.pubkey,
+ pubkey_pre_confirmed=False)
list(workflow)
items = get_queue_messages('virgin', expected_count=1)
message = items[0].msg
@@ -159,14 +182,12 @@ class TestConfirmPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
self.assertTrue(wrapped.is_encrypted())
def test_receive_confirmation(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True,
- pubkey=self.sender_key.pubkey,
- pubkey_pre_confirmed=False)
+ workflow = PGPTestWorkflow(self.mlist, self.sender,
+ pubkey=self.sender_key.pubkey,
+ pubkey_pre_confirmed=False)
list(workflow)
- receive_workflow = ConfirmSubscriptionPolicy(self.mlist)
+ receive_workflow = PGPTestWorkflow(self.mlist)
receive_workflow.token = workflow.token
receive_workflow.restore()
receive_workflow.run_thru('receive_key_confirmation')
@@ -175,13 +196,12 @@ class TestConfirmPubkeyMixin(PubkeyMixinSetup, unittest.TestCase):
step.assert_called_once_with()
-class TestBothPubkeyMixins(PubkeyMixinSetup, unittest.TestCase):
+@public
+class TestBothPubkeyMixins(PubkeyMixinTestSetup, unittest.TestCase):
layer = PGPConfigLayer
def test_pended_data_key_request(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
with suppress(StopIteration):
workflow.run_thru('send_key_request')
self.assertIsNotNone(workflow.token)
@@ -193,10 +213,8 @@ class TestBothPubkeyMixins(PubkeyMixinSetup, unittest.TestCase):
self.assertEqual(pendable['token_owner'], 'subscriber')
def test_pended_data_key_confirmation(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True,
- pubkey=self.sender_key.pubkey)
+ workflow = PGPTestWorkflow(self.mlist, self.sender,
+ pubkey=self.sender_key.pubkey)
with suppress(StopIteration):
workflow.run_thru('send_key_confirm_request')
self.assertIsNotNone(workflow.token)
@@ -208,9 +226,7 @@ class TestBothPubkeyMixins(PubkeyMixinSetup, unittest.TestCase):
self.assertEqual(pendable['token_owner'], 'subscriber')
def test_exisitng_pgp_address(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
with transaction() as t:
pgp_address = PGPAddress(self.sender)
@@ -224,9 +240,7 @@ class TestBothPubkeyMixins(PubkeyMixinSetup, unittest.TestCase):
step.assert_called_once_with()
def test_exisitng_pgp_address_not_confirmed(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
with transaction() as t:
pgp_address = PGPAddress(self.sender)
@@ -239,9 +253,7 @@ class TestBothPubkeyMixins(PubkeyMixinSetup, unittest.TestCase):
step.assert_called_once_with()
def test_exisitng_pgp_address_no_key(self):
- workflow = ConfirmSubscriptionPolicy(self.mlist, self.sender,
- pre_verified=True,
- pre_confirmed=True)
+ workflow = PGPTestWorkflow(self.mlist, self.sender)
with transaction() as t:
pgp_address = PGPAddress(self.sender)
diff --git a/src/mailman_pgp/workflows/tests/test_subscription.py b/src/mailman_pgp/workflows/tests/test_subscription.py
index 9464a83..f9fa1e1 100644
--- a/src/mailman_pgp/workflows/tests/test_subscription.py
+++ b/src/mailman_pgp/workflows/tests/test_subscription.py
@@ -20,6 +20,7 @@ import unittest
from mailman.app.lifecycle import create_list
from mailman.interfaces.usermanager import IUserManager
+from public import public
from zope.component import getUtility
from mailman_pgp.database import mm_transaction
@@ -29,6 +30,7 @@ from mailman_pgp.workflows.subscription import (
ModerationSubscriptionPolicy, OpenSubscriptionPolicy)
+@public
class TestSubscriptionWorkflows(unittest.TestCase):
layer = PGPConfigLayer
generated by cgit v1.2.3-70-g09d2 (git 2.51.2) at 2025-11-08 13:59:14 +0000