summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/mailman/docs/NEWS.rst6
-rw-r--r--src/mailman/rest/addresses.py3
-rw-r--r--src/mailman/rest/tests/test_addresses.py116
-rw-r--r--src/mailman/rest/users.py71
4 files changed, 125 insertions, 71 deletions
diff --git a/src/mailman/docs/NEWS.rst b/src/mailman/docs/NEWS.rst
index 0a4b2fb51..daa0e8bfe 100644
--- a/src/mailman/docs/NEWS.rst
+++ b/src/mailman/docs/NEWS.rst
@@ -63,6 +63,12 @@ REST
internal change only.
* The JSON representation `http_etag` key uses an algorithm that is
insensitive to Python's dictionary sort order.
+ * The address resource now has an additional '/user' sub-resource which can
+ be used to GET the address's linked user if there is one. This
+ sub-resource also supports POST to link an unlinked address (with an
+ optional 'auto_create' flag), and PUT to link the address to a different
+ user. It also supports DELETE to unlink the address. (LP: #1312884)
+ Given by Aurélien Bompard based on work by nicolask.
3.0 beta 4 -- "Time and Motion"
diff --git a/src/mailman/rest/addresses.py b/src/mailman/rest/addresses.py
index 4321b7664..f8516bc37 100644
--- a/src/mailman/rest/addresses.py
+++ b/src/mailman/rest/addresses.py
@@ -164,7 +164,8 @@ class AnAddress(_AddressBase):
"""/addresses/<email>/user"""
if self._address is None:
return NotFound(), []
- from mailman.rest.users import AddressUser # avoid circular imports
+ # Avoid circular imports.
+ from mailman.rest.users import AddressUser
return AddressUser(self._address)
diff --git a/src/mailman/rest/tests/test_addresses.py b/src/mailman/rest/tests/test_addresses.py
index 9d2d2bd24..bbdd7d763 100644
--- a/src/mailman/rest/tests/test_addresses.py
+++ b/src/mailman/rest/tests/test_addresses.py
@@ -208,42 +208,57 @@ class TestAddresses(unittest.TestCase):
self.assertEqual(cm.exception.code, 404)
def test_address_with_user(self):
+ # An address which is already linked to a user has a 'user' key in the
+ # JSON representation.
with transaction():
- anne = getUtility(IUserManager).create_user('anne@example.com')
+ getUtility(IUserManager).create_user('anne@example.com')
json, headers = call_api(
- 'http://localhost:9001/3.0/addresses/anne@example.com')
+ 'http://localhost:9001/3.0/addresses/anne@example.com')
self.assertEqual(headers['status'], '200')
- self.assertIn("user", json)
- self.assertEqual(json["user"], "http://localhost:9001/3.0/users/1")
+ self.assertEqual(json['user'], 'http://localhost:9001/3.0/users/1')
def test_address_without_user(self):
+ # The 'user' key is missing from the JSON representation of an address
+ # with no linked user.
with transaction():
- anne = getUtility(IUserManager).create_address('anne@example.com')
+ getUtility(IUserManager).create_address('anne@example.com')
json, headers = call_api(
- 'http://localhost:9001/3.0/addresses/anne@example.com')
+ 'http://localhost:9001/3.0/addresses/anne@example.com')
self.assertEqual(headers['status'], '200')
- self.assertNotIn("user", json)
+ self.assertNotIn('user', json)
def test_user_subresource_on_unlinked_address(self):
+ # Trying to access the 'user' subresource on an address that is not
+ # linked to a user will return a 404 error.
with transaction():
- anne = getUtility(IUserManager).create_address('anne@example.com')
+ getUtility(IUserManager).create_address('anne@example.com')
with self.assertRaises(HTTPError) as cm:
call_api(
'http://localhost:9001/3.0/addresses/anne@example.com/user')
self.assertEqual(cm.exception.code, 404)
def test_user_subresource(self):
+ # For an address which is linked to a user, accessing the user
+ # subresource of the address path returns the user JSON representation.
user_manager = getUtility(IUserManager)
with transaction():
- anne = user_manager.create_user('anne@example.com', 'Anne')
+ user_manager.create_user('anne@example.com', 'Anne')
json, headers = call_api(
- 'http://localhost:9001/3.0/addresses/anne@example.com/user')
+ 'http://localhost:9001/3.0/addresses/anne@example.com/user')
self.assertEqual(headers['status'], '200')
- self.assertEqual(json["user_id"], 1)
- self.assertEqual(json["display_name"], "Anne")
- self.assertEqual(json["self_link"], "http://localhost:9001/3.0/users/1")
+ self.assertEqual(json['user_id'], 1)
+ self.assertEqual(json['display_name'], 'Anne')
+ user_resource = json['self_link']
+ self.assertEqual(user_resource, 'http://localhost:9001/3.0/users/1')
+ # The self_link points to the correct user.
+ json, headers = call_api(user_resource)
+ self.assertEqual(json['user_id'], 1)
+ self.assertEqual(json['display_name'], 'Anne')
+ self.assertEqual(json['self_link'], user_resource)
def test_user_subresource_post(self):
+ # If the address is not yet linked to a user, POSTing a user id to the
+ # 'user' subresource links the address to the given user.
user_manager = getUtility(IUserManager)
with transaction():
anne = user_manager.create_user('anne.person@example.org', 'Anne')
@@ -258,6 +273,8 @@ class TestAddresses(unittest.TestCase):
['anne.person@example.org', 'anne@example.com'])
def test_user_subresource_post_new_user(self):
+ # If the address is not yet linked to a user, POSTing to the 'user'
+ # subresources creates a new user object and links it to the address.
user_manager = getUtility(IUserManager)
with transaction():
anne_addr = user_manager.create_address('anne@example.com')
@@ -267,7 +284,7 @@ class TestAddresses(unittest.TestCase):
})
self.assertEqual(headers['status'], '201')
anne = user_manager.get_user('anne@example.com')
- self.assertTrue(anne is not None)
+ self.assertIsNotNone(anne)
self.assertEqual(anne.display_name, 'Anne')
self.assertEqual([a.email for a in anne.addresses],
['anne@example.com'])
@@ -276,8 +293,10 @@ class TestAddresses(unittest.TestCase):
'http://localhost:9001/3.0/users/1')
def test_user_subresource_post_conflict(self):
+ # If the address is already linked to a user, trying to link it to
+ # another user produces a 409 Conflict error.
with transaction():
- anne = getUtility(IUserManager).create_user('anne@example.com')
+ getUtility(IUserManager).create_user('anne@example.com')
with self.assertRaises(HTTPError) as cm:
call_api(
'http://localhost:9001/3.0/addresses/anne@example.com/user', {
@@ -285,60 +304,79 @@ class TestAddresses(unittest.TestCase):
})
self.assertEqual(cm.exception.code, 409)
- def test_user_subresource_post_new_user_no_autocreate(self):
+ def test_user_subresource_post_new_user_no_auto_create(self):
+ # By default, POSTing to the 'user' resource of an unlinked address
+ # will automatically create the user. By setting a boolean
+ # 'auto_create' flag to false, you can prevent this.
with transaction():
- anne = getUtility(IUserManager).create_address('anne@example.com')
+ getUtility(IUserManager).create_address('anne@example.com')
with self.assertRaises(HTTPError) as cm:
json, headers = call_api(
'http://localhost:9001/3.0/addresses/anne@example.com/user', {
'display_name': 'Anne',
- 'autocreate': 0,
+ 'auto_create': 0,
})
- print("test_user_subresource_post_new_user_no_autocreate", headers, json)
- self.assertEqual(cm.exception.code, 404)
+ self.assertEqual(cm.exception.code, 403)
def test_user_subresource_unlink(self):
+ # By DELETEing the usr subresource, you can unlink a user from an
+ # address.
user_manager = getUtility(IUserManager)
with transaction():
- anne = user_manager.create_user('anne@example.com')
+ user_manager.create_user('anne@example.com')
response, headers = call_api(
- 'http://localhost:9001/3.0/addresses/anne@example.com/user',
- method="DELETE")
- self.assertEqual(headers["status"], '204')
+ 'http://localhost:9001/3.0/addresses/anne@example.com/user',
+ method='DELETE')
+ self.assertEqual(headers['status'], '204')
anne_addr = user_manager.get_address('anne@example.com')
- self.assertTrue(anne_addr.user is None, "The address is still linked")
- self.assertTrue(user_manager.get_user('anne@example.com') is None)
+ self.assertIsNone(anne_addr.user, 'The address is still linked')
+ self.assertIsNone(user_manager.get_user('anne@example.com'))
+
+ def test_user_subresource_unlink_unlinked(self):
+ # If you try to unlink an unlinked address, you get a 404 error.
+ user_manager = getUtility(IUserManager)
+ with transaction():
+ user_manager.create_address('anne@example.com')
+ with self.assertRaises(HTTPError) as cm:
+ response, headers = call_api(
+ 'http://localhost:9001/3.0/addresses/anne@example.com/user',
+ method='DELETE')
+ self.assertEqual(cm.exception.code, 404)
def test_user_subresource_put(self):
+ # By PUTing to the 'user' resource, you can change the user that an
+ # address is linked to.
user_manager = getUtility(IUserManager)
with transaction():
- anne_1 = user_manager.create_user('anne@example.com', 'Anne 1')
- anne_2 = user_manager.create_user(display_name='Anne 2')
+ anne = user_manager.create_user('anne@example.com', 'Anne')
+ bart = user_manager.create_user(display_name='Bart')
response, headers = call_api(
'http://localhost:9001/3.0/addresses/anne@example.com/user', {
- 'user_id': anne_2.user_id.int,
- }, method="PUT")
+ 'user_id': bart.user_id.int,
+ }, method='PUT')
self.assertEqual(headers['status'], '200')
- self.assertEqual(anne_1.addresses, [])
- self.assertEqual([a.email for a in anne_2.addresses],
+ self.assertEqual(anne.addresses, [])
+ self.assertEqual([address.email for address in bart.addresses],
['anne@example.com'])
- self.assertEqual(anne_2,
- user_manager.get_address('anne@example.com').user)
+ self.assertEqual(bart,
+ user_manager.get_address('anne@example.com').user)
def test_user_subresource_put_create(self):
+ # PUTing to the 'user' resource creates the user, just like with POST.
user_manager = getUtility(IUserManager)
with transaction():
anne = user_manager.create_user('anne@example.com', 'Anne')
response, headers = call_api(
'http://localhost:9001/3.0/addresses/anne@example.com/user', {
'email': 'anne.person@example.org',
- }, method="PUT")
+ }, method='PUT')
self.assertEqual(headers['status'], '201')
self.assertEqual(anne.addresses, [])
anne_person = user_manager.get_user('anne.person@example.org')
- self.assertTrue(anne_person is not None)
- self.assertEqual(sorted([a.email for a in anne_person.addresses]),
- ["anne.person@example.org", "anne@example.com"])
+ self.assertIsNotNone(anne_person)
+ self.assertEqual(
+ sorted([address.email for address in anne_person.addresses]),
+ ['anne.person@example.org', 'anne@example.com'])
anne_addr = user_manager.get_address('anne@example.com')
- self.assertTrue(anne_addr is not None)
+ self.assertIsNotNone(anne_addr)
self.assertEqual(anne_addr.user, anne_person)
diff --git a/src/mailman/rest/users.py b/src/mailman/rest/users.py
index 80c53b565..7ab1d6818 100644
--- a/src/mailman/rest/users.py
+++ b/src/mailman/rest/users.py
@@ -27,6 +27,7 @@ __all__ = [
]
+from lazr.config import as_boolean
from passlib.utils import generate_password as generate
from uuid import UUID
from zope.component import getUtility
@@ -39,8 +40,8 @@ from mailman.interfaces.usermanager import IUserManager
from mailman.rest.addresses import UserAddresses
from mailman.rest.helpers import (
BadRequest, CollectionMixin, GetterSetter, NotFound, bad_request, child,
- created, etag, forbidden, no_content, not_found, okay, paginate, path_to,
- conflict)
+ conflict, created, etag, forbidden, no_content, not_found, okay, paginate,
+ path_to)
from mailman.rest.preferences import Preferences
from mailman.rest.validator import PatchValidator, Validator
@@ -63,6 +64,7 @@ ATTRIBUTES = dict(
cleartext_password=PasswordEncrypterGetterSetter(),
)
+
CREATION_FIELDS = dict(
email=unicode,
display_name=unicode,
@@ -73,21 +75,21 @@ CREATION_FIELDS = dict(
def create_user(arguments, response):
"""Create a new user."""
- # We can't pass the 'password' argument to the user creation method,
- # so strip that out (if it exists), then create the user, adding the
- # password after the fact if successful.
+ # We can't pass the 'password' argument to the user creation method, so
+ # strip that out (if it exists), then create the user, adding the password
+ # after the fact if successful.
password = arguments.pop('password', None)
try:
user = getUtility(IUserManager).create_user(**arguments)
except ExistingAddressError as error:
bad_request(
- response, b'Address already exists: {0}'.format(error.address))
- return
+ response, b'Address already exists: {}'.format(error.address))
+ return None
if password is None:
# This will have to be reset since it cannot be retrieved.
password = generate(int(config.passwords.password_length))
user.password = config.password_context.encrypt(password)
- location = path_to('users/{0}'.format(user.user_id.int))
+ location = path_to('users/{}'.format(user.user_id.int))
created(response, location)
return user
@@ -106,7 +108,7 @@ class _UserBase(CollectionMixin):
resource = dict(
user_id=user_id,
created_on=user.created_on,
- self_link=path_to('users/{0}'.format(user_id)),
+ self_link=path_to('users/{}'.format(user_id)),
)
# Add the password attribute, only if the user has a password. Same
# with the real name. These could be None or the empty string.
@@ -280,12 +282,15 @@ class AddressUser(_UserBase):
if self._user:
conflict(response)
return
- # Check for an existing user
+ # When creating a linked user by POSTing, the user either must already
+ # exist, or it can be automatically created, if the auto_create flag
+ # is given and true (if missing, it defaults to true). However, in
+ # this case we do not accept 'email' as a POST field.
fields = CREATION_FIELDS.copy()
- del fields["email"]
- fields["user_id"] = int
- fields["autocreate"] = int
- fields["_optional"] = fields["_optional"] + ('user_id', 'autocreate')
+ del fields['email']
+ fields['user_id'] = int
+ fields['auto_create'] = as_boolean
+ fields['_optional'] = fields['_optional'] + ('user_id', 'auto_create')
try:
validator = Validator(**fields)
arguments = validator(request)
@@ -293,32 +298,34 @@ class AddressUser(_UserBase):
bad_request(response, str(error))
return
user_manager = getUtility(IUserManager)
- if "user_id" in arguments:
- user_id = UUID(int=arguments["user_id"])
+ if 'user_id' in arguments:
+ raw_uid = arguments['user_id']
+ user_id = UUID(int=raw_uid)
user = user_manager.get_user_by_id(user_id)
if user is None:
- not_found(response, b"No user with ID {0}".format(
- arguments["user_id"]))
+ not_found(response, b'No user with ID {}'.format(raw_uid))
return
okay(response)
else:
- if arguments.get("autocreate", True): # autocreate by default
- arguments.pop("autocreate", None)
- user = create_user(arguments, response) # will set "201 Created"
+ auto_create = arguments.pop('auto_create', True)
+ if auto_create:
+ # This sets the 201 or 400 status.
+ user = create_user(arguments, response)
+ if user is None:
+ return
else:
- not_found(response, b"No such user, and automatic "
- "creation is disabled.")
+ forbidden(response)
return
user.link(self._address)
def on_put(self, request, response):
- """Set or replace the addresse's user."""
+ """Set or replace the addresses's user."""
if self._user:
self._user.unlink(self._address)
- # Process post data and check for an existing user
+ # Process post data and check for an existing user.
fields = CREATION_FIELDS.copy()
- fields["user_id"] = int
- fields["_optional"] = fields["_optional"] + ('user_id', 'email')
+ fields['user_id'] = int
+ fields['_optional'] = fields['_optional'] + ('user_id', 'email')
try:
validator = Validator(**fields)
arguments = validator(request)
@@ -327,15 +334,17 @@ class AddressUser(_UserBase):
return
user_manager = getUtility(IUserManager)
if 'user_id' in arguments:
- user_id = UUID(int=arguments["user_id"])
+ raw_uid = arguments['user_id']
+ user_id = UUID(int=raw_uid)
user = user_manager.get_user_by_id(user_id)
if user is None:
- not_found(response, b"No user with ID {0}".format(
- arguments["user_id"]))
+ not_found(response, b'No user with ID {}'.format(raw_uid))
return
okay(response)
else:
- user = create_user(arguments, response) # will set "201 Created"
+ user = create_user(arguments, response)
+ if user is None:
+ return
user.link(self._address)
generated by cgit v1.2.3-70-g09d2 (git 2.51.2) at 2025-11-09 12:54:41 +0000