Merge lp:~mpontillo/maas/make-ipaddresses-api-great-again--bug-1629061 into lp:~maas-committers/maas/trunk

[Mike Pontillo]

Make the 'ipaddresses' API do what everyone expected it to do.

 * Modify 'read' operation to show additional relevant fields, so that
   API users can determine which address they want to release.
   - Adds 'user' field so that administrators can verify that addresses
     they want to release belong to a particular user
   - Add 'alloc_type_name' field, so that administrators can verify the
     expected [human readable] type of the address.
   - Add 'interface_set' field, so that administrators can check which
     interfaces are linked to the each IP address.
 * Add 'user=<user>' option to 'read' operation, so that administrators
   can list addresses belonging to a specific user.
 * Add 'all=true' option to 'read' operation, so that administrators
   can list all addresses without restriction. (Normally, only
   "User reserved" addresses are shown. This option also removes the
   'user' filter.)
 * Add 'ip=<ip-address>' option to 'read' operation, so that
   administrators can verify that a specific address is the intended
   address to release before releasing it.
 * Allow the 'release' operation to release any IP address, not just
   a user-reserved IP address belonging to the current user.
   (Requires the 'force=true' parameter to be used, otherwise restricts
   the release operation the same way as before.)
 * Add 'ip=<ip-address>' optional argument to 'reserve' operation, so
   that the user experience is consistent across all operations on
   the 'ipaddresses' endpoint. This replaces the now-deprecated
   'ip_address' parameter, which may still be used, and acts as an alias
   to the 'ip' parameter (if the 'ip' parameter is not specified).
 * Drive-by DRY up for displayed interface fields in the API.

[Gavin Panella (allenap)]

A few things need fixing, like the "Undefined" default in allow_type_name, but it's a good change in all, especially for consumers of the API.

This could have been done in multiple branches. This *should* have been done in multiple branches. If not developed that way, at least split out for review. It is much easier to review small changes. This change is hardly egregious and it wasn't too hard to review, but it could have been better.

If nothing else know this: as a reviewer, when there are multiple branches to review, I will always review short branches first.

review: Needs Fixing

[Gavin Panella (allenap)]

The alloc_type_name thing is the only thing that shouldn't land, but as agreed you're going to change it to None or the empty string, so +1. That's not to say I think you should ignore my other comments :)

review: Approve

[Mike Pontillo (mpontillo)]

Thanks for the feedback, Gavin. (I've changed `alloc_type_name` it to an empty string; will add a unit test for it as well.)

Small rant regarding multiple branches: I really wanted to propose this as multiple branches. Honest. ;-) I think my development style was getting in the way here. My process is, first I iterate on the API code and get the user experience to be how I want it. I'll iterate on changing the API code and doing end-to-end testing, periodically deploying to a live MAAS. By the time I was done with that, I had a ~400 line diff, and most of the existing tests were failing. At that time, I thought about splitting up the branch into a second --tests branch, but that didn't make sense because there was no way to land it by itself.

The key problem for me was, I didn't know where I was going with it until I was done. If I had a better plan in mind, doing 2 or 3 branches would have been easy.

Now, if we were using git, I might have used a "git rebase -i" to split out the branch into multiple, smaller feature branches. But that's a pain with bzr, and I wasn't sure if I was going to go back and change the code again after I completed the unit testing. So I left it how it was, and hoped I would't exceed our 800 line limit. Sorry. =(

In retrospect, it wouldn't have been too bad to do a branch for each API operation I changed. (though the test case refactoring I did might have gotten in the way of that.) But the return on investment isn't clear. If I spend an hour cleaning up the branch for easier review, how much time do I save you while reviewing it?

[Mike Pontillo (mpontillo)]

Some replies below. Thanks for the review!

=== modified file 'src/maasserver/api/devices.py'
--- src/maasserver/api/devices.py	2016-05-12 19:07:37 +0000
+++ src/maasserver/api/devices.py	2016-10-12 17:51:07 +0000
@@ -6,6 +6,7 @@
     "DevicesHandler",
     ]
 
+from maasserver.api.interfaces import DISPLAYED_INTERFACE_FIELDS
 from maasserver.api.logger import maaslog
 from maasserver.api.nodes import (
     NodeHandler,
@@ -35,21 +36,7 @@
     'tag_names',
     'address_ttl',
     'ip_addresses',
-    ('interface_set', (
-        'id',
-        'name',
-        'type',
-        'vlan',
-        'mac_address',
-        'parents',
-        'children',
-        'tags',
-        'enabled',
-        'links',
-        'params',
-        'discovered',
-        'effective_mtu',
-        )),
+    ('interface_set', DISPLAYED_INTERFACE_FIELDS),
     'zone',
     'node_type',
     'node_type_name',

=== modified file 'src/maasserver/api/ip_addresses.py'
--- src/maasserver/api/ip_addresses.py	2016-07-30 01:17:54 +0000
+++ src/maasserver/api/ip_addresses.py	2016-10-12 17:51:07 +0000
@@ -8,6 +8,12 @@
     ]
 
 from django.http import Http404
