Merge "NSX|v update edge device when the user changes the port ip address"

vmware_nsx/plugins/nsx_v/drivers/abstract_router_driver.py

@@ -18,6 +18,9 @@ import six
 
 from neutron.db import l3_db
 from neutron.db import models_v2
+from vmware_nsx._i18n import _
+from vmware_nsx.common import exceptions as nsxv_exc
+from vmware_nsx.plugins.nsx_v.vshield import edge_utils
 
 
 @six.add_metaclass(abc.ABCMeta)
@@ -68,6 +71,7 @@ class RouterBaseDriver(RouterAbstractDriver):
         self.plugin = plugin
         self.nsx_v = plugin.nsx_v
         self.edge_manager = plugin.edge_manager
+        self.vcns = self.nsx_v.vcns
 
     def _get_external_network_id_by_router(self, context, router_id):
         """Get router's external network id if it has."""
@@ -79,4 +83,36 @@ class RouterBaseDriver(RouterAbstractDriver):
             id=router['gw_port_id']).all()
 
         if gw_ports:
-            return gw_ports[0]['network_id']
+            return gw_ports[0]['network_id']
+
+    def _get_edge_id_or_raise(self, context, router_id):
+        edge_id = edge_utils.get_router_edge_id(context, router_id)
+        if not edge_id:
+            error = (_("Failed to get router %(rid)s edge Id") %
+                     {'rid': router_id})
+            raise nsxv_exc.NsxPluginException(err_msg=error)
+        return edge_id
+
+    def update_nat_rules(self, context, router, router_id):
+        self.plugin._update_nat_rules(context, router, router_id)
+
+    def update_router_interface_ip(self, context, router_id, port_id,
+                                   int_net_id, old_ip, new_ip, subnet_mask):
+        """Update the fixed ip of a router interface.
+        This implementation will not work for distributed routers,
+        and there is a different implementation in that driver class
+        """
+        # get the edge-id of this router
+        edge_id = self._get_edge_id_or_raise(context, router_id)
+        # find out if the port is uplink or internal
+        router = self.plugin._get_router(context, router_id)
+        is_uplink = (port_id == router.gw_port_id)
+
+        # update the edge interface configuration
+        self.edge_manager.update_interface_addr(
+            context, edge_id, old_ip, new_ip,
+            subnet_mask, is_uplink=is_uplink)
+
+        # Also update the nat rules
+        if is_uplink:
+            self.update_nat_rules(context, router, router_id)

vmware_nsx/plugins/nsx_v/drivers/distributed_router_driver.py

@@ -462,3 +462,28 @@ class RouterDistributedDriver(router_driver.RouterBaseDriver):
         self.plugin._update_nat_rules(context, router, router_id=plr_id)
         self.plugin._update_subnets_and_dnat_firewall(context, router,
                                                       router_id=plr_id)
+
+    def update_router_interface_ip(self, context, router_id,
+                                   port_id, int_net_id,
+                                   old_ip, new_ip, subnet_mask):
+        """Update the fixed ip of a distributed router interface. """
+        router = self.plugin._get_router(context, router_id)
+        if port_id == router.gw_port_id:
+            # external port / Uplink
+            plr_id = self.edge_manager.get_plr_by_tlr_id(context, router_id)
+            edge_id = self._get_edge_id_or_raise(context, plr_id)
+            self.edge_manager.update_interface_addr(
+                context, edge_id, old_ip, new_ip, subnet_mask, is_uplink=True)
+            # Also update the nat rules
+            self.plugin._update_nat_rules(context, router, plr_id)
+        else:
+            # Internal port:
+            # get the edge-id of this router
+            edge_id = self._get_edge_id_or_raise(context, router_id)
+            # Get the vnic index
+            edge_vnic_binding = nsxv_db.get_edge_vnic_binding(
+                context.session, edge_id, int_net_id)
+            vnic_index = edge_vnic_binding.vnic_index
+            self.edge_manager.update_vdr_interface_addr(
+                context, edge_id, vnic_index, old_ip, new_ip,
+                subnet_mask)

