Remove logic for old LBaaS(v2)

There are some logic left which checks device owner values set by
LBaaS v1 or v2, but LBaaS was already retired in favor of Octavia.

Change-Id: Ieca7a29a45f0686babe68619ff9da133edd89dec
This commit is contained in:
Takashi Kajinami 2023-11-28 17:08:12 +09:00
parent 56172ed5ae
commit 656028f311
6 changed files with 4 additions and 57 deletions

View File

@ -196,9 +196,7 @@ def get_other_dvr_serviced_device_owners(host_dvr_for_dhcp=True):
prefix, not a complete device_owner name, so should be handled
separately (see is_dvr_serviced() below)
"""
device_owners = [n_const.DEVICE_OWNER_LOADBALANCER,
n_const.DEVICE_OWNER_LOADBALANCERV2,
trunk_constants.TRUNK_SUBPORT_OWNER]
device_owners = [trunk_constants.TRUNK_SUBPORT_OWNER]
if host_dvr_for_dhcp:
device_owners.append(n_const.DEVICE_OWNER_DHCP)
return device_owners
@ -213,10 +211,7 @@ def get_dvr_allowed_address_pair_device_owners():
Later if other device owners are used for allowed_address_pairs those
device_owners should be added to the list below.
"""
# TODO(Swami): Convert these methods to constants.
# Add the constants variable to the neutron-lib
return [n_const.DEVICE_OWNER_LOADBALANCER,
n_const.DEVICE_OWNER_LOADBALANCERV2]
return []
def is_dvr_serviced(device_owner):

View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
from unittest import mock
from neutron_lib import constants
@ -121,16 +120,6 @@ class FdbPopulationExtensionTestCase(base.BaseTestCase):
self.assertIn(updated_mac, updated_macs_for_device)
self.assertNotIn(mac, updated_macs_for_device)
def test_unpermitted_device_owner(self):
fdb_extension = self._get_fdb_extension()
details = copy.deepcopy(self.UPDATE_MSG)
details['device_owner'] = constants.DEVICE_OWNER_LOADBALANCER
fdb_extension.handle_port(context=None, details=details)
updated_macs_for_device = (
fdb_extension.fdb_tracker.device_to_macs.get(self.DEVICE))
mac = self.UPDATE_MSG['mac_address']
self.assertNotIn(mac, updated_macs_for_device)
def test_catch_init_exception(self):
self.mock_add.side_effect = netlink_exceptions.NetlinkError
fdb_extension = self._get_fdb_extension()

View File

@ -594,18 +594,7 @@ class TestDvrRouterOperations(base.BaseTestCase):
'subnet_id': subnet_id}],
'allowed_address_pairs': [
{'ip_address': '10.20.30.40',
'mac_address': '00:11:22:33:44:55'}]},
{'mac_address': '11:22:33:44:55:66',
'device_owner': lib_constants.DEVICE_OWNER_LOADBALANCER,
'fixed_ips': [{'ip_address': '1.2.3.5',
'prefixlen': 24,
'subnet_id': subnet_id}]},
{'mac_address': '22:33:44:55:66:77',
'device_owner':
lib_constants.DEVICE_OWNER_LOADBALANCERV2,
'fixed_ips': [{'ip_address': '1.2.3.6',
'prefixlen': 24,
'subnet_id': subnet_id}]}]
'mac_address': '00:11:22:33:44:55'}]}]
self.plugin_api.get_ports_by_subnet.return_value = test_ports

View File

@ -179,12 +179,6 @@ class TestDvrServices(base.BaseTestCase):
def _test_is_dvr_serviced(self, device_owner, expected):
self.assertEqual(expected, utils.is_dvr_serviced(device_owner))
def test_is_dvr_serviced_with_lb_port(self):
self._test_is_dvr_serviced(constants.DEVICE_OWNER_LOADBALANCER, True)
def test_is_dvr_serviced_with_lbv2_port(self):
self._test_is_dvr_serviced(constants.DEVICE_OWNER_LOADBALANCERV2, True)
def test_is_dvr_serviced_with_dhcp_port(self):
self._test_is_dvr_serviced(constants.DEVICE_OWNER_DHCP, True)
@ -197,12 +191,6 @@ class TestFipServices(base.BaseTestCase):
def _test_is_fip_serviced(self, device_owner, expected):
self.assertEqual(expected, utils.is_fip_serviced(device_owner))
def test_is_fip_serviced_with_lb_port(self):
self._test_is_fip_serviced(constants.DEVICE_OWNER_LOADBALANCER, True)
def test_is_fip_serviced_with_lbv2_port(self):
self._test_is_fip_serviced(constants.DEVICE_OWNER_LOADBALANCERV2, True)
def test_is_fip_serviced_with_dhcp_port(self):
self._test_is_fip_serviced(constants.DEVICE_OWNER_DHCP, False)

View File

@ -195,10 +195,6 @@ class DvrDbMixinTestCase(test_plugin.Ml2PluginV2TestCase):
device_owner=constants.DEVICE_OWNER_DHCP,
is_admin=True,
arg_list=arg_list, **host_arg) as dhcp_port,\
self.port(subnet=subnet,
device_owner=constants.DEVICE_OWNER_LOADBALANCER,
is_admin=True,
arg_list=arg_list, **host_arg) as lb_port,\
self.port(device_owner=constants.DEVICE_OWNER_COMPUTE_PREFIX,
is_admin=True,
arg_list=arg_list, **host_arg),\
@ -212,7 +208,7 @@ class DvrDbMixinTestCase(test_plugin.Ml2PluginV2TestCase):
is_admin=True,
arg_list=arg_list, **host_arg):
expected_ids = [port['port']['id'] for port in
[compute_port, dhcp_port, lb_port]]
[compute_port, dhcp_port]]
dvr_ports = self.mixin.get_ports_on_host_by_subnet(
self.ctx, HOST, subnet['subnet']['id'])
self.assertEqual(len(expected_ids), len(dvr_ports))

View File

@ -1366,16 +1366,6 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self.ctx, port)
self.assertEqual(3, l3_notify.del_arp_entry.call_count)
def test_update_arp_entry_for_dvr_service_port_added(self):
action = 'add'
device_owner = const.DEVICE_OWNER_LOADBALANCER
self._test_update_arp_entry_for_dvr_service_port(device_owner, action)
def test_update_arp_entry_for_dvr_service_port_deleted(self):
action = 'del'
device_owner = const.DEVICE_OWNER_LOADBALANCER
self._test_update_arp_entry_for_dvr_service_port(device_owner, action)
def test_add_router_interface_csnat_ports_failure(self):
router_dict = {'name': 'test_router', 'admin_state_up': True,
'distributed': True}