+from django.http.response import (
+    HttpResponseBadRequest,
+    HttpResponseForbidden,
+)
+from formencode.validators import StringBool
+from maasserver.api.interfaces import DISPLAYED_INTERFACE_FIELDS
 from maasserver.api.support import (
     operation,
     OperationsHandler,
@@ -39,6 +45,9 @@
     Subnet,
 )
 from maasserver.utils.orm import transactional
+from netaddr import IPAddress
+from netaddr.core import AddrFormatError
+from piston3.utils import rc
 from provisioningserver.logger import get_maas_logger
 
 
@@ -49,10 +58,22 @@
     """Manage IP addresses allocated by MAAS."""
     api_doc_section_name = "IP Addresses"
     model = StaticIPAddress
-    fields = ('alloc_type', 'created', 'ip', 'subnet')
+    fields = (
+        'alloc_type',
+        'alloc_type_name',
+        'created',
+        'ip',
+        'owner',
+        ('interface_set', DISPLAYED_INTERFACE_FIELDS),
+        'subnet'
+    )
     create = update = delete = None
 
     @classmethod
+    def owner(cls, static_ip_address):
+        return static_ip_address.user
+
+    @classmethod
     def resource_uri(cls, *args, **kwargs):
         return ('ipaddresses_handler', [])
 
@@ -144,8 +165,10 @@
 
         :param subnet: CIDR representation of the subnet on which the IP
             reservation is required. e.g. 10.1.2.0/24
-        :param ip_address: The IP address, which must be within
+        :param ip: The IP address, which must be within
             a known subnet.
+        :param ip_address: (Deprecated.) Alias for 'ip' parameter. Provided
+            for backward compatibility.
         :param hostname: The hostname to use for the specified IP address.  If
             no domain component is given, the default domain will be used.
         :param mac: The MAC address that should be linked to this reservation.
@@ -156,33 +179,43 @@
         Returns 503 if there are no more IP addresses available.
         """
         subnet = get_optional_param(request.POST, "subnet")
-        ip_address = get_optional_param(
-            request.POST, "ip_address")
+        ip = get_optional_param(request.POST, "ip")
+        ip_address = get_optional_param(request.POST, "ip_address")
         hostname = get_optional_param(request.POST, "hostname")
         mac_address = get_optional_param(request.POST, "mac")
 
-        form = ClaimIPForMACForm(request.POST)
+        # Fix to make the API backward compatible, yet consistent with the
+        # other APIs in this file.
+        if ip is None and ip_address is not None:
+            ip = ip_address
+
+        form = ClaimIPForMACForm(data={
+            'subnet': subnet,
+            'ip_address': ip,
+            'hostname': hostname,
+            'mac': mac_address,
+        })
         if not form.is_valid():
             raise MAASAPIValidationError(form.errors)
 
-        if ip_address is not None:
-            subnet = Subnet.objects.get_best_subnet_for_ip(ip_address)
+        if ip is not None:
+            subnet = Subnet.objects.get_best_subnet_for_ip(ip)
             if subnet is None:
                 raise MAASAPIBadRequest(
-                    "No known subnet for IP %s." % ip_address)
+                    "No known subnet for IP address: %s." % ip)
         elif subnet is not None:
             try:
                 subnet = Subnet.objects.get_object_by_specifiers_or_raise(
                     subnet)
             except Http404:
                 raise MAASAPIBadRequest(
-                    "Unable to identify subnet %s." % subnet) from None
+                    "Unable to identify subnet: %s." % subnet) from None
         else:
             raise MAASAPIBadRequest(
-                "Must supply either a subnet or a ip_address.")
+                "Must supply either the 'subnet' or the 'ip' parameter.")
 
         return self._claim_ip(
-            request.user, subnet, ip_address, mac_address,
+            request.user, subnet, ip, mac_address,
             hostname=hostname)
 
     @operation(idempotent=False)
@@ -192,21 +225,48 @@
         :param ip: The IP address to release.
         :type ip: unicode
 
+        :param force: If True, allows a MAAS administrator to force an IP
+            address to be released, even if it is not a user-reserved IP
+            address or does not belong to the requesting user. Use with
+            caution.
+        :type force: bool
+
         Returns 404 if the provided IP address is not found.
         """
         ip = get_mandatory_param(request.POST, "ip")
+        force = get_optional_param(
+            request.POST, 'force', default=False, validator=StringBool)
+
+        if force is True and not request.user.is_superuser:
+            return HttpResponseForbidden(
+                content_type='text/plain',
+                content="Force-releasing an IP address requires admin "
+                        "privileges.")
+
         form = ReleaseIPForm(request.POST)
         if not form.is_valid():
             raise MAASAPIValidationError(form.errors)
 
+        if force:
+            query_args = dict(ip=ip)
+        else:
+            query_args = dict(
+                alloc_type=IPADDRESS_TYPE.USER_RESERVED, ip=ip,
+                user=request.user)
+
         # Get the reserved IP address, or raise bad request.
         try:
