1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
# Copyright (C) 2015-2017 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# GNU Mailman. If not, see <http://www.gnu.org/licenses/>.
"""Test the IUserManager implementation."""
import unittest
from mailman.app.lifecycle import create_list
from mailman.config import config
from mailman.interfaces.address import ExistingAddressError
from mailman.interfaces.autorespond import IAutoResponseSet, Response
from mailman.interfaces.member import DeliveryMode
from mailman.interfaces.usermanager import (
AddressCreatedEvent, AddressCreatingEvent, AddressDeletedEvent,
AddressDeletingEvent, IUserManager)
from mailman.testing.helpers import event_subscribers
from mailman.testing.layers import ConfigLayer
from mailman.utilities.datetime import now
from zope.component import getUtility
class TestUserManager(unittest.TestCase):
layer = ConfigLayer
def setUp(self):
self._usermanager = getUtility(IUserManager)
self._events = []
def _record_event(self, event):
self._events.append(event)
def test_create_address_event(self):
with event_subscribers(self._record_event):
address = self._usermanager.create_address('anne@example.com')
self.assertEqual(len(self._events), 2)
self.assertIsInstance(self._events[0], AddressCreatingEvent)
self.assertEqual(self._events[0].email, 'anne@example.com')
self.assertIsInstance(self._events[1], AddressCreatedEvent)
self.assertEqual(self._events[1].address, address)
def test_delete_address_event(self):
address = self._usermanager.create_address('anne@example.com')
with event_subscribers(self._record_event):
self._usermanager.delete_address(address)
self.assertEqual(len(self._events), 2)
self.assertIsInstance(self._events[0], AddressDeletingEvent)
self.assertEqual(self._events[0].address, address)
self.assertIsInstance(self._events[1], AddressDeletedEvent)
self.assertEqual(self._events[1].email, 'anne@example.com')
def test_create_user_with_existing_address(self):
# LP: #1418280. If a user is created when an email address is passed
# in, and that address already exists, the user object should not get
# created.
# Create the address we're going to try to duplicate.
self._usermanager.create_address('anne@example.com')
# There are no users.
self.assertEqual(len(list(self._usermanager.users)), 0)
# Now create the user with an already existing address.
with self.assertRaises(ExistingAddressError) as cm:
self._usermanager.create_user('anne@example.com')
self.assertEqual(cm.exception.address, 'anne@example.com')
# There are still no users.
self.assertEqual(len(list(self._usermanager.users)), 0)
def test_make_new_user(self):
# Neither the user nor address objects exist yet.
self.assertIsNone(self._usermanager.get_user('anne@example.com'))
self.assertIsNone(self._usermanager.get_address('anne@example.com'))
user = self._usermanager.make_user('anne@example.com', 'Anne Person')
self.assertIn('anne@example.com',
[address.email for address in user.addresses])
addresses = list(user.addresses)
self.assertEqual(len(addresses), 1)
address = addresses[0]
self.assertEqual(address.email, 'anne@example.com')
self.assertEqual(address.display_name, 'Anne Person')
self.assertEqual(address.user.display_name, 'Anne Person')
self.assertIs(address.user, user)
def test_make_linked_user(self):
# The address exists, but there is no linked user.
self.assertIsNone(self._usermanager.get_user('anne@example.com'))
address = self._usermanager.create_address('anne@example.com')
user = self._usermanager.make_user('anne@example.com', 'Anne Person')
self.assertIsNotNone(address.user)
self.assertIs(user, address.user)
self.assertIn(address, user.addresses)
self.assertEqual(user.display_name, 'Anne Person')
def test_make_user_exists(self):
user = self._usermanager.create_user('anne@example.com', 'Anne Person')
other_user = self._usermanager.make_user('anne@example.com')
self.assertIs(user, other_user)
def test_get_user_by_id(self):
original = self._usermanager.make_user('anne@example.com')
copy = self._usermanager.get_user_by_id(original.user_id)
self.assertEqual(original, copy)
def test_delete_user(self):
user = self._usermanager.make_user('anne@example.com', 'Anne Person')
address = self._usermanager.create_address('anne.address@example.com')
address.verified_on = now()
user.preferred_address = address
# Subscribe the user and the address to a list.
mlist = create_list('ant@example.com')
mlist.subscribe(user)
mlist.subscribe(address)
# Now delete the user.
self._usermanager.delete_user(user)
# Flush the database to provoke an integrity error on PostgreSQL
# without the fix.
config.db.store.flush()
self.assertIsNone(self._usermanager.get_user('anne@example.com'))
self.assertIsNone(
self._usermanager.get_address('anne.address@example.com'))
def test_delete_address(self):
address = self._usermanager.create_address('anne@example.com')
address.verified_on = now()
# Subscribe the address to a list.
mlist = create_list('ant@example.com')
mlist.subscribe(address)
# Set an autorespond record.
response_set = IAutoResponseSet(mlist)
response_set.response_sent(address, Response.hold)
# And add a digest record.
mlist.send_one_last_digest_to(address, DeliveryMode.plaintext_digests)
# Now delete the address.
self._usermanager.delete_address(address)
# Flush the database to provoke an integrity error on PostgreSQL
# without the fix.
config.db.store.flush()
self.assertIsNone(self._usermanager.get_address('anne@example.com'))
|