vmware_nsx/plugins/nsx_v/drivers/shared_router_driver.py

@@ -157,6 +157,11 @@ class RouterSharedDriver(router_driver.RouterBaseDriver):
 
         return all_vnic_indices
 
+    def update_nat_rules(self, context, router, router_id):
+        router_ids = self.edge_manager.get_routers_on_same_edge(
+            context, router_id)
+        self._update_nat_rules_on_routers(context, router_id, router_ids)
+
     def _update_nat_rules_on_routers(self, context,
                                      target_router_id, router_ids):
         snats = []

vmware_nsx/plugins/nsx_v/plugin.py

@@ -1016,6 +1016,17 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         self._apply_dict_extend_functions('ports', port_data, port_model)
         return port_data
 
+    def _get_port_subnet_mask(self, context, port):
+        if len(port['fixed_ips']) > 0 and 'subnet_id' in port['fixed_ips'][0]:
+            subnet_id = port['fixed_ips'][0]['subnet_id']
+            subnet = self._get_subnet(context, subnet_id)
+            return str(netaddr.IPNetwork(subnet.cidr).netmask)
+
+    def _get_port_fixed_ip_addr(self, port):
+        if (len(port['fixed_ips']) > 0 and
+            'ip_address' in port['fixed_ips'][0]):
+            return port['fixed_ips'][0]['ip_address']
+
     def update_port(self, context, id, port):
         attrs = port[attr.PORT]
         port_data = port['port']
@@ -1027,6 +1038,14 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         has_port_security = (cfg.CONF.nsxv.spoofguard_enabled and
                              original_port[psec.PORTSECURITY])
 
+        port_ip_change = port_data.get('fixed_ips') is not None
+        device_owner_change = port_data.get('device_owner') is not None
+        # We do not support updating the port ip and device owner together
+        if port_ip_change and device_owner_change:
+            msg = (_('Cannot set fixed ips and device owner together for port '
+                     '%s') % original_port['id'])
+            raise n_exc.BadRequest(resource='port', msg=msg)
+
         # TODO(roeyc): create a method '_process_vnic_index_update' from the
         # following code block
         # Process update for vnic-index
@@ -1104,6 +1123,44 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         if comp_owner_update:
             # Create dhcp bindings, the port is now owned by an instance
             self._create_dhcp_static_binding(context, ret_port)
+        elif port_ip_change:
+            owner = original_port['device_owner']
+            # If port IP has changed we should update according to device
+            # owner
+            if is_compute_port:
+                # This is an instance port, so re-create DHCP entry
+                self._delete_dhcp_static_binding(context, original_port)
+                self._create_dhcp_static_binding(context, ret_port)
+            elif owner == constants.DEVICE_OWNER_DHCP:
+                # Update the ip of the dhcp port
+                address_groups = self._create_network_dhcp_address_group(
+                    context, ret_port['network_id'])
+                self._update_dhcp_edge_service(
+                    context, ret_port['network_id'], address_groups)
+            elif (owner == constants.DEVICE_OWNER_ROUTER_GW or
+                  owner == constants.DEVICE_OWNER_ROUTER_INTF):
+                # This is a router port - update the edge appliance
+                old_ip = self._get_port_fixed_ip_addr(original_port)
+                new_ip = self._get_port_fixed_ip_addr(ret_port)
+                if ((old_ip is not None or new_ip is not None) and
+                    (old_ip != new_ip)):
+                    if attr.is_attr_set(original_port.get('device_id')):
+                        router_id = original_port['device_id']
+                        router_driver = self._find_router_driver(context,
+                                                                 router_id)
+                        # subnet mask is needed for adding new ip to the vnic
+                        sub_mask = self._get_port_subnet_mask(context,
+                                                              ret_port)
+                        router_driver.update_router_interface_ip(
+                            context,
+                            router_id,
+                            original_port['id'],
+                            ret_port['network_id'],
+                            old_ip, new_ip, sub_mask)
+            else:
+                LOG.info(_LI('Not updating fixed IP on backend for '
+                             'device owner [%(dev_own)s] and port %(pid)s'),
+                         {'dev_own': owner, 'pid': original_port['id']})
 
         # Processing compute port update
         vnic_idx = original_port.get(ext_vnic_idx.VNIC_INDEX)

vmware_nsx/plugins/nsx_v/vshield/edge_utils.py

@@ -1136,6 +1136,116 @@ class EdgeManager(object):
         self.set_sysctl_rp_filter_for_vdr_dhcp(
             context, dhcp_edge_id, network_id)
 
+    def _update_address_in_dict(self, address_groups, old_ip, new_ip,
+                                subnet_mask):
+        """Update the address_groups data structure to replace the old ip
+        with a new one.
+        If the old ip is None - if the ip matches an existing subnet:
+                                add it as a secondary ip.
+                                else - add a new address group for the new ip
+        If the new ip is none - delete the primary/secondary entry with the
+                                old ip.
+        If the old ip was not found - return False
+        Otherwise - return True
+        """
+        if old_ip is None:
+            # Adding a new IP
+            # look for an address group with a primary ip in the same subnet
+            # as the new ip
+            for address_group in address_groups['addressGroups']:
+                if (netaddr.IPAddress(new_ip) in
+                    netaddr.IPNetwork(address_group['primaryAddress'] + '/' +
+                                      address_group['subnetPrefixLength'])):
+                    # we should add the new ip as a secondary address in this
+                    # address group
+                    if (address_group.get('secondaryAddresses') is not None):
+                        secondary = address_group['secondaryAddresses']
+                        secondary['ipAddress'].append(new_ip)
+                    else:
+                        address_group['secondaryAddresses'] = {
+                            'type': 'secondary_addresses',
+                            'ipAddress': [new_ip]}
+                    return True
+            # Could not find the same subnet - add a new address group
+            address_group = {
+                'primaryAddress': new_ip,
+                'subnetMask': subnet_mask
+            }
+            address_groups['addressGroups'].append(address_group)
+            return True
+        else:
+            for ind, address_group in enumerate(
+                address_groups['addressGroups']):
+                if address_group['primaryAddress'] == old_ip:
+                    # this is the one we should update
+                    if new_ip:
+                        address_group['primaryAddress'] = new_ip
+                    else:
+                        # delete this entry
+                        address_groups['addressGroups'].pop(ind)
+                    return True
+                # try to find a match in the secondary ips
+                if (address_group.get('secondaryAddresses') is not None):
+                    secondary = address_group['secondaryAddresses']
+                    secondary_ips = secondary['ipAddress']
+                    if old_ip in secondary_ips:
+                        # We should update the secondary addresses
+                        if new_ip:
+                            # replace the old with the new
+                            secondary_ips.remove(old_ip)
+                            secondary_ips.append(new_ip)
+                        else:
+                            # delete this entry
+                            if len(secondary_ips) == 1:
+                                # delete the whole structure
+                                del address_group['secondaryAddresses']
+                            else:
+                                secondary_ips.remove(old_ip)
+                        return True
+
+        # The old ip was not found
+        return False
+
+    def update_interface_addr(self, context, edge_id, old_ip, new_ip,
+                              subnet_mask, is_uplink=False):
+        with locking.LockManager.get_lock(edge_id):
+            # get the current interfaces configuration
+            r = self.nsxv_manager.vcns.get_interfaces(edge_id)[1]
+            vnics = r.get('vnics', [])
+            # Go over the vnics to find the one we should update
+            for vnic in vnics:
+                if ((is_uplink and vnic['type'] == 'uplink') or
+                    not is_uplink and vnic['type'] != 'uplink'):
+                    if self._update_address_in_dict(
+                        vnic['addressGroups'], old_ip, new_ip, subnet_mask):
+                        self.nsxv_manager.vcns.update_interface(edge_id, vnic)
+                        return
+
+        # If we got here - we didn't find the old ip:
+        error = (_("Failed to update interface ip "
+                   "on edge %(eid)s: Cannot find the previous ip %(ip)s") %
+                 {'eid': edge_id, 'ip': old_ip})
+        raise nsx_exc.NsxPluginException(err_msg=error)
+
+    def update_vdr_interface_addr(self, context, edge_id, vnic_index,
+                                  old_ip, new_ip, subnet_mask):
+        with locking.LockManager.get_lock(edge_id):
+            # get the current interfaces configuration
+            vnic = self.nsxv_manager.vcns.get_vdr_internal_interface(
+                edge_id, vnic_index)[1]
+            if self._update_address_in_dict(
+                vnic['addressGroups'], old_ip, new_ip, subnet_mask):
+                interface_req = {'interface': vnic}
+                self.nsxv_manager.vcns.update_vdr_internal_interface(
+                    edge_id, vnic_index, interface_req)
+                return
+
+        # If we got here - we didn't find the old ip:
+        error = (_("Failed to update VDR interface ip "
+                   "on edge %(eid)s: Cannot find the previous ip %(ip)s") %
+                 {'eid': edge_id, 'ip': old_ip})
+        raise nsx_exc.NsxPluginException(err_msg=error)
+
     def _get_sub_interface_id(self, context, edge_id, network_id):
         vnic_binding = nsxv_db.get_edge_vnic_binding(
             context.session, edge_id, network_id)

vmware_nsx/plugins/nsx_v/vshield/vcns.py

@@ -169,6 +169,10 @@ class Vcns(object):
         uri = "%s/%s/interfaces?action=patch" % (URI_PREFIX, edge_id)
         return self.do_request(HTTP_POST, uri, interface, decode=True)
 
+    def get_vdr_internal_interface(self, edge_id, interface_index):
+        uri = "%s/%s/interfaces/%s" % (URI_PREFIX, edge_id, interface_index)
+        return self.do_request(HTTP_GET, uri, decode=True)
+
     def update_vdr_internal_interface(self, edge_id,
                                       interface_index, interface):
         uri = "%s/%s/interfaces/%s" % (URI_PREFIX, edge_id, interface_index)

vmware_nsx/tests/unit/nsx_v/test_plugin.py

@@ -52,6 +52,8 @@ from vmware_nsx.extensions import routersize as router_size
 from vmware_nsx.extensions import routertype as router_type
 from vmware_nsx.extensions import securitygrouplogging
 from vmware_nsx.extensions import vnicindex as ext_vnic_idx
+from vmware_nsx.plugins.nsx_v.drivers import (
+    shared_router_driver as router_driver)
 from vmware_nsx.plugins.nsx_v.vshield.common import constants as vcns_const
 from vmware_nsx.plugins.nsx_v.vshield import edge_utils
 from vmware_nsx.tests import unit as vmware
@@ -1096,6 +1098,125 @@ class TestPortsV2(NsxVPluginV2TestCase,
                 self.assertEqual(ips[0]['ip_address'], '10.0.0.10')
                 self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
 
+    def test_update_port_update_ip_dhcp(self):
+        #Test updating a port IP when the device owner is DHCP
+        with self.subnet(enable_dhcp=False) as subnet:
+            with self.port(subnet=subnet,
+                           device_owner=constants.DEVICE_OWNER_DHCP) as port:
+                data = {'port': {'fixed_ips': [{'subnet_id':
+                                                subnet['subnet']['id'],
+                                                'ip_address': "10.0.0.10"}]}}
+                plugin = manager.NeutronManager.get_plugin()
+                ctx = context.get_admin_context()
+                with mock.patch.object(
+                    plugin.edge_manager,
+                    'update_dhcp_edge_service') as update_dhcp:
+                    plugin.update_port(ctx, port['port']['id'], data)
+                    self.assertTrue(update_dhcp.called)
+
+    def test_update_port_update_ip_compute(self):
+        #Test that updating a port IP succeed if the device owner starts
+        #with compute.
+        owner = constants.DEVICE_OWNER_COMPUTE_PREFIX + 'xxx'
+        with self.subnet(enable_dhcp=False) as subnet:
+            with self.port(subnet=subnet, device_id=_uuid(),
+                           device_owner=owner) as port:
+                data = {'port': {'fixed_ips': [{'subnet_id':
+                                                subnet['subnet']['id'],
+                                                'ip_address': "10.0.0.10"}]}}
+                plugin = manager.NeutronManager.get_plugin()
+                with mock.patch.object(
+                    plugin.edge_manager,
+                    'delete_dhcp_binding') as delete_dhcp:
+                    with mock.patch.object(
+                        plugin.edge_manager,
+                        'create_static_binding') as create_static:
+                        with mock.patch.object(
+                            plugin.edge_manager,
+                            'create_dhcp_bindings') as create_dhcp:
+                            plugin.update_port(context.get_admin_context(),
+                                               port['port']['id'], data)
+                            self.assertTrue(delete_dhcp.called)
+                            self.assertTrue(create_static.called)
+                            self.assertTrue(create_dhcp.called)
+
+    def test_update_port_update_ip_and_owner_fail(self):
+        #Test that updating a port IP and device owner at the same
+        #transaction fails
+        with self.subnet(enable_dhcp=False) as subnet:
+            with self.port(subnet=subnet,
+                           device_owner='aaa') as port:
+                data = {'port': {'device_owner': 'bbb',
+                                 'fixed_ips': [{'subnet_id':
+                                                subnet['subnet']['id'],
+                                                'ip_address': "10.0.0.10"}]}}
+                plugin = manager.NeutronManager.get_plugin()
+                self.assertRaises(n_exc.BadRequest,
+                                  plugin.update_port,
+                                  context.get_admin_context(),
+                                  port['port']['id'], data)
+
+    def test_update_port_update_ip_router(self):
+        #Test that updating a port IP succeed if the device owner is a router
+        owner = constants.DEVICE_OWNER_ROUTER_GW
+        router_id = _uuid()
+        old_ip = '10.0.0.3'
+        new_ip = '10.0.0.10'
+        with self.subnet(enable_dhcp=False) as subnet:
+            with self.port(subnet=subnet, device_id=router_id,
+                           device_owner=owner,
+                           fixed_ips=[{'ip_address': old_ip}]) as port:
+                data = {'port': {'fixed_ips': [{'subnet_id':
+                                                subnet['subnet']['id'],
+                                                'ip_address': new_ip}]}}
+                plugin = manager.NeutronManager.get_plugin()
+                ctx = context.get_admin_context()
+                router_obj = router_driver.RouterSharedDriver(plugin)
+                with mock.patch.object(plugin, '_find_router_driver',
+                                       return_value=router_obj):
+                    with mock.patch.object(
+                        router_obj,
+                        'update_router_interface_ip') as update_router:
+                        port_id = port['port']['id']
+                        plugin.update_port(ctx, port_id, data)
+                        net_id = port['port']['network_id']
+                        update_router.assert_called_once_with(
+                            ctx,
+                            router_id,
+                            port_id,
+                            net_id,
+                            old_ip,
+                            new_ip, "255.255.255.0")
+
+    def test_update_port_delete_ip_router(self):
+        #Test that deleting a port IP succeed if the device owner is a router
+        owner = constants.DEVICE_OWNER_ROUTER_GW
+        router_id = _uuid()
+        old_ip = '10.0.0.3'
+        with self.subnet(enable_dhcp=False) as subnet:
+            with self.port(subnet=subnet, device_id=router_id,
+                           device_owner=owner,
+                           fixed_ips=[{'ip_address': old_ip}]) as port:
+                data = {'port': {'fixed_ips': []}}
+                plugin = manager.NeutronManager.get_plugin()
+                ctx = context.get_admin_context()
+                router_obj = router_driver.RouterSharedDriver(plugin)
+                with mock.patch.object(plugin, '_find_router_driver',
+                                       return_value=router_obj):
+                    with mock.patch.object(
+                        router_obj,
+                        'update_router_interface_ip') as update_router:
+                        port_id = port['port']['id']
+                        plugin.update_port(ctx, port_id, data)
+                        net_id = port['port']['network_id']
+                        update_router.assert_called_once_with(
+                            ctx,
+                            router_id,
+                            port_id,
+                            net_id,
+                            old_ip,
+                            None, None)
+
     def test_update_port_update_ip_address_only(self):
         with self.subnet(enable_dhcp=False) as subnet:
             with self.port(subnet=subnet) as port:
@@ -1117,6 +1238,9 @@ class TestPortsV2(NsxVPluginV2TestCase,
                 self.assertEqual(ips[1]['ip_address'], '10.0.0.10')
                 self.assertEqual(ips[1]['subnet_id'], subnet['subnet']['id'])
 
+    def test_update_dhcp_port_with_exceeding_fixed_ips(self):
+        self.skipTest('Updating dhcp port IP is not supported')
+
     def test_requested_subnet_id_v4_and_v6_slaac(self):
         self.skipTest('No DHCP v6 Support yet')
 
@@ -2476,6 +2600,22 @@ class TestExclusiveRouterTestCase(L3NatTest, L3NatTestCaseBase,
                                               None,
                                               expected_code=expected_code)
 
+    @mock.patch.object(edge_utils.EdgeManager,
+                       'update_interface_addr')
+    def test_router_update_gateway_with_different_external_subnet(self, mock):
+        # This test calls the backend, so we need a mock for the edge_utils
+        super(
+            TestExclusiveRouterTestCase,
+            self).test_router_update_gateway_with_different_external_subnet()
+
+    @mock.patch.object(edge_utils.EdgeManager,
+                       'update_interface_addr')
+    def test_router_add_interface_multiple_ipv6_subnets_same_net(self, mock):
+        # This test calls the backend, so we need a mock for the edge_utils
+        super(
+            TestExclusiveRouterTestCase,
+            self).test_router_add_interface_multiple_ipv6_subnets_same_net()
+
 
 class ExtGwModeTestCase(NsxVPluginV2TestCase,
                         test_ext_gw_mode.ExtGwModeIntTestCase):
@@ -2670,6 +2810,14 @@ class TestVdrTestCase(L3NatTest, L3NatTestCaseBase,
                       IPv6ExpectedFailuresTestMixin,
                       NsxVPluginV2TestCase):
 
+    @mock.patch.object(edge_utils.EdgeManager,
+                       'update_interface_addr')
+    def test_router_update_gateway_with_different_external_subnet(self, mock):
+        # This test calls the backend, so we need a mock for the edge_utils
+        super(
+            TestVdrTestCase,
+            self).test_router_update_gateway_with_different_external_subnet()
+
     def test_floatingip_multi_external_one_internal(self):
         self.skipTest('skipped')
 
@@ -2935,6 +3083,12 @@ class TestSharedRouterTestCase(L3NatTest, L3NatTestCaseBase,
 
         return router_req.get_response(self.ext_api)
 
+    @mock.patch.object(edge_utils.EdgeManager,
+                       'update_interface_addr')
+    def test_router_add_interface_multiple_ipv6_subnets_same_net(self, mock):
+        super(TestSharedRouterTestCase,
+              self).test_router_add_interface_multiple_ipv6_subnets_same_net()
+
     def test_router_create_with_no_edge(self):
         name = 'router1'
         tenant_id = _uuid()

vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py

@@ -213,6 +213,16 @@ class FakeVcns(object):
         response = ''
         return (header, response)
 
+    def get_vdr_internal_interface(self, edge_id, interface_index):
+        response = {}
+        header = {
+            'status': 200
+        }
+        for interface in self._edges[edge_id].get('interfaces', []):
+            if int(interface['index']) == int(interface_index):
+                response = interface
+        return (header, response)
+
     def delete_vdr_internal_interface(self, edge_id, interface_index):
         for interface in self._edges[edge_id].get('interfaces', []):
             if int(interface['index']) == int(interface_index):

vmware_nsx/tests/unit/nsx_v/vshield/test_edge_utils.py

@@ -22,6 +22,7 @@ from neutron import context
 from neutron.plugins.common import constants as plugin_const
 from neutron.tests.unit import testlib_api
 from neutron_lib import exceptions as n_exc
+from vmware_nsx.common import exceptions as nsx_exc
 from vmware_nsx.common import nsxv_constants
 from vmware_nsx.db import nsxv_db
 from vmware_nsx.plugins.nsx_v.vshield.common import (
@@ -139,8 +140,48 @@ class EdgeDHCPManagerTestCase(EdgeUtilsTestCaseMixin):
 
 class EdgeUtilsTestCase(EdgeUtilsTestCaseMixin):
 
+    def setUp(self):
+        super(EdgeUtilsTestCase, self).setUp()
+        self.edge_manager = edge_utils.EdgeManager(self.nsxv_manager, None)
+
+        # Args for vcns interface configuration
+        self.internal_ip = '10.0.0.1'
+        self.uplink_ip = '192.168.111.30'
+        self.subnet_mask = '255.255.255.0'
+        self.pref_len = '24'
+        self.edge_id = 'dummy'
+        self.orig_vnics = ({},
+                           {'vnics': [
+                                {'addressGroups':
+                                    {'addressGroups': [
+                                        {'subnetMask': self.subnet_mask,
+                                         'subnetPrefixLength': self.pref_len,
+                                         'primaryAddress': self.uplink_ip}]},
+                                 'type': 'uplink',
+                                 'index': 1},
+                                {'addressGroups':
+                                    {'addressGroups': [
+                                        {'subnetMask': self.subnet_mask,
+                                         'subnetPrefixLength': self.pref_len,
+                                         'primaryAddress': self.internal_ip}]},
+                                 'type': 'internal',
+                                 'index': 2}]}
+                           )
+
+        # Args for vcns vdr interface configuration
+        self.vdr_ip = '10.0.0.1'
+        self.vnic = 1
+        self.orig_vdr = ({},
+                         {'index': 2,
+                          'addressGroups': {'addressGroups':
+                                [{'subnetMask': self.subnet_mask,
+                                  'subnetPrefixLength': self.pref_len,
+                                  'primaryAddress': self.vdr_ip}]},
+                          'type': 'internal'})
+
     def test_create_lrouter(self):
         lrouter = self._create_router()
+        self.nsxv_manager.deploy_edge.reset_mock()
         edge_utils.create_lrouter(self.nsxv_manager, self.ctx, lrouter,
                                   lswitch=None, dist=False)
         self.nsxv_manager.deploy_edge.assert_called_once_with(
@@ -152,6 +193,116 @@ class EdgeUtilsTestCase(EdgeUtilsTestCaseMixin):
                      'context': self.ctx},
             appliance_size=vcns_const.SERVICE_SIZE_MAPPING['router'])
 
+    def _test_update_intereface_primary_addr(self, old_ip, new_ip, isUplink):
+        fixed_vnic = {'addressGroups':
+                      {'addressGroups': [
+                              {'subnetMask': self.subnet_mask,
+                               'subnetPrefixLength': self.pref_len,
+                               'primaryAddress': new_ip}] if new_ip else []},
+                      'type': 'uplink' if isUplink else 'internal',
+                      'index': 1 if isUplink else 2}
+
+        with mock.patch.object(self.nsxv_manager.vcns,
+            'get_interfaces', return_value=self.orig_vnics):
+            self.edge_manager.update_interface_addr(
+                self.ctx, self.edge_id, old_ip, new_ip,
+                self.subnet_mask, is_uplink=isUplink)
+            self.nsxv_manager.vcns.update_interface.assert_called_once_with(
+                self.edge_id, fixed_vnic)
+
+    def test_update_interface_addr_intrernal(self):
+        self._test_update_intereface_primary_addr(
+            self.internal_ip, '10.0.0.2', False)
+
+    def test_remove_interface_primary_addr_intrernal(self):
+        self._test_update_intereface_primary_addr(
+            self.internal_ip, None, False)
+
+    def test_update_interface_addr_uplink(self):
+        self._test_update_intereface_primary_addr(
+            self.uplink_ip, '192.168.111.31', True)
+
+    def test_remove_interface_primary_addr_uplink(self):
+        self._test_update_intereface_primary_addr(
+            self.uplink_ip, None, True)
+
+    def _test_update_intereface_secondary_addr(self, old_ip, new_ip):
+        addr_group = {'subnetMask': self.subnet_mask,
+                      'subnetPrefixLength': self.pref_len,
+                      'primaryAddress': self.uplink_ip,
+                      'secondaryAddresses': {'type': 'secondary_addresses',
+                                             'ipAddress': [new_ip]}}
+        fixed_vnic = {'addressGroups': {'addressGroups': [addr_group]},
+                      'type': 'uplink',
+                      'index': 1}
+
+        with mock.patch.object(self.nsxv_manager.vcns,
+            'get_interfaces', return_value=self.orig_vnics):
+            self.edge_manager.update_interface_addr(
+                self.ctx, self.edge_id, old_ip, new_ip,
+                self.subnet_mask, is_uplink=True)
+            self.nsxv_manager.vcns.update_interface.assert_called_once_with(
+                self.edge_id, fixed_vnic)
+
+    def test_add_secondary_interface_addr(self):
+        self._test_update_intereface_secondary_addr(
+            None, '192.168.111.31')
+
+    def test_update_interface_addr_fail(self):
+        # Old ip is not configured on the interface, so we should fail
+        old_ip = '192.168.111.32'
+        new_ip = '192.168.111.31'
+
+        with mock.patch.object(self.nsxv_manager.vcns,
+            'get_interfaces', return_value=self.orig_vnics):
+            self.assertRaises(
+                nsx_exc.NsxPluginException,
+                self.edge_manager.update_interface_addr,
+                self.ctx, self.edge_id, old_ip, new_ip,
+                self.subnet_mask, is_uplink=True)
+
+    def _test_update_vdr_intereface_primary_addr(self, old_ip,
+                                                 new_ip):
+        fixed_vnic = {'addressGroups':
+                      {'addressGroups': [
+                              {'subnetMask': self.subnet_mask,
+                               'subnetPrefixLength': self.pref_len,
+                               'primaryAddress': new_ip}] if new_ip else []},
+                      'type': 'internal',
+                      'index': 2}
+
+        with mock.patch.object(self.nsxv_manager.vcns,
+            'get_vdr_internal_interface', return_value=self.orig_vdr):
+            with mock.patch.object(self.nsxv_manager.vcns,
+                'update_vdr_internal_interface') as vcns_update:
+                self.edge_manager.update_vdr_interface_addr(
+                    self.ctx, self.edge_id, self.vnic, old_ip, new_ip,
+                    self.subnet_mask)
+                vcns_update.assert_called_once_with(self.edge_id,
+                                                    self.vnic,
+                                                    {'interface': fixed_vnic})
+
+    def test_update_vdr_interface_addr_intrernal(self):
+        self._test_update_vdr_intereface_primary_addr(
+            self.vdr_ip, '20.0.0.2')
+
+    def test_remove_vdr_interface_primary_addr_intrernal(self):
+        self._test_update_vdr_intereface_primary_addr(
+            self.vdr_ip, None)
+
+    def test_update_vdr_interface_addr_fail(self):
+        # Old ip is not configured on the vdr interface, so we should fail
+        old_ip = '192.168.111.32'
+        new_ip = '192.168.111.31'
+
+        with mock.patch.object(self.nsxv_manager.vcns,
+            'get_vdr_internal_interface', return_value=self.orig_vdr):
+            self.assertRaises(
+                nsx_exc.NsxPluginException,
+                self.edge_manager.update_vdr_interface_addr,
+                self.ctx, self.edge_id, self.vnic, old_ip, new_ip,
+                self.subnet_mask)
+
 
 class EdgeManagerTestCase(EdgeUtilsTestCaseMixin):