Allows manual reset of deallocate_at for admins

Done through PUT to /ip_addresses with "reset_allocation_time": true in the
body.

A few modifications were done to base_tests to support the admin context.

Had to prevent admin_roles from being loaded on context.

Removed the DEFAULT_ROUTE thing for great justice!

Non-admin access will raise HTTPForbidden.

Created _create_patch for the extra patching love.

RM6300

quark/plugin_modules/ip_addresses.py

@@ -13,6 +13,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import webob
+
 from neutron.common import exceptions
 from neutron.openstack.common import importutils
 from neutron.openstack.common import log as logging
@@ -92,6 +94,11 @@ def create_ip_address(context, ip_address):
     return v._make_ip_dict(address)
 
 
+def _get_deallocated_override():
+    """This function exists to mock and for future requirements if needed."""
+    return '2000-01-01 00:00:00'
+
+
 def update_ip_address(context, id, ip_address):
     LOG.info("update_ip_address %s for tenant %s" %
             (id, context.tenant_id))
@@ -104,6 +111,16 @@ def update_ip_address(context, id, ip_address):
             raise exceptions.NotFound(
                 message="No IP address found with id=%s" % id)
 
+        reset = ip_address['ip_address'].get('reset_allocation_time',
+                                             False)
+        if reset and address['deallocated'] == 1:
+            if context.is_admin:
+                LOG.info("IP's deallocated time being manually reset")
+                address['deallocated_at'] = _get_deallocated_override()
+            else:
+                msg = "Modification of reset_allocation_time requires admin"
+                raise webob.exc.HTTPForbidden(detail=msg)
+
         old_ports = address['ports']
         port_ids = ip_address['ip_address'].get('port_ids')
         if port_ids is None:

quark/plugin_views.py

@@ -31,7 +31,6 @@ from quark import utils
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 STRATEGY = network_strategy.STRATEGY
-DEFAULT_ROUTE = netaddr.IPNetwork("0.0.0.0/0")
 
 quark_view_opts = [
     cfg.BoolOpt('show_allocation_pools',

quark/tests/plugin_modules/test_ip_addresses.py

@@ -14,8 +14,10 @@
 #  under the License.
 
 import contextlib
+import webob
 
 import mock
+from mock import patch
 from neutron.common import exceptions
 
 from quark.db import models
@@ -149,6 +151,54 @@ class TestQuarkUpdateIPAddress(test_quark_plugin.TestQuarkPlugin):
                                                      ip_address)
             self.assertEqual(response['port_ids'], [port['id']])
 
+    def _create_patch(self, path):
+        patcher = patch(path)
+        mocked = patcher.start()
+        self.addCleanup(patcher.stop)
+        return mocked
+
+    def test_update_ip_address_update_deallocated_at(self):
+        port = dict(id=1, network_id=2, ip_addresses=[])
+        ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
+                  subnet_id=1, network_id=2, version=4, deallocated=1,
+                  deallocated_at='2020-01-01 00:00:00')
+
+        path = 'quark.plugin_modules.ip_addresses'
+        lookup = self._create_patch('%s._get_deallocated_override' % path)
+
+        with self._stubs(ports=[port], addr=ip):
+            ip_address = {'ip_address': {'reset_allocation_time': True}}
+            self.plugin.update_ip_address(self.admin_context, ip['id'],
+                                          ip_address)
+            self.assertTrue(lookup.called)
+
+    def test_update_ip_address_update_deallocated_at_not_deallocated(self):
+        port = dict(id=1, network_id=2, ip_addresses=[])
+        ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
+                  subnet_id=1, network_id=2, version=4, deallocated=0,
+                  deallocated_at='2020-01-01 00:00:00')
+
+        path = 'quark.plugin_modules.ip_addresses'
+        lookup = self._create_patch('%s._get_deallocated_override' % path)
+
+        with self._stubs(ports=[port], addr=ip):
+            ip_address = {'ip_address': {'reset_allocation_time': True}}
+            self.plugin.update_ip_address(self.admin_context, ip['id'],
+                                          ip_address)
+            self.assertFalse(lookup.called)
+
+    def test_update_ip_address_update_deallocated_at_not_admin(self):
+        port = dict(id=1, network_id=2, ip_addresses=[])
+        ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
+                  subnet_id=1, network_id=2, version=4, deallocated=1,
+                  deallocated_at='2020-01-01 00:00:00')
+
+        with self._stubs(ports=[port], addr=ip):
+            ip_address = {'ip_address': {'reset_allocation_time': True}}
+            with self.assertRaises(webob.exc.HTTPForbidden):
+                self.plugin.update_ip_address(self.context, ip['id'],
+                                              ip_address)
+
     def test_update_ip_address_no_ports(self):
         port = dict(id=1, network_id=2, ip_addresses=[])
         ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",

quark/tests/test_base.py

@@ -24,6 +24,8 @@ class TestBase(unittest2.TestCase):
     def setUp(self):
         super(TestBase, self).setUp()
         self.context = context.Context('fake', 'fake', is_admin=False)
+        self.admin_context = context.Context('fake', 'fake', is_admin=True,
+                                             load_admin_roles=False)
 
         class FakeContext(object):
             def __new__(cls, *args, **kwargs):