Update docs with information about how to use thread-unsafe code in
[salmon.git] / tests / test_confirm.py
blobc32da82800ad8800e6e27485c72e42706bcfedf5
1 from unittest.mock import Mock, patch
3 import jinja2
5 from salmon import mail, view
6 from salmon.confirm import ConfirmationEngine, ConfirmationStorage
7 from salmon.queue import Queue
8 from salmon.testing import delivered, relay
10 from .setup_env import SalmonTestCase
13 class ConfirmationTestCase(SalmonTestCase):
14 def setUp(self):
15 super().setUp()
16 self.storage = ConfirmationStorage()
17 self.engine = ConfirmationEngine('run/confirm', self.storage)
18 view.LOADER = jinja2.Environment(loader=jinja2.FileSystemLoader('tests/data/templates'))
20 def tearDown(self):
21 super().tearDown()
22 view.LOADER = None
24 def test_ConfirmationStorage(self):
25 self.storage.store('testing', 'somedude@localhost',
26 '12345', '567890')
27 secret, pending_id = self.storage.get('testing', 'somedude@localhost')
28 self.assertEqual(secret, '12345')
29 self.assertEqual(pending_id, '567890')
31 self.storage.delete('testing', 'somedude@localhost')
32 self.assertEqual(len(self.storage.confirmations), 0)
34 self.storage.store('testing', 'somedude@localhost',
35 '12345', '567890')
36 self.assertEqual(len(self.storage.confirmations), 1)
37 self.storage.clear()
38 self.assertEqual(len(self.storage.confirmations), 0)
40 @patch("smtplib.SMTP")
41 def test_ConfirmationEngine_send(self, smtp_mock):
42 smtp_mock.return_value = Mock()
44 Queue('run/queue').clear()
45 self.engine.clear()
47 list_name = 'testing'
48 action = 'subscribing to'
49 host = 'localhost'
51 message = mail.MailRequest('fakepeer', 'somedude@localhost',
52 'testing-subscribe@localhost', 'Fake body.')
54 self.engine.send(relay(port=0), 'testing', message, 'confirmation.msg', locals())
56 self.assertEqual(smtp_mock.return_value.sendmail.call_count, 1)
57 self.assertEqual(smtp_mock.return_value.quit.call_count, 1)
58 assert delivered('somedude', to_queue=self.engine.pending)
60 return smtp_mock.return_value.sendmail.call_args[0][2]
62 def test_ConfirmationEngine_verify(self):
63 confirm = self.test_ConfirmationEngine_send()
64 confirm = mail.MailRequest(None, None, None, confirm)
66 resp = mail.MailRequest('fakepeer', '"Somedude Smith" <somedude@localhost>', confirm['Reply-To'], 'Fake body')
68 target, _, expect_secret = confirm['Reply-To'].split('-')
69 expect_secret = expect_secret.split('@')[0]
71 found = self.engine.verify(target, resp['from'], 'invalid_secret')
72 assert not found
74 pending = self.engine.verify(target, resp['from'], expect_secret)
75 assert pending, "Verify failed: %r not in %r." % (expect_secret,
76 self.storage.confirmations)
78 self.assertEqual(pending['from'], 'somedude@localhost')
79 self.assertEqual(pending['to'], 'testing-subscribe@localhost')
81 def test_ConfirmationEngine_cancel(self):
82 confirm = self.test_ConfirmationEngine_send()
83 confirm = mail.MailRequest(None, None, None, confirm)
85 target, _, expect_secret = confirm['Reply-To'].split('-')
86 expect_secret = expect_secret.split('@')[0]
88 self.engine.cancel(target, confirm['To'], expect_secret)
90 self.assertNotIn(b"testing:somedude@localhost", self.engine.storage.confirmations.keys())
92 def test_ConfirmationEngine_cancel_bad_secret(self):
93 confirm = self.test_ConfirmationEngine_send()
94 confirm = mail.MailRequest(None, None, None, confirm)
96 target = confirm['Reply-To'].split('-')[0]
97 expect_secret = "bad"
99 self.engine.cancel(target, confirm['To'], expect_secret)
101 self.assertIn(b"testing:somedude@localhost", self.engine.storage.confirmations.keys())