-            ip_address = StaticIPAddress.objects.get(
-                alloc_type=IPADDRESS_TYPE.USER_RESERVED, ip=ip,
-                user=request.user)
+            ip_address = StaticIPAddress.objects.get(**query_args)
         except StaticIPAddress.DoesNotExist:
-            raise MAASAPIBadRequest(
-                "User reserved IP %s does not exist." % ip) from None
+            if force:
+                error = "IP address %s does not exist."
+            else:
+                error = (
+                    "IP address %s does not exist, is not a user-reserved "
+                    "address, or does not belong to the requesting user.\n"
+                    "If you are sure you want to release this address, use "
+                    "force=true as a MAAS administrator.")
+            raise MAASAPIBadRequest(error % ip) from None
 
         # Unlink the IP address from the interfaces it is connected.
         interfaces = list(ip_address.interface_set.all())
@@ -222,12 +282,83 @@
             if interface.node is None and interface.only_has_link_up():
                 interface.delete()
 
-        maaslog.info("User %s released IP %s", request.user.username, ip)
+        maaslog.info(
+            "User %s%s released IP address: %s (%s).", request.user.username,
+            " forcibly" if force else "", ip,
+            ip_address.alloc_type_name)
+
+        return rc.DELETED
 
     def read(self, request):
-        """List IPAddresses.
-
-        Get a listing of all IPAddresses allocated to the requesting user.
+        """List IP addresses known to MAAS.
+
+        By default, gets a listing of all IP addresses allocated to the
+        requesting user.
+
+        :param ip: If specified, will only display information for the
+            specified IP address.
+        :type ip: unicode (must be an IPv4 or IPv6 address)
+
+        If the requesting user is a MAAS administrator, the following options
+        may also be supplied:
+
+        :param all: If True, all reserved IP addresses will be shown. (By
+            default, only addresses of type 'User reserved' that are assigned
+            to the requesting user are shown.)
+        :type all: bool
+
+        :param owner: If specified, filters the list to show only IP addresses
+            owned by the specified username.
+        :type user: unicode
         """
-        return StaticIPAddress.objects.filter(
-            user=request.user).order_by('id')
+        _all = get_optional_param(
+            request.GET, 'all', default=False, validator=StringBool)
+        ip = get_optional_param(request.GET, 'ip', default=None)
+        owner = get_optional_param(request.GET, 'owner', default=None)
+        # If an IP address was specified, validate that it is indeed a valid
+        # IP address before trying to hit the database with it.
+        if ip is not None:
+            try:
+                IPAddress(ip)
+            except AddrFormatError:
+                return HttpResponseBadRequest(
+                    content_type='text/plain',
+                    content="Parameter 'ip' must contain an IP address.")
+        # It doesn't make sense for this API to display NULL IP addresses.
+        # These indicate things like "do DHCP on <interface>", "we observed a
+        # commissioned node connected to <subnet>", and "we would assign an
+        # automatic address to <machine-interface>, but it isn't deployed at
+        # the moment".
+        query = StaticIPAddress.objects.exclude(ip__isnull=True)
+        if _all and not request.user.is_superuser:
+            return HttpResponseForbidden(
+                content_type='text/plain',
+                content="Listing all IP addresses requires admin "
+                        "privileges.")
+        if owner is not None and not request.user.is_superuser:
+            return HttpResponseForbidden(
+                content_type='text/plain',
+                content="Listing another user's IP addresses requires admin "
+                        "privileges.")
+        # Add additional filters based on permissions, and based on the
+        # request parameters.
+        if not request.user.is_superuser:
+            # If the requesting user isn't an admin, always filter by the
+            # currently-logged-in API user.
+            query = query.filter(user=request.user)
+        elif owner is not None:
+            # If the admin specified a username filter, use that.
+            query = query.filter(user__username=owner)
+        elif _all is False:
+            # If the admin didn't specify 'all=true' (and didn't specify a
+            # specific username), only show IP addresses belonging to the
+            # admin. (This preserves API compatibility.)
+            query = query.filter(user=request.user)
+        # If we're not displaying all addresses, we also need to filter by
+        # address type (again, to preserve API compatibility).
+        if not _all:
+            query = query.filter(alloc_type=IPADDRESS_TYPE.USER_RESERVED)
+        if ip is not None:
+            query = query.filter(ip=ip)
+        query = query.order_by('id')
+        return query

=== modified file 'src/maasserver/api/machines.py'
--- src/maasserver/api/machines.py	2016-09-29 19:23:13 +0000
+++ src/maasserver/api/machines.py	2016-10-12 17:51:07 +0000
@@ -24,6 +24,7 @@
     StringBool,
 )
 from maasserver import locks
+from maasserver.api.interfaces import DISPLAYED_INTERFACE_FIELDS
 from maasserver.api.logger import maaslog
 from maasserver.api.nodes import (
     AnonNodeHandler,
@@ -111,21 +112,7 @@
     'tag_names',
     'address_ttl',
     'ip_addresses',
-    ('interface_set', (
-        'id',
-        'name',
-        'type',
-        'vlan',
-        'mac_address',
-        'parents',
-        'children',
-        'tags',
-        'enabled',
-        'links',
-        'params',
-        'discovered',
-        'effective_mtu',
-        )),
+    ('interface_set', DISPLAYED_INTERFACE_FIELDS),
     'zone',
     'disable_ipv4',
     'constraints_by_type',

=== modified file 'src/maasserver/api/rackcontrollers.py'
--- src/maasserver/api/rackcontrollers.py	2016-06-08 20:20:40 +0000
+++ src/maasserver/api/rackcontrollers.py	2016-10-12 17:51:07 +0000
@@ -9,6 +9,7 @@
 
 from django.conf import settings
 from django.http import HttpResponse
+from maasserver.api.interfaces import DISPLAYED_INTERFACE_FIELDS
 from maasserver.api.nodes import (
     NodeHandler,
     NodesHandler,
@@ -43,21 +44,7 @@
     'power_type',
     'power_state',
     'ip_addresses',
-    ('interface_set', (
-        'id',
-        'name',
-        'type',
-        'vlan',
-        'mac_address',
-        'parents',
-        'children',
-        'tags',
-        'enabled',
-        'links',
-        'params',
-        'discovered',
-        'effective_mtu',
-        )),
+    ('interface_set', DISPLAYED_INTERFACE_FIELDS),
     'zone',
     'status_action',
     'node_type',

=== modified file 'src/maasserver/api/regioncontrollers.py'
--- src/maasserver/api/regioncontrollers.py	2016-04-28 01:14:14 +0000
+++ src/maasserver/api/regioncontrollers.py	2016-10-12 17:51:07 +0000
@@ -6,6 +6,7 @@
     'RegionControllersHandler',
     ]
 
+from maasserver.api.interfaces import DISPLAYED_INTERFACE_FIELDS
 from maasserver.api.nodes import (
     NodeHandler,
     NodesHandler,
@@ -31,21 +32,7 @@
     'power_type',
     'power_state',
     'ip_addresses',
-    ('interface_set', (
-        'id',
-        'name',
-        'type',
-        'vlan',
-        'mac_address',
-        'parents',
-        'children',
-        'tags',
-        'enabled',
-        'links',
-        'params',
-        'discovered',
-        'effective_mtu',
-        )),
+    ('interface_set', DISPLAYED_INTERFACE_FIELDS),
     'zone',
     'status_action',
     'node_type',

=== modified file 'src/maasserver/api/tests/test_ipaddresses.py'
--- src/maasserver/api/tests/test_ipaddresses.py	2016-07-30 01:17:54 +0000
+++ src/maasserver/api/tests/test_ipaddresses.py	2016-10-12 17:51:07 +0000
@@ -2,6 +2,8 @@
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Tests for IP addresses API."""
+from testtools.matchers import HasLength
+
 
 __all__ = []
 
@@ -22,20 +24,342 @@
 from maasserver.testing.factory import factory
 from maasserver.utils.converters import json_load_bytes
 from maasserver.utils.orm import reload_object
+from maastesting.matchers import DocTestMatches
 from netaddr import IPAddress
 from testtools.matchers import Equals
 
 
-class TestIPAddressesAPI(APITestCase.ForUser):
+class TestIPAddressesAPI(APITestCase.ForUserAndAdmin):
+
+    def test_handler_path(self):
+        self.assertEqual(
+            '/api/2.0/ipaddresses/', reverse('ipaddresses_handler'))
+
+    def test_GET_returns_ipaddresses(self):
+        original_ipaddress = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        response = self.client.get(reverse('ipaddresses_handler'))
+        self.assertEqual(
+            http.client.OK, response.status_code, response.content)
+
+        parsed_result = json_load_bytes(response.content)
+        self.assertEqual(1, len(parsed_result), response.content)
+        [returned_address] = parsed_result
+        fields = {'alloc_type', 'alloc_type_name', 'ip'}
+        self.assertEqual(
+            fields.union({
+                'resource_uri',
+                'created',
+                'subnet',
+                'interface_set',
+                'owner'}),
+            set(returned_address.keys()))
+        expected_values = {
+            field: getattr(original_ipaddress, field)
+            for field in fields
+            if field not in ('resource_uri', 'created')
+        }
+        # Test that these exist, but not for exact values.
+        del returned_address['created']
+        del returned_address['subnet']
+        del returned_address['owner']
+        del returned_address['interface_set']
+        expected_values['resource_uri'] = reverse('ipaddresses_handler')
+        self.assertEqual(expected_values, returned_address)
+
+    def test_GET_returns_empty_if_no_ipaddresses(self):
+        response = self.client.get(reverse('ipaddresses_handler'))
+        self.assertEqual(
+            http.client.OK, response.status_code, response.content)
+        self.assertEqual([], json_load_bytes(response.content))
+
+    def test_GET_only_returns_request_users_addresses(self):
+        ipaddress = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=factory.make_User())
+        response = self.client.get(reverse('ipaddresses_handler'))
+        self.assertEqual(
+            http.client.OK, response.status_code, response.content)
+        parsed_result = json_load_bytes(response.content)
+        [returned_address] = parsed_result
+        self.assertEqual(ipaddress.ip, returned_address['ip'])
+
+    def test_GET_returns_all_addresses_if_admin_and_all_specified(self):
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=factory.make_User())
+        response = self.client.get(reverse('ipaddresses_handler'), {
+            "all": "1"
+        })
+        if self.user.is_superuser:
+            self.assertEqual(
+                http.client.OK, response.status_code, response.content)
+            parsed_result = json_load_bytes(response.content)
+            self.assertThat(parsed_result, HasLength(2))
+        else:
+            self.assertEqual(
+                http.client.FORBIDDEN, response.status_code, response.content)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                Equals("Listing all IP addresses requires admin privileges."))
+
+    def test_GET_returns_other_user_addresses_if_admin_and_user_specified(
+            self):
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        user2 = factory.make_User()
+        ipaddress2 = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=user2)
+        response = self.client.get(reverse('ipaddresses_handler'), {
+            "owner": user2.username
+        })
+        if self.user.is_superuser:
+            self.assertEqual(
+                http.client.OK, response.status_code, response.content)
+            parsed_result = json_load_bytes(response.content)
+            self.assertThat(parsed_result, HasLength(1))
+            self.assertEqual(ipaddress2.ip, parsed_result[0]['ip'])
+        else:
+            self.assertEqual(
+                http.client.FORBIDDEN, response.status_code, response.content)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                Equals(
+                    "Listing another user's IP addresses requires admin "
+                    "privileges."))
+
+    def test_GET_returns_other_users_ip_address_for_admin_with_all_with_ip(
+            self):
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        user2 = factory.make_User()
+        ipaddress2 = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=user2)
+        response = self.client.get(reverse('ipaddresses_handler'), {
+            "ip": str(ipaddress2.ip),
+            "all": "true"
+        })
+        if self.user.is_superuser:
+            self.assertEqual(
+                http.client.OK, response.status_code, response.content)
+            parsed_result = json_load_bytes(response.content)
+            self.assertThat(parsed_result, HasLength(1))
+            self.assertEqual(ipaddress2.ip, parsed_result[0]['ip'])
+        else:
+            self.assertEqual(
+                http.client.FORBIDDEN, response.status_code, response.content)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                Equals("Listing all IP addresses requires admin privileges."))
+
+    def test_GET_with_all_for_admin_returns_non_user_reserved_types(
+            self):
+        factory.make_StaticIPAddress(alloc_type=IPADDRESS_TYPE.STICKY)
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED)
+        response = self.client.get(reverse('ipaddresses_handler'), {
+            "all": "true"
+        })
+        if self.user.is_superuser:
+            self.assertEqual(
+                http.client.OK, response.status_code, response.content)
+            parsed_result = json_load_bytes(response.content)
+            self.assertThat(parsed_result, HasLength(2))
+        else:
+            self.assertEqual(
+                http.client.FORBIDDEN, response.status_code, response.content)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                Equals("Listing all IP addresses requires admin privileges."))
+
+    def test_GET_returns_own_ip_address_with_ip(self):
+        factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        ipaddress2 = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        response = self.client.get(reverse('ipaddresses_handler'), {
+            "ip": str(ipaddress2.ip),
+        })
+        self.assertEqual(
+            http.client.OK, response.status_code, response.content)
+        parsed_result = json_load_bytes(response.content)
+        self.assertThat(parsed_result, HasLength(1))
+        self.assertEqual(ipaddress2.ip, parsed_result[0]['ip'])
+
+    def test_GET_sorts_by_id(self):
+        addrs = []
+        for _ in range(3):
+            addrs.append(
+                factory.make_StaticIPAddress(
+                    alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user))
+        response = self.client.get(reverse('ipaddresses_handler'))
+        self.assertEqual(
+            http.client.OK, response.status_code, response.content)
+        parsed_result = json_load_bytes(response.content)
+        expected = [
+            addr.ip for addr in
+            sorted(addrs, key=lambda addr: getattr(addr, "id"))]
+        observed = [result['ip'] for result in parsed_result]
+        self.assertEqual(expected, observed)
+
+
+class TestIPAddressesReleaseAPI(APITestCase.ForUserAndAdmin):
+
+    scenarios = (
+        ("normal", {"force": None}),
+        ("without_force", {"force": False}),
+        ("with_force", {"force": True}),
+    )
+
+    @property
+    def force_should_work(self):
+        # The 'force' parameter should only work if (1) the user-under-test
+        # is a superuser, and (2) force=true was specified.
+        return self.user.is_superuser and self.force is True
+
+    @property
+    def expect_forbidden(self):
+        # The 'force' parameter should only work if (1) the user-under-test
+        # is a superuser, and (2) force=true was specified.
+        return not self.user.is_superuser and self.force is True
+
+    def expected_status(self, status):
+        # Non-administrators always get a FORBIDDEN (403) when requesting
+        # a forced delete.
+        if self.force and not self.user.is_superuser:
+            return http.client.FORBIDDEN
+        else:
+            return status
+
+    def post_release_request(self, ip, mac=None):
+        params = {
+            'op': 'release',
+            'ip': ip,
+        }
+        if mac is not None:
+            params["mac"] = mac
+        if self.force is not None:
+            params["force"] = str(self.force)
+        return self.client.post(reverse('ipaddresses_handler'), params)
+
+    def test_POST_release_rejects_invalid_ip(self):
+        response = self.post_release_request("1690.254.0.1")
+        expected_status = self.expected_status(http.client.BAD_REQUEST)
+        self.assertEqual(expected_status, response.status_code)
+        if expected_status != http.client.FORBIDDEN:
+            self.assertEqual(
+                dict(ip=["Enter a valid IPv4 or IPv6 address."]),
+                json_load_bytes(response.content))
+        else:
+            self.assertEqual(
+                "Force-releasing an IP address requires admin privileges.",
+                response.content.decode("utf-8"))
+
+    def test_POST_release_allows_admin_to_release_other_users_ip(self):
+        factory.make_StaticIPAddress(
+            user=factory.make_User(), alloc_type=IPADDRESS_TYPE.USER_RESERVED,
+            ip="192.168.0.1")
+        response = self.post_release_request("192.168.0.1")
+        if self.expect_forbidden:
+            self.assertEqual(http.client.FORBIDDEN, response.status_code)
+            self.assertEqual(
+                "Force-releasing an IP address requires admin privileges.",
+                response.content.decode("utf-8"))
+        elif not self.force:
+            self.assertEqual(http.client.BAD_REQUEST, response.status_code)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                DocTestMatches("...does not belong to the requesting user..."))
+        else:
+            self.assertEqual(http.client.NO_CONTENT, response.status_code)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                Equals(""))
+
+    def test_POST_release_allows_admin_to_release_sticky_ip(self):
+        # Make an "orphaned" IP address, like the one in bug #1630034.
+        static_ip = factory.make_StaticIPAddress(
+            user=factory.make_User(), alloc_type=IPADDRESS_TYPE.STICKY)
+        response = self.post_release_request(str(static_ip.ip))
+        if self.expect_forbidden:
+            self.assertEqual(http.client.FORBIDDEN, response.status_code)
+            self.assertEqual(
+                "Force-releasing an IP address requires admin privileges.",
+                response.content.decode("utf-8"))
+        elif not self.force:
+            self.assertEqual(http.client.BAD_REQUEST, response.status_code)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                DocTestMatches("...does not belong to the requesting user..."))
+        else:
+            self.assertEqual(http.client.NO_CONTENT, response.status_code)
+            self.assertThat(
+                response.content.decode("utf-8"),
+                Equals(""))
+
+    def test_POST_release_deallocates_address(self):
+        ipaddress = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        response = self.post_release_request(ipaddress.ip)
+        self.assertEqual(
+            self.expected_status(http.client.NO_CONTENT),
+            response.status_code, response.content)
+        if not self.expect_forbidden:
+            self.assertIsNone(reload_object(ipaddress))
+
+    def test_POST_release_deletes_unknown_interface(self):
+        subnet = factory.make_Subnet()
+        unknown_nic = factory.make_Interface(INTERFACE_TYPE.UNKNOWN)
+        ipaddress = unknown_nic.link_subnet(
+            INTERFACE_LINK_TYPE.STATIC, subnet,
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        response = self.post_release_request(ipaddress.ip)
+        self.assertEqual(
+            self.expected_status(http.client.NO_CONTENT),
+            response.status_code, response.content)
+        if not self.expect_forbidden:
+            self.assertIsNone(reload_object(unknown_nic))
+
+    def test_POST_release_does_not_delete_interfaces_linked_to_nodes(self):
+        node = factory.make_Node()
+        attached_nic = factory.make_Interface(node=node)
+        subnet = factory.make_Subnet()
+        ipaddress = attached_nic.link_subnet(
+            INTERFACE_LINK_TYPE.STATIC, subnet,
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+
+        self.post_release_request(ipaddress.ip)
+        self.assertEqual(attached_nic, reload_object(attached_nic))
+
+    def test_POST_release_does_not_delete_other_IPs_I_own(self):
+        ipaddress = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        other_address = factory.make_StaticIPAddress(
+            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
+        response = self.post_release_request(ipaddress.ip)
+        self.assertEqual(
+            self.expected_status(http.client.NO_CONTENT),
+            response.status_code, response.content)
+        self.assertIsNotNone(reload_object(other_address))
+
+
+class TestIPAddressesReserveAPI(APITestCase.ForUser):
+
+    scenarios = (
+        ("with_ip_param", {"ip_param": "ip"}),
+        ("with_ip_address_param", {"ip_param": "ip_address"}),
+    )
 
     def post_reservation_request(
-            self, subnet=None, ip_address=None, network=None,
-            mac=None, hostname=None):
+            self, subnet=None, ip_address=None, network=None, mac=None,
+            hostname=None):
         params = {
             'op': 'reserve',
         }
         if ip_address is not None:
-            params["ip_address"] = ip_address
+            params[self.ip_param] = ip_address
         if subnet is not None:
             params["subnet"] = subnet.cidr
         if network is not None and subnet is None:
@@ -46,40 +370,29 @@
             params["hostname"] = hostname
         return self.client.post(reverse('ipaddresses_handler'), params)
 
-    def post_release_request(self, ip, mac=None):
-        params = {
-            'op': 'release',
-            'ip': ip,
-        }
-        if mac is not None:
-            params["mac"] = mac
-        return self.client.post(reverse('ipaddresses_handler'), params)
-
     def assertNoMatchingNetworkError(self, response, net):
         self.assertEqual(
             http.client.BAD_REQUEST, response.status_code, response.content)
         expected = (
-            "Unable to identify subnet %s." % str(net))
+            "Unable to identify subnet: %s." % str(net))
         self.assertEqual(
             expected.encode(settings.DEFAULT_CHARSET),
             response.content)
 
-    def test_handler_path(self):
-        self.assertEqual(
-            '/api/2.0/ipaddresses/', reverse('ipaddresses_handler'))
-
     def test_POST_reserve_creates_ipaddress(self):
         subnet = factory.make_Subnet()
         response = self.post_reservation_request(subnet)
         self.assertEqual(http.client.OK, response.status_code)
         returned_address = json_load_bytes(response.content)
         [staticipaddress] = StaticIPAddress.objects.all()
-        # We don't need to test the value of the 'created' datetime
-        # field. By removing it, we also test for its presence.
+        # Test that these fields exist, but don't test for exact values.
         del returned_address['created']
         del returned_address['subnet']
+        del returned_address['interface_set']
+        del returned_address['owner']
         expected = dict(
             alloc_type=staticipaddress.alloc_type,
+            alloc_type_name=staticipaddress.alloc_type_name,
             ip=staticipaddress.ip,
             resource_uri=reverse('ipaddresses_handler'),
             )
@@ -102,8 +415,11 @@
         # field. By removing it, we also test for its presence.
         del returned_address['created']
         del returned_address['subnet']
+        del returned_address['interface_set']
+        del returned_address['owner']
         expected = dict(
             alloc_type=staticipaddress.alloc_type,
+            alloc_type_name=staticipaddress.alloc_type_name,
             ip=staticipaddress.ip,
             resource_uri=reverse('ipaddresses_handler'),
             )
@@ -281,112 +597,3 @@
         self.assertEqual(
             dict(ip_address=["Enter a valid IPv4 or IPv6 address."]),
             json_load_bytes(response.content))
-
-    def test_POST_release_rejects_invalid_ip(self):
-        response = self.post_release_request("1690.254.0.1")
-        self.assertEqual(http.client.BAD_REQUEST, response.status_code)
-        self.assertEqual(
-            dict(ip=["Enter a valid IPv4 or IPv6 address."]),
-            json_load_bytes(response.content))
-
-    def test_GET_returns_ipaddresses(self):
-        original_ipaddress = factory.make_StaticIPAddress(
-            user=self.user)
-        response = self.client.get(reverse('ipaddresses_handler'))
-        self.assertEqual(
-            http.client.OK, response.status_code, response.content)
-
-        parsed_result = json_load_bytes(response.content)
-        self.assertEqual(1, len(parsed_result), response.content)
-        [returned_address] = parsed_result
-        fields = {'alloc_type', 'ip'}
-        self.assertEqual(
-            fields.union({'resource_uri', 'created', 'subnet'}),
-            set(returned_address.keys()))
-        expected_values = {
-            field: getattr(original_ipaddress, field)
-            for field in fields
-            if field not in ('resource_uri', 'created')
-        }
-        # We don't need to test the value of the 'created' datetime
-        # field.
-        del returned_address['created']
-        del returned_address['subnet']
-        expected_values['resource_uri'] = reverse('ipaddresses_handler')
-        self.assertEqual(expected_values, returned_address)
-
-    def test_GET_returns_empty_if_no_ipaddresses(self):
-        response = self.client.get(reverse('ipaddresses_handler'))
-        self.assertEqual(
-            http.client.OK, response.status_code, response.content)
-        self.assertEqual([], json_load_bytes(response.content))
-
-    def test_GET_only_returns_request_users_addresses(self):
-        ipaddress = factory.make_StaticIPAddress(user=self.user)
-        factory.make_StaticIPAddress(user=factory.make_User())
-        response = self.client.get(reverse('ipaddresses_handler'))
-        self.assertEqual(
-            http.client.OK, response.status_code, response.content)
-        parsed_result = json_load_bytes(response.content)
-        [returned_address] = parsed_result
-        self.assertEqual(ipaddress.ip, returned_address['ip'])
-
-    def test_GET_sorts_by_id(self):
-        addrs = []
-        for _ in range(3):
-            addrs.append(
-                factory.make_StaticIPAddress(user=self.user))
-        response = self.client.get(reverse('ipaddresses_handler'))
-        self.assertEqual(
-            http.client.OK, response.status_code, response.content)
-        parsed_result = json_load_bytes(response.content)
-        expected = [
-            addr.ip for addr in
-            sorted(addrs, key=lambda addr: getattr(addr, "id"))]
-        observed = [result['ip'] for result in parsed_result]
-        self.assertEqual(expected, observed)
-
-    def test_POST_release_deallocates_address(self):
-        ipaddress = factory.make_StaticIPAddress(
-            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
-        response = self.post_release_request(ipaddress.ip)
-        self.assertEqual(
-            http.client.OK, response.status_code, response.content)
-        self.assertIsNone(reload_object(ipaddress))
-
-    def test_POST_release_deletes_unknown_interface(self):
-        subnet = factory.make_Subnet()
-        unknown_nic = factory.make_Interface(INTERFACE_TYPE.UNKNOWN)
-        ipaddress = unknown_nic.link_subnet(
-            INTERFACE_LINK_TYPE.STATIC, subnet,
-            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
-
-        self.post_release_request(ipaddress.ip)
-        self.assertIsNone(reload_object(unknown_nic))
-
-    def test_POST_release_does_not_delete_interfaces_linked_to_nodes(self):
-        node = factory.make_Node()
-        attached_nic = factory.make_Interface(node=node)
-        subnet = factory.make_Subnet()
-        ipaddress = attached_nic.link_subnet(
-            INTERFACE_LINK_TYPE.STATIC, subnet,
-            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
-
-        self.post_release_request(ipaddress.ip)
-        self.assertEqual(attached_nic, reload_object(attached_nic))
-
-    def test_POST_release_does_not_delete_IP_that_I_dont_own(self):
-        ipaddress = factory.make_StaticIPAddress(user=factory.make_User())
-        response = self.post_release_request(ipaddress.ip)
-        self.assertEqual(
-            http.client.BAD_REQUEST, response.status_code, response.content)
-
-    def test_POST_release_does_not_delete_other_IPs_I_own(self):
-        ipaddress = factory.make_StaticIPAddress(
-            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
-        other_address = factory.make_StaticIPAddress(
-            alloc_type=IPADDRESS_TYPE.USER_RESERVED, user=self.user)
-        response = self.post_release_request(ipaddress.ip)
-        self.assertEqual(
-            http.client.OK, response.status_code, response.content)
-        self.assertIsNotNone(reload_object(other_address))

=== modified file 'src/maasserver/models/staticipaddress.py'
--- src/maasserver/models/staticipaddress.py	2016-09-22 02:53:33 +0000
+++ src/maasserver/models/staticipaddress.py	2016-10-12 17:51:07 +0000
@@ -558,6 +558,11 @@
         strtype = type_names.get(self.alloc_type, '%s' % self.alloc_type)
         return "%s:type=%s" % (self.ip, strtype)
 
+    @property
+    def alloc_type_name(self):
+        """Returns a human-readable representation of the `alloc_type`."""
+        return IPADDRESS_TYPE_CHOICES_DICT.get(self.alloc_type, "")
+
     def get_node(self):
         """Return the Node of the first Interface connected to this IP
         address."""

=== modified file 'src/maasserver/models/tests/test_staticipaddress.py'
--- src/maasserver/models/tests/test_staticipaddress.py	2016-09-22 02:53:33 +0000
+++ src/maasserver/models/tests/test_staticipaddress.py	2016-10-12 17:51:07 +0000
@@ -20,6 +20,7 @@
     INTERFACE_TYPE,
     IPADDRESS_FAMILY,
     IPADDRESS_TYPE,
+    IPADDRESS_TYPE_CHOICES_DICT,
     IPRANGE_TYPE,
 )
 from maasserver.exceptions import (
@@ -1062,3 +1063,19 @@
         node_summary = json["node_summary"]
         self.expectThat(node_summary["system_id"], Equals(node.system_id))
         self.expectThat(node_summary["node_type"], Equals(node.node_type))
+
+
+class TestAllocTypeName(MAASServerTestCase):
+
+    def test__provides_human_readable_values_for_known_types(self):
+        ip = factory.make_StaticIPAddress()
+        self.assertThat(
+            ip.alloc_type_name,
+            Equals(IPADDRESS_TYPE_CHOICES_DICT[ip.alloc_type]))
+
+    def test__returns_empty_string_for_unknown_types(self):
+        ip = factory.make_StaticIPAddress()
+        ip.alloc_type = randint(2**16, 2**32)
+        self.assertThat(
+            ip.alloc_type_name,
+            Equals(""))