# Copyright (C) 2011-2016 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 .
"""REST membership tests."""
__all__ = [
'TestAPI31Members',
'TestMembership',
'TestNonmembership',
]
import unittest
from mailman.app.lifecycle import create_list
from mailman.config import config
from mailman.database.transaction import transaction
from mailman.interfaces.member import DeliveryMode, MemberRole
from mailman.interfaces.usermanager import IUserManager
from mailman.testing.helpers import (
TestableMaster, call_api, get_lmtp_client, make_testable_runner,
subscribe, wait_for_webservice)
from mailman.runners.incoming import IncomingRunner
from mailman.testing.layers import ConfigLayer, RESTLayer
from mailman.utilities.datetime import now
from urllib.error import HTTPError
from zope.component import getUtility
def _set_preferred(user):
preferred = list(user.addresses)[0]
preferred.verified_on = now()
user.preferred_address = preferred
class TestMembership(unittest.TestCase):
layer = RESTLayer
def setUp(self):
with transaction():
self._mlist = create_list('test@example.com')
self._usermanager = getUtility(IUserManager)
def test_try_to_join_missing_list(self):
# A user tries to join a non-existent list.
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members', {
'list_id': 'missing.example.com',
'subscriber': 'nobody@example.com',
})
self.assertEqual(cm.exception.code, 400)
self.assertEqual(cm.exception.reason, b'No such list')
def test_try_to_leave_missing_list(self):
# A user tries to leave a non-existent list.
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/lists/missing@example.com'
'/member/nobody@example.com',
method='DELETE')
self.assertEqual(cm.exception.code, 404)
def test_try_to_leave_list_with_bogus_address(self):
# Try to leave a mailing list using an invalid membership address.
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1', method='DELETE')
self.assertEqual(cm.exception.code, 404)
def test_try_to_leave_a_list_twice(self):
with transaction():
anne = self._usermanager.create_address('anne@example.com')
self._mlist.subscribe(anne)
url = 'http://localhost:9001/3.0/members/1'
content, response = call_api(url, method='DELETE')
# For a successful DELETE, the response code is 204 and there is no
# content.
self.assertEqual(content, None)
self.assertEqual(response.status, 204)
with self.assertRaises(HTTPError) as cm:
call_api(url, method='DELETE')
self.assertEqual(cm.exception.code, 404)
def test_try_to_join_a_list_twice(self):
with transaction():
anne = self._usermanager.create_address('anne@example.com')
self._mlist.subscribe(anne)
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members', {
'list_id': 'test.example.com',
'subscriber': 'anne@example.com',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
self.assertEqual(cm.exception.code, 409)
self.assertEqual(cm.exception.reason, b'Member already subscribed')
def test_subscribe_user_without_preferred_address(self):
with transaction():
getUtility(IUserManager).create_user('anne@example.com')
# Subscribe the user to the mailing list by hex UUID.
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.1/members', {
'list_id': 'test.example.com',
'subscriber': '00000000000000000000000000000001',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
self.assertEqual(cm.exception.code, 400)
self.assertEqual(cm.exception.reason, b'User has no preferred address')
def test_add_member_with_mixed_case_email(self):
# LP: #1425359 - Mailman is case-perserving, case-insensitive. This
# test subscribes the lower case address and ensures the original mixed
# case address can't be subscribed.
with transaction():
anne = self._usermanager.create_address('anne@example.com')
self._mlist.subscribe(anne)
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members', {
'list_id': 'test.example.com',
'subscriber': 'ANNE@example.com',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
self.assertEqual(cm.exception.code, 409)
self.assertEqual(cm.exception.reason, b'Member already subscribed')
def test_add_member_with_lower_case_email(self):
# LP: #1425359 - Mailman is case-perserving, case-insensitive. This
# test subscribes the mixed case address and ensures the lower cased
# address can't be added.
with transaction():
anne = self._usermanager.create_address('ANNE@example.com')
self._mlist.subscribe(anne)
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members', {
'list_id': 'test.example.com',
'subscriber': 'anne@example.com',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
self.assertEqual(cm.exception.code, 409)
self.assertEqual(cm.exception.reason, b'Member already subscribed')
def test_join_with_invalid_delivery_mode(self):
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members', {
'list_id': 'test.example.com',
'subscriber': 'anne@example.com',
'display_name': 'Anne Person',
'delivery_mode': 'invalid-mode',
})
self.assertEqual(cm.exception.code, 400)
self.assertEqual(cm.exception.reason,
b'Cannot convert parameters: delivery_mode')
def test_join_email_contains_slash(self):
content, response = call_api('http://localhost:9001/3.0/members', {
'list_id': 'test.example.com',
'subscriber': 'hugh/person@example.com',
'display_name': 'Hugh Person',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
self.assertEqual(content, None)
self.assertEqual(response.status, 201)
self.assertEqual(response['location'],
'http://localhost:9001/3.0/members/1')
# Reset any current transaction.
config.db.abort()
members = list(self._mlist.members.members)
self.assertEqual(len(members), 1)
self.assertEqual(members[0].address.email, 'hugh/person@example.com')
def test_join_as_user_with_preferred_address(self):
with transaction():
anne = self._usermanager.create_user('anne@example.com')
_set_preferred(anne)
self._mlist.subscribe(anne)
content, response = call_api('http://localhost:9001/3.0/members')
self.assertEqual(response.status, 200)
self.assertEqual(int(content['total_size']), 1)
entry_0 = content['entries'][0]
self.assertEqual(entry_0['self_link'],
'http://localhost:9001/3.0/members/1')
self.assertEqual(entry_0['role'], 'member')
self.assertEqual(entry_0['user'], 'http://localhost:9001/3.0/users/1')
self.assertEqual(entry_0['email'], 'anne@example.com')
self.assertEqual(
entry_0['address'],
'http://localhost:9001/3.0/addresses/anne@example.com')
self.assertEqual(entry_0['list_id'], 'test.example.com')
def test_member_changes_preferred_address(self):
with transaction():
anne = self._usermanager.create_user('anne@example.com')
_set_preferred(anne)
self._mlist.subscribe(anne)
# Take a look at Anne's current membership.
content, response = call_api('http://localhost:9001/3.0/members')
self.assertEqual(int(content['total_size']), 1)
entry_0 = content['entries'][0]
self.assertEqual(entry_0['email'], 'anne@example.com')
self.assertEqual(
entry_0['address'],
'http://localhost:9001/3.0/addresses/anne@example.com')
# Anne registers a new address and makes it her preferred address.
# There are no changes to her membership.
with transaction():
new_preferred = anne.register('aperson@example.com')
new_preferred.verified_on = now()
anne.preferred_address = new_preferred
# Take another look at Anne's current membership.
content, response = call_api('http://localhost:9001/3.0/members')
self.assertEqual(int(content['total_size']), 1)
entry_0 = content['entries'][0]
self.assertEqual(entry_0['email'], 'aperson@example.com')
self.assertEqual(
entry_0['address'],
'http://localhost:9001/3.0/addresses/aperson@example.com')
def test_get_nonexistent_member(self):
# /members/ returns 404
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/bogus')
self.assertEqual(cm.exception.code, 404)
def test_patch_nonexistent_member(self):
# /members/ PATCH returns 404
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/801',
{}, method='PATCH')
self.assertEqual(cm.exception.code, 404)
def test_patch_member_bogus_attribute(self):
# /members/ PATCH 'bogus' returns 400
with transaction():
anne = self._usermanager.create_address('anne@example.com')
self._mlist.subscribe(anne)
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1', {
'powers': 'super',
}, method='PATCH')
self.assertEqual(cm.exception.code, 400)
self.assertEqual(cm.exception.reason, b'Unexpected parameters: powers')
def test_member_all_without_preferences(self):
# /members//all should return a 404 when it isn't trailed by
# `preferences`
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1/all')
self.assertEqual(cm.exception.code, 404)
def test_patch_member_invalid_moderation_action(self):
# /members/ PATCH with invalid 'moderation_action' returns 400.
with transaction():
anne = self._usermanager.create_address('anne@example.com')
self._mlist.subscribe(anne)
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1', {
'moderation_action': 'invalid',
}, method='PATCH')
self.assertEqual(cm.exception.code, 400)
self.assertEqual(cm.exception.reason,
b'Cannot convert parameters: moderation_action')
def test_bad_preferences_url(self):
with transaction():
subscribe(self._mlist, 'Anne')
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1/preferences/bogus')
self.assertEqual(cm.exception.code, 404)
def test_not_a_member_preferences(self):
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1/preferences')
self.assertEqual(cm.exception.code, 404)
def test_not_a_member_all_preferences(self):
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.0/members/1/all/preferences')
self.assertEqual(cm.exception.code, 404)
def test_delete_other_role(self):
with transaction():
subscribe(self._mlist, 'Anne', MemberRole.moderator)
response, headers = call_api(
'http://localhost:9001/3.0/members/1',
method='DELETE')
self.assertEqual(headers.status, 204)
self.assertEqual(len(list(self._mlist.moderators.members)), 0)
class CustomLayer(ConfigLayer):
"""Custom layer which starts both the REST and LMTP servers."""
server = None
client = None
@classmethod
def _wait_for_both(cls):
cls.client = get_lmtp_client(quiet=True)
wait_for_webservice()
@classmethod
def setUp(cls):
assert cls.server is None, 'Layer already set up'
cls.server = TestableMaster(cls._wait_for_both)
cls.server.start('lmtp', 'rest')
@classmethod
def tearDown(cls):
assert cls.server is not None, 'Layer is not set up'
cls.server.stop()
cls.server = None
class TestNonmembership(unittest.TestCase):
layer = CustomLayer
def setUp(self):
with transaction():
self._mlist = create_list('test@example.com')
self._usermanager = getUtility(IUserManager)
def _go(self, message):
lmtp = get_lmtp_client(quiet=True)
lmtp.lhlo('remote.example.org')
lmtp.sendmail('nonmember@example.com', ['test@example.com'], message)
lmtp.close()
# The message will now be sitting in the `in` queue. Run the incoming
# runner once to process it, which should result in the nonmember
# showing up.
inq = make_testable_runner(IncomingRunner, 'in')
inq.run()
def test_nonmember_findable_after_posting(self):
# A nonmember we have never seen before posts a message to the mailing
# list. They are findable through the /members/find API using a role
# of nonmember.
self._go("""\
From: nonmember@example.com
To: test@example.com
Subject: Nonmember post
Message-ID:
Some text.
""")
# Now use the REST API to try to find the nonmember.
response, content = call_api(
'http://localhost:9001/3.0/members/find', {
#'list_id': 'test.example.com',
'role': 'nonmember',
})
self.assertEqual(response['total_size'], 1)
nonmember = response['entries'][0]
self.assertEqual(nonmember['role'], 'nonmember')
self.assertEqual(nonmember['email'], 'nonmember@example.com')
self.assertEqual(
nonmember['address'],
'http://localhost:9001/3.0/addresses/nonmember@example.com')
# There is no user key in the JSON data because there is no user
# record associated with the address record.
self.assertNotIn('user', nonmember)
def test_linked_nonmember_findable_after_posting(self):
# Like above, a nonmember posts a message to the mailing list. In
# this case though, the nonmember already has a user record. They are
# findable through the /members/find API using a role of nonmember.
with transaction():
self._usermanager.create_user('nonmember@example.com')
self._go("""\
From: nonmember@example.com
To: test@example.com
Subject: Nonmember post
Message-ID:
Some text.
""")
# Now use the REST API to try to find the nonmember.
response, content = call_api(
'http://localhost:9001/3.0/members/find', {
#'list_id': 'test.example.com',
'role': 'nonmember',
})
self.assertEqual(response['total_size'], 1)
nonmember = response['entries'][0]
self.assertEqual(nonmember['role'], 'nonmember')
self.assertEqual(nonmember['email'], 'nonmember@example.com')
self.assertEqual(
nonmember['address'],
'http://localhost:9001/3.0/addresses/nonmember@example.com')
# There is a user key in the JSON data because the address had
# previously been linked to a user record.
self.assertEqual(nonmember['user'],
'http://localhost:9001/3.0/users/1')
class TestAPI31Members(unittest.TestCase):
layer = RESTLayer
def setUp(self):
with transaction():
self._mlist = create_list('ant@example.com')
def test_member_ids_are_hex(self):
with transaction():
subscribe(self._mlist, 'Anne')
subscribe(self._mlist, 'Bart')
response, headers = call_api('http://localhost:9001/3.1/members')
entries = response['entries']
self.assertEqual(len(entries), 2)
self.assertEqual(
entries[0]['self_link'],
'http://localhost:9001/3.1/members/00000000000000000000000000000001')
self.assertEqual(
entries[0]['member_id'],
'00000000000000000000000000000001')
self.assertEqual(
entries[0]['user'],
'http://localhost:9001/3.1/users/00000000000000000000000000000001')
self.assertEqual(
entries[1]['self_link'],
'http://localhost:9001/3.1/members/00000000000000000000000000000002')
self.assertEqual(
entries[1]['member_id'],
'00000000000000000000000000000002')
self.assertEqual(
entries[1]['user'],
'http://localhost:9001/3.1/users/00000000000000000000000000000002')
def test_get_member_id_by_hex(self):
with transaction():
subscribe(self._mlist, 'Anne')
response, headers = call_api(
'http://localhost:9001/3.1/members/00000000000000000000000000000001')
self.assertEqual(
response['member_id'],
'00000000000000000000000000000001')
self.assertEqual(
response['self_link'],
'http://localhost:9001/3.1/members/00000000000000000000000000000001')
self.assertEqual(
response['user'],
'http://localhost:9001/3.1/users/00000000000000000000000000000001')
self.assertEqual(
response['address'],
'http://localhost:9001/3.1/addresses/aperson@example.com')
def test_cannot_get_member_id_by_int(self):
with transaction():
subscribe(self._mlist, 'Anne')
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.1/members/1')
self.assertEqual(cm.exception.code, 404)
def test_preferences(self):
with transaction():
member = subscribe(self._mlist, 'Anne')
member.preferences.delivery_mode = DeliveryMode.summary_digests
response, headers = call_api(
'http://localhost:9001/3.1/members'
'/00000000000000000000000000000001/preferences')
self.assertEqual(response['delivery_mode'], 'summary_digests')
def test_all_preferences(self):
with transaction():
member = subscribe(self._mlist, 'Anne')
member.preferences.delivery_mode = DeliveryMode.summary_digests
response, headers = call_api(
'http://localhost:9001/3.1/members'
'/00000000000000000000000000000001/all/preferences')
self.assertEqual(response['delivery_mode'], 'summary_digests')
def test_create_new_membership_by_hex(self):
with transaction():
user = getUtility(IUserManager).create_user('anne@example.com')
_set_preferred(user)
# Subscribe the user to the mailing list by hex UUID.
response, headers = call_api(
'http://localhost:9001/3.1/members', {
'list_id': 'ant.example.com',
'subscriber': '00000000000000000000000000000001',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
self.assertEqual(headers.status, 201)
self.assertEqual(
headers['location'],
'http://localhost:9001/3.1/members/00000000000000000000000000000001'
)
def test_create_new_owner_by_hex(self):
with transaction():
user = getUtility(IUserManager).create_user('anne@example.com')
_set_preferred(user)
# Subscribe the user to the mailing list by hex UUID.
response, headers = call_api(
'http://localhost:9001/3.1/members', {
'list_id': 'ant.example.com',
'subscriber': '00000000000000000000000000000001',
'role': 'owner',
})
self.assertEqual(headers.status, 201)
self.assertEqual(
headers['location'],
'http://localhost:9001/3.1/members/00000000000000000000000000000001'
)
def test_cannot_create_new_membership_by_int(self):
with transaction():
user = getUtility(IUserManager).create_user('anne@example.com')
_set_preferred(user)
# We can't use the int representation of the UUID with API 3.1.
with self.assertRaises(HTTPError) as cm:
call_api('http://localhost:9001/3.1/members', {
'list_id': 'ant.example.com',
'subscriber': '1',
'pre_verified': True,
'pre_confirmed': True,
'pre_approved': True,
})
# This is a bad request because the `subscriber` value isn't something
# that's known to the system, in API 3.1. It's not technically a 404
# because that's reserved for URL lookups.
self.assertEqual(cm.exception.code, 400)