openstack/kuryr-kubernetes
Code Issues Proposed changes

Refactor of os_vif_util module.

Browse Source 
Besides refactor itself, there was several issues resolved, like
appropriate data handling - because macvlan driver is still using
neutron client, we have to deal with dict data and openstacksdk objects
at the same time in couple of helper methods in os_vif_util module.

Furthermore, helper method for creating port object for test purposes
was moved to fake module, since it is also utilized in test for os vif
utils.

Change-Id: Iefc758ac22e1688ca9d017d5a4c1d32c4cf0583a
Implements: blueprint switch-to-openstacksdk
This commit is contained in:
Roman Dobosz 2019-12-09 15:22:19 +01:00
parent 3d32182920
commit e54ff6221c
7 changed files with 240 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
kuryr_kubernetes/controller/drivers/utils.py
View File

@ -23,7 +23,6 @@ from six.moves.urllib import parse
from kuryr_kubernetes import clients
from kuryr_kubernetes import constants
from kuryr_kubernetes import exceptions as k_exc
from kuryr_kubernetes import os_vif_util as ovu
from kuryr_kubernetes import utils
from neutronclient.common import exceptions as n_exc
@ -51,7 +50,7 @@ cache.configure_cache_region(CONF, pod_ip_cache_region)
def get_network_id(subnets):
ids = ovu.osvif_to_neutron_network_ids(subnets)
ids = list(set(net.id for net in subnets.values()))
if len(ids) != 1:
raise k_exc.IntegrityError(

146
kuryr_kubernetes/os_vif_util.py
View File

@ -120,14 +120,21 @@ def _make_vif_subnets(neutron_port, subnets):
"""Gets a list of os-vif Subnet objects for port.
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
neutron client's 'show_port' or
openstack.network.v2.port.Port object
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:return: list of os-vif Subnet object
"""
vif_subnets = {}
try:
fixed_ips = neutron_port.get('fixed_ips', [])
port_id = neutron_port.get('id')
except TypeError:
fixed_ips = neutron_port.fixed_ips
port_id = neutron_port.get.id
for neutron_fixed_ip in neutron_port.get('fixed_ips', []):
for neutron_fixed_ip in fixed_ips:
subnet_id = neutron_fixed_ip['subnet_id']
ip_address = neutron_fixed_ip['ip_address']
@ -145,7 +152,7 @@ def _make_vif_subnets(neutron_port, subnets):
if not vif_subnets:
raise k_exc.IntegrityError(_(
"No valid subnets found for port %(port_id)s") % {
'port_id': neutron_port.get('id')})
'port_id': port_id})
return list(vif_subnets.values())
@ -154,20 +161,30 @@ def _make_vif_network(neutron_port, subnets):
"""Get an os-vif Network object for port.
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
neutron client's 'show_port', or
openstack.network.v2.port.Port object
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:return: os-vif Network object
"""
# NOTE(gryf): Because we didn't convert macvlan driver, neutron_port can
# be either a dict or an object
try:
network_id = neutron_port.get('network_id')
port_id = neutron_port.get('id')
except TypeError:
network_id = neutron_port.network_id
port_id = neutron_port.id
try:
network = next(net.obj_clone() for net in subnets.values()
if net.id == neutron_port.get('network_id'))
if net.id == network_id)
except StopIteration:
raise k_exc.IntegrityError(_(
"Port %(port_id)s belongs to network %(network_id)s, "
"but requested networks are: %(requested_networks)s") % {
'port_id': neutron_port.get('id'),
'network_id': neutron_port.get('network_id'),
'port_id': port_id,
'network_id': network_id,
'requested_networks': [net.id for net in subnets.values()]})
network.subnets = osv_subnet.SubnetList(
@ -180,103 +197,115 @@ def _get_vif_name(neutron_port):
"""Gets a VIF device name for port.
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
neutron client's 'show_port', or an port object
returned by openstack client.
"""
vif_name, _ = kl_utils.get_veth_pair_names(neutron_port['id'])
try:
port_id = neutron_port['id']
except TypeError:
port_id = neutron_port.id
vif_name, _ = kl_utils.get_veth_pair_names(port_id)
return vif_name
def _get_ovs_hybrid_bridge_name(neutron_port):
def _get_ovs_hybrid_bridge_name(os_port):
"""Gets a name of the Linux bridge name for hybrid OpenVSwitch port.
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
:param os_port: openstack.network.v2.port.Port object
"""
return ('qbr' + neutron_port['id'])[:kl_const.NIC_NAME_LEN]
return ('qbr' + os_port.id)[:kl_const.NIC_NAME_LEN]
def _is_port_active(neutron_port):
"""Checks if port is active.
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
neutron client's 'show_port' or
openstack.network.v2.port.Port object
"""
return (neutron_port['status'] == kl_const.PORT_STATUS_ACTIVE)
try:
return (neutron_port['status'] == kl_const.PORT_STATUS_ACTIVE)
except TypeError:
return (neutron_port.status == kl_const.PORT_STATUS_ACTIVE)
def neutron_to_osvif_vif_ovs(vif_plugin, neutron_port, subnets):
def neutron_to_osvif_vif_ovs(vif_plugin, os_port, subnets):
"""Converts Neutron port to VIF object for os-vif 'ovs' plugin.
:param vif_plugin: name of the os-vif plugin to use (i.e. 'ovs')
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
:param os_port: openstack.network.v2.port.Port object
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:return: os-vif VIF object
"""
# TODO(gryf): check where it is used and what exactly is neutron_port now.
profile = osv_vif.VIFPortProfileOpenVSwitch(
interface_id=neutron_port['id'])
try:
# TODO(gryf): get rid of the except part after neutron_vif plugin is
# migrated to the openstacksdk.
port_id = os_port.id
mac_address = os_port.mac_address
details = os_port.binding_vif_details or {}
except AttributeError:
port_id = os_port['id']
mac_address = os_port['mac_address']
details = os_port.get('binding:vif_details', {})
profile = osv_vif.VIFPortProfileOpenVSwitch(interface_id=port_id)
details = neutron_port.get('binding:vif_details', {})
ovs_bridge = details.get('bridge_name',
config.CONF.neutron_defaults.ovs_bridge)
if not ovs_bridge:
raise oslo_cfg.RequiredOptError('ovs_bridge', 'neutron_defaults')
network = _make_vif_network(neutron_port, subnets)
network = _make_vif_network(os_port, subnets)
network.bridge = ovs_bridge
if details.get('ovs_hybrid_plug'):
vif = osv_vif.VIFBridge(
id=neutron_port['id'],
address=neutron_port['mac_address'],
id=port_id,
address=mac_address,
network=network,
has_traffic_filtering=details.get('port_filter', False),
preserve_on_delete=False,
active=_is_port_active(neutron_port),
active=_is_port_active(os_port),
port_profile=profile,
plugin=vif_plugin,
vif_name=_get_vif_name(neutron_port),
bridge_name=_get_ovs_hybrid_bridge_name(neutron_port))
vif_name=_get_vif_name(os_port),
bridge_name=_get_ovs_hybrid_bridge_name(os_port))
else:
vif = osv_vif.VIFOpenVSwitch(
id=neutron_port['id'],
address=neutron_port['mac_address'],
id=port_id,
address=mac_address,
network=network,
has_traffic_filtering=details.get('port_filter', False),
preserve_on_delete=False,
active=_is_port_active(neutron_port),
active=_is_port_active(os_port),
port_profile=profile,
plugin=vif_plugin,
vif_name=_get_vif_name(neutron_port),
vif_name=_get_vif_name(os_port),
bridge_name=network.bridge)
return vif
def neutron_to_osvif_vif_nested_vlan(neutron_port, subnets, vlan_id):
def neutron_to_osvif_vif_nested_vlan(os_port, subnets, vlan_id):
"""Converts Neutron port to VIF object for VLAN nested containers.
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
:param os_port: openstack.network.v2.port.Port object
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:param vlan_id: VLAN id associated to the VIF object for the pod
:return: kuryr-k8s native VIF object for VLAN nested
"""
# TODO(gryf): same as above
details = neutron_port.get('binding:vif_details', {})
details = os_port.binding_vif_details or {}
return k_vif.VIFVlanNested(
id=neutron_port['id'],
address=neutron_port['mac_address'],
network=_make_vif_network(neutron_port, subnets),
id=os_port.id,
address=os_port.mac_address,
network=_make_vif_network(os_port, subnets),
has_traffic_filtering=details.get('port_filter', False),
preserve_on_delete=False,
active=_is_port_active(neutron_port),
active=_is_port_active(os_port),
plugin=const.K8S_OS_VIF_NOOP_PLUGIN,
vif_name=_get_vif_name(neutron_port),
vif_name=_get_vif_name(os_port),
vlan_id=vlan_id)
@ -288,7 +317,6 @@ def neutron_to_osvif_vif_nested_macvlan(neutron_port, subnets):
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:return: kuryr-k8s native VIF object for MACVLAN nested
"""
# TODO(gryf): same as above
details = neutron_port.get('binding:vif_details', {})
return k_vif.VIFMacvlanNested(
@ -302,46 +330,42 @@ def neutron_to_osvif_vif_nested_macvlan(neutron_port, subnets):
vif_name=_get_vif_name(neutron_port))
def neutron_to_osvif_vif_sriov(vif_plugin, neutron_port, subnets):
def neutron_to_osvif_vif_sriov(vif_plugin, os_port, subnets):
"""Converts Neutron port to VIF object for SRIOV containers.
:param vif_plugin: name of the os-vif plugin to use (i.e. 'noop')
:param neutron_port: dict containing port information as returned by
neutron client's 'show_port'
:param os_port: openstack.network.v2.port.Port object
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:return: osv_vif VIFSriov object
"""
# TODO(gryf): same as above
details = neutron_port.get('binding:vif_details', {})
network = _make_vif_network(neutron_port, subnets)
details = os_port.binding_vif_details or {}
network = _make_vif_network(os_port, subnets)
vlan_name = network.vlan if network.should_provide_vlan else ''
vif = k_vif.VIFSriov(
id=neutron_port['id'],
address=neutron_port['mac_address'],
id=os_port.id,
address=os_port.mac_address,
network=network,
has_traffic_filtering=details.get('port_filter', False),
preserve_on_delete=False,
active=_is_port_active(neutron_port),
active=_is_port_active(os_port),
plugin=vif_plugin,
mode='passthrough',
vlan_name=vlan_name,
vif_name=_get_vif_name(neutron_port),
vif_name=_get_vif_name(os_port),
)
return vif
def neutron_to_osvif_vif(vif_translator, neutron_port, subnets):
def neutron_to_osvif_vif(vif_translator, os_port, subnets):
"""Converts Neutron port to os-vif VIF object.
:param vif_translator: name of the traslator for the os-vif plugin to use
:param neutron_port: dict containing port information as returned by
neutron client
:param os_port: openstack.network.v2.port.Port object
:param subnets: subnet mapping as returned by PodSubnetsDriver.get_subnets
:return: os-vif VIF object
"""
# TODO(gryf): same as above
try:
mgr = _VIF_MANAGERS[vif_translator]
@ -351,7 +375,7 @@ def neutron_to_osvif_vif(vif_translator, neutron_port, subnets):
name=vif_translator, invoke_on_load=False)
_VIF_MANAGERS[vif_translator] = mgr
return mgr.driver(vif_translator, neutron_port, subnets)
return mgr.driver(vif_translator, os_port, subnets)
def osvif_to_neutron_fixed_ips(subnets):
@ -376,7 +400,3 @@ def osvif_to_neutron_fixed_ips(subnets):
fixed_ips.append({'subnet_id': subnet_id})
return fixed_ips
def osvif_to_neutron_network_ids(subnets):
return list(set(net.id for net in subnets.values()))

56
kuryr_kubernetes/tests/fake.py
View File

@ -15,10 +15,13 @@
import uuid
from openstack.network.v2 import port as os_port
from os_vif import objects as osv_objects
from os_vif.objects import vif as osv_vif
from oslo_serialization import jsonutils
from kuryr_kubernetes import constants
def _fake_vif(cls=osv_vif.VIFOpenVSwitch):
vif = cls(
@ -83,3 +86,56 @@ def _fake_vifs_string(dictionary=None):
return jsonutils.dumps(dictionary)
else:
return jsonutils.dumps(_fake_vifs_dict())
def get_port_obj(port_id='07cfe856-11cc-43d9-9200-ff4dc02d3620',
device_owner='compute:kuryr', ip_address=None,
vif_details=None):
fixed_ips = [{'subnet_id': 'e1942bb1-5f51-4646-9885-365b66215592',
'ip_address': '10.10.0.5'},
{'subnet_id': '4894baaf-df06-4a54-9885-9cd99d1cc245',
'ip_address': 'fd35:7db5:e3fc:0:f816:3eff:fe80:d421'}]
if ip_address:
fixed_ips[0]['ip_address'] = ip_address
security_group_ids = ['cfb3dfc4-7a43-4ba1-b92d-b8b2650d7f88']
if not vif_details:
vif_details = {'port_filter': True, 'ovs_hybrid_plug': False}
port_data = {'allowed_address_pairs': [],
'binding_host_id': 'kuryr-devstack',
'binding_profile': {},
'binding_vif_details': vif_details,
'binding_vif_type': 'ovs',
'binding_vnic_type': 'normal',
'created_at': '2017-06-09T13:23:24Z',
'data_plane_status': None,
'description': '',
'device_id': '',
'device_owner': device_owner,
'dns_assignment': None,
'dns_domain': None,
'dns_name': None,
'extra_dhcp_opts': [],
'fixed_ips': fixed_ips,
'id': port_id,
'ip_address': None,
'is_admin_state_up': True,
'is_port_security_enabled': True,
'location': None,
'mac_address': 'fa:16:3e:80:d4:21',
'name': constants.KURYR_PORT_NAME,
'network_id': 'ba44f957-c467-412b-b985-ae720514bc46',
'option_name': None,
'option_value': None,
'project_id': 'b6e8fb2bde594673923afc19cf168f3a',
'qos_policy_id': None,
'revision_number': 9,
'security_group_ids': security_group_ids,
'status': u'DOWN',
'subnet_id': None,
'tags': [],
'trunk_details': None,
'updated_at': u'2019-12-04T15:06:09Z'}
return os_port.Port(**port_data)

16
kuryr_kubernetes/tests/unit/controller/drivers/test_neutron_vif.py
View File

@ -326,22 +326,6 @@ class NeutronPodVIFDriver(test_base.TestCase):
self.assertEqual(vif_plugin, cls._get_vif_plugin(m_driver, port))
@mock.patch('kuryr_kubernetes.os_vif_util.osvif_to_neutron_network_ids')
def test_get_network_id(self, m_to_net_ids):
subnets = mock.sentinel.subnets
network_id = mock.sentinel.network_id
m_to_net_ids.return_value = [network_id]
self.assertEqual(network_id, utils.get_network_id(subnets))
m_to_net_ids.assert_called_once_with(subnets)
@mock.patch('kuryr_kubernetes.os_vif_util.osvif_to_neutron_network_ids')
def test_get_network_id_invalid(self, m_to_net_ids):
subnets = mock.sentinel.subnets
m_to_net_ids.return_value = []
self.assertRaises(k_exc.IntegrityError, utils.get_network_id, subnets)
def test_get_port_name(self):
pod_name = mock.sentinel.pod_name
port_name = 'default/' + str(pod_name)

29
kuryr_kubernetes/tests/unit/controller/drivers/test_utils.py
View File

@ -34,3 +34,32 @@ class TestUtils(test_base.TestCase):
self.assertIsNone(resp)
kubernetes.get.assert_called_once_with('{}/namespaces/{}'.format(
constants.K8S_API_BASE, namespace_name))
def test_get_network_id(self):
id_a = mock.sentinel.id_a
net1 = mock.Mock()
net1.id = id_a
net2 = mock.Mock()
net2.id = id_a
subnets = {1: net1, 2: net2}
ret = utils.get_network_id(subnets)
self.assertEqual(ret, id_a)
def test_get_network_id_invalid(self):
id_a = mock.sentinel.id_a
id_b = mock.sentinel.id_b
net1 = mock.Mock()
net1.id = id_a
net2 = mock.Mock()
net2.id = id_b
net3 = mock.Mock()
net3.id = id_a
subnets = {1: net1, 2: net2, 3: net3}
self.assertRaises(exceptions.IntegrityError, utils.get_network_id,
subnets)
def test_get_network_id_empty(self):
self.assertRaises(exceptions.IntegrityError, utils.get_network_id, {})

82
kuryr_kubernetes/tests/unit/controller/drivers/test_vif_pool.py
View File

@ -32,6 +32,7 @@ from kuryr_kubernetes import exceptions
from kuryr_kubernetes.objects import vif
from kuryr_kubernetes import os_vif_util as ovu
from kuryr_kubernetes.tests import base as test_base
from kuryr_kubernetes.tests import fake
from kuryr_kubernetes.tests.unit import kuryr_fixtures as k_fix
@ -67,8 +68,8 @@ def get_pod_name(pod):
return "%(namespace)s/%(name)s" % pod['metadata']
def get_port_obj(port_id=None, device_owner=None, ip_address=None):
port_obj = munch.Munch({
def get_neutron_port(port_id=None, device_owner=None, ip_address=None):
port_obj = {
'allowed_address_pairs': [],
'extra_dhcp_opts': [],
'device_owner': 'compute:kuryr',
@ -76,17 +77,14 @@ def get_port_obj(port_id=None, device_owner=None, ip_address=None):
'port_security_enabled': True,
'binding:profile': {},
'fixed_ips': [
munch.Munch({
'subnet_id': 'e1942bb1-5f51-4646-9885-365b66215592',
'ip_address': '10.10.0.5'}),
munch.Munch({
'subnet_id': '4894baaf-df06-4a54-9885-9cd99d1cc245',
'ip_address': 'fd35:7db5:e3fc:0:f816:3eff:fe80:d421'})],
{'subnet_id': 'e1942bb1-5f51-4646-9885-365b66215592',
'ip_address': '10.10.0.5'},
{'subnet_id': '4894baaf-df06-4a54-9885-9cd99d1cc245',
'ip_address': 'fd35:7db5:e3fc:0:f816:3eff:fe80:d421'}],
'id': '07cfe856-11cc-43d9-9200-ff4dc02d3620',
'security_groups': ['cfb3dfc4-7a43-4ba1-b92d-b8b2650d7f88'],
'binding:vif_details': munch.Munch({
'port_filter': True,
'ovs_hybrid_plug': False}),
'binding:vif_details': {'port_filter': True,
'ovs_hybrid_plug': False},
'binding:vif_type': 'ovs',
'mac_address': 'fa:16:3e:80:d4:21',
'project_id': 'b6e8fb2bde594673923afc19cf168f3a',
@ -100,14 +98,14 @@ def get_port_obj(port_id=None, device_owner=None, ip_address=None):
'network_id': 'ba44f957-c467-412b-b985-ae720514bc46',
'tenant_id': 'b6e8fb2bde594673923afc19cf168f3a',
'created_at': '2017-06-09T13:23:24Z',
'binding:vnic_type': 'normal'})
'binding:vnic_type': 'normal'}
if ip_address:
port_obj.fixed_ips[0].ip_address = ip_address
port_obj['fixed_ips'][0].ip_address = ip_address
if port_id:
port_obj.id = port_id
port_obj['id'] = port_id
if device_owner:
port_obj.device_owner = device_owner
port_obj['device_owner'] = device_owner
return port_obj
@ -373,7 +371,7 @@ class BaseVIFPool(test_base.TestCase):
os_net = self.useFixture(k_fix.MockNetworkClient()).client
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
net_id = port['network_id']
tags = 'clusterTest'
port['tags'] = [tags]
@ -398,7 +396,7 @@ class BaseVIFPool(test_base.TestCase):
os_net = self.useFixture(k_fix.MockNetworkClient()).client
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
tags = 'clusterTest'
port['tags'] = [tags]
m_get_ports.return_value = [port]
@ -421,7 +419,7 @@ class BaseVIFPool(test_base.TestCase):
os_net = self.useFixture(k_fix.MockNetworkClient()).client
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
net_id = port['network_id']
tags = 'clusterTest'
port['tags'] = [tags]
@ -447,7 +445,7 @@ class BaseVIFPool(test_base.TestCase):
os_net = self.useFixture(k_fix.MockNetworkClient()).client
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
net_id = port['network_id']
tags = 'clusterTest'
m_get_ports.return_value = [port]
@ -471,7 +469,7 @@ class BaseVIFPool(test_base.TestCase):
os_net = self.useFixture(k_fix.MockNetworkClient()).client
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
m_get_ports.return_value = [port]
cls._cleanup_leftover_ports(m_driver)
@ -486,13 +484,13 @@ class BaseVIFPool(test_base.TestCase):
os_net = self.useFixture(k_fix.MockNetworkClient()).client
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
port['binding:host_id'] = None
m_get_ports.return_value = [port]
cls._cleanup_leftover_ports(m_driver)
os_net.networks.assert_not_called()
os_net.delete_port.assert_called_once_with(port.id)
os_net.delete_port.assert_called_once_with(port['id'])
@ddt.ddt
@ -887,7 +885,7 @@ class NeutronVIFPool(test_base.TestCase):
m_driver._available_ports_pools = {}
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_neutron_port(port_id=port_id)
filtered_ports = [port]
m_get_ports.return_value = filtered_ports
vif_plugin = mock.sentinel.plugin
@ -1449,7 +1447,7 @@ class NestedVIFPool(test_base.TestCase):
port_id = str(uuid.uuid4())
ip_address = mock.sentinel.ip_address
port_obj = get_port_obj(ip_address=ip_address)
port_obj = fake.get_port_obj(ip_address=ip_address)
os_net.get_port.return_value = port_obj
self.assertEqual(ip_address, cls._get_parent_port_ip(m_driver,
@ -1462,7 +1460,7 @@ class NestedVIFPool(test_base.TestCase):
m_driver = mock.MagicMock(spec=cls)
port_id = str(uuid.uuid4())
trunk_port = get_port_obj(port_id=port_id)
trunk_port = get_neutron_port(port_id=port_id)
trunk_id = str(uuid.uuid4())
trunk_details = {
'trunk_id': trunk_id,
@ -1473,8 +1471,8 @@ class NestedVIFPool(test_base.TestCase):
trunk_port['trunk_details'] = trunk_details
subport_id = str(uuid.uuid4())
subport = get_port_obj(port_id=subport_id,
device_owner='trunk:subport')
subport = get_neutron_port(port_id=subport_id,
device_owner='trunk:subport')
m_get_ports.return_value = [trunk_port, subport]
m_driver._get_in_use_ports.return_value = []
subnet = mock.sentinel.subnet
@ -1512,8 +1510,7 @@ class NestedVIFPool(test_base.TestCase):
m_driver = mock.MagicMock(spec=cls)
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id)
port = get_port_obj(port_id=port_id, device_owner='compute:nova')
port = get_neutron_port(port_id=port_id, device_owner='compute:nova')
m_get_ports.return_value = [port]
m_driver._get_in_use_ports.return_value = []
@ -1540,7 +1537,7 @@ class NestedVIFPool(test_base.TestCase):
port_id = str(uuid.uuid4())
trunk_id = str(uuid.uuid4())
trunk_obj = self._get_trunk_obj(port_id=trunk_id, subport_id=port_id)
port = get_port_obj(port_id=port_id, device_owner='trunk:subport')
port = get_neutron_port(port_id=port_id, device_owner='trunk:subport')
p_ports = self._get_parent_ports([trunk_obj])
a_subports = {port_id: port}
@ -1586,10 +1583,10 @@ class NestedVIFPool(test_base.TestCase):
port_id = str(uuid.uuid4())
trunk_id = str(uuid.uuid4())
trunk_obj = self._get_trunk_obj(port_id=trunk_id, subport_id=port_id)
port = get_port_obj(port_id=port_id, device_owner='trunk:subport')
port = get_neutron_port(port_id=port_id, device_owner='trunk:subport')
port_to_delete_id = str(uuid.uuid4())
port_to_delete = get_port_obj(port_id=port_to_delete_id,
device_owner='trunk:subport')
port_to_delete = get_neutron_port(port_id=port_to_delete_id,
device_owner='trunk:subport')
p_ports = self._get_parent_ports([trunk_obj])
a_subports = {port_id: port, port_to_delete_id: port_to_delete}
@ -1633,7 +1630,7 @@ class NestedVIFPool(test_base.TestCase):
port_id = str(uuid.uuid4())
trunk_id = str(uuid.uuid4())
trunk_obj = self._get_trunk_obj(port_id=trunk_id, subport_id=port_id)
port = get_port_obj(port_id=port_id, device_owner='trunk:subport')
port = get_neutron_port(port_id=port_id, device_owner='trunk:subport')
p_ports = self._get_parent_ports([trunk_obj])
a_subports = {port_id: port}
@ -1692,8 +1689,10 @@ class NestedVIFPool(test_base.TestCase):
subport_id=port_id2,
trunk_id=str(uuid.uuid4()))
port1 = get_port_obj(port_id=port_id1, device_owner='trunk:subport')
port2 = get_port_obj(port_id=port_id2, device_owner='trunk:subport')
port1 = get_neutron_port(port_id=port_id1,
device_owner='trunk:subport')
port2 = get_neutron_port(port_id=port_id2,
device_owner='trunk:subport')
p_ports = self._get_parent_ports([trunk_obj1, trunk_obj2])
a_subports = {port_id1: port1, port_id2: port2}
@ -1741,8 +1740,10 @@ class NestedVIFPool(test_base.TestCase):
trunk_obj['sub_ports'].append({'port_id': port_id2,
'segmentation_type': 'vlan',
'segmentation_id': 101})
port1 = get_port_obj(port_id=port_id1, device_owner='trunk:subport')
port2 = get_port_obj(port_id=port_id2, device_owner='trunk:subport')
port1 = get_neutron_port(port_id=port_id1,
device_owner='trunk:subport')
port2 = get_neutron_port(port_id=port_id2,
device_owner='trunk:subport')
p_ports = self._get_parent_ports([trunk_obj])
a_subports = {port_id1: port1, port_id2: port2}
@ -1817,11 +1818,12 @@ class NestedVIFPool(test_base.TestCase):
group='kubernetes')
port_id = str(uuid.uuid4())
port = get_port_obj(port_id=port_id, device_owner='trunk:subport')
port = fake.get_port_obj(port_id=port_id,
device_owner='trunk:subport')
p_ports = {}
a_subports = {}
subnet_id = port['fixed_ips'][0]['subnet_id']
subnet_id = port.fixed_ips[0]['subnet_id']
subnet = mock.sentinel.subnet
subnets = {subnet_id: {subnet_id: subnet}}
m_driver._get_trunks_info.return_value = (p_ports, a_subports,

88
kuryr_kubernetes/tests/unit/test_os_vif_util.py
View File

@ -28,6 +28,7 @@ from kuryr_kubernetes import constants as const
from kuryr_kubernetes import exceptions as k_exc
from kuryr_kubernetes import os_vif_util as ovu
from kuryr_kubernetes.tests import base as test_base
from kuryr_kubernetes.tests import fake
# REVISIT(ivc): move to kuryr-lib along with 'os_vif_util'
@ -177,7 +178,6 @@ class TestOSVIFUtils(test_base.TestCase):
m_get_ovs_hybrid_bridge_name):
vif_plugin = 'ovs'
port_id = mock.sentinel.port_id
mac_address = mock.sentinel.mac_address
ovs_bridge = mock.sentinel.ovs_bridge
port_filter = mock.sentinel.port_filter
subnets = mock.sentinel.subnets
@ -187,6 +187,10 @@ class TestOSVIFUtils(test_base.TestCase):
vif_name = mock.sentinel.vif_name
hybrid_bridge = mock.sentinel.hybrid_bridge
vif = mock.sentinel.vif
port = fake.get_port_obj(port_id=port_id,
vif_details={'ovs_hybrid_plug': True,
'bridge_name': ovs_bridge,
'port_filter': port_filter})
m_mk_profile.return_value = port_profile
m_make_vif_network.return_value = network
@ -195,14 +199,6 @@ class TestOSVIFUtils(test_base.TestCase):
m_get_ovs_hybrid_bridge_name.return_value = hybrid_bridge
m_mk_vif.return_value = vif
port = {'id': port_id,
'mac_address': mac_address,
'binding:vif_details': {
'ovs_hybrid_plug': True,
'bridge_name': ovs_bridge,
'port_filter': port_filter},
}
self.assertEqual(vif, ovu.neutron_to_osvif_vif_ovs(vif_plugin, port,
subnets))
@ -214,9 +210,9 @@ class TestOSVIFUtils(test_base.TestCase):
self.assertEqual(ovs_bridge, network.bridge)
m_mk_vif.assert_called_once_with(
id=port_id,
address=mac_address,
address=port.mac_address,
network=network,
has_traffic_filtering=port_filter,
has_traffic_filtering=port.binding_vif_details['port_filter'],
preserve_on_delete=False,
active=port_active,
port_profile=port_profile,
@ -236,36 +232,31 @@ class TestOSVIFUtils(test_base.TestCase):
m_is_port_active,
m_get_vif_name):
vif_plugin = 'ovs'
port_id = mock.sentinel.port_id
mac_address = mock.sentinel.mac_address
ovs_bridge = mock.sentinel.ovs_bridge
vif_details = {'ovs_hybrid_plug': False,
'bridge_name': mock.sentinel.ovs_bridge}
port = fake.get_port_obj(vif_details=vif_details)
port.active = mock.sentinel.port_active
port.profile = mock.sentinel.port_profile
subnets = mock.sentinel.subnets
port_profile = mock.sentinel.port_profile
network = mock.sentinel.network
port_active = mock.sentinel.port_active
vif_name = mock.sentinel.vif_name
vif = mock.sentinel.vif
m_mk_profile.return_value = port_profile
m_mk_profile.return_value = port.profile
m_make_vif_network.return_value = network
m_is_port_active.return_value = port_active
m_is_port_active.return_value = port.active
m_get_vif_name.return_value = vif_name
m_mk_vif.return_value = vif
port = {'id': port_id,
'mac_address': mac_address,
'binding:vif_details': {
'ovs_hybrid_plug': False,
'bridge_name': ovs_bridge},
}
self.assertEqual(vif, ovu.neutron_to_osvif_vif_ovs(vif_plugin, port,
subnets))
m_mk_profile.assert_called_once_with(interface_id=port_id)
m_mk_profile.assert_called_once_with(interface_id=port.id)
m_make_vif_network.assert_called_once_with(port, subnets)
m_is_port_active.assert_called_once_with(port)
m_get_vif_name.assert_called_once_with(port)
self.assertEqual(ovs_bridge, network.bridge)
self.assertEqual(network.bridge,
port.binding_vif_details['bridge_name'])
@mock.patch('kuryr_kubernetes.os_vif_util._get_vif_name')
@mock.patch('kuryr_kubernetes.os_vif_util._is_port_active')
@ -283,18 +274,15 @@ class TestOSVIFUtils(test_base.TestCase):
vif_name = mock.sentinel.vif_name
vif = mock.sentinel.vif
vlan_id = mock.sentinel.vlan_id
port = fake.get_port_obj(port_id=port_id,
vif_details={'port_filter': port_filter})
port.mac_address = mac_address
m_make_vif_network.return_value = network
m_is_port_active.return_value = port_active
m_get_vif_name.return_value = vif_name
m_mk_vif.return_value = vif
port = {'id': port_id,
'mac_address': mac_address,
'binding:vif_details': {
'port_filter': port_filter},
}
self.assertEqual(vif, ovu.neutron_to_osvif_vif_nested_vlan(port,
subnets, vlan_id))
@ -358,8 +346,7 @@ class TestOSVIFUtils(test_base.TestCase):
def test_neutron_to_osvif_vif_ovs_no_bridge(self):
vif_plugin = 'ovs'
port = munch.Munch({'id': str(uuid.uuid4()),
'binding_vif_details': {}})
port = fake.get_port_obj(port_id=str(uuid.uuid4()))
subnets = {}
self.assertRaises(o_cfg.RequiredOptError,
@ -367,31 +354,30 @@ class TestOSVIFUtils(test_base.TestCase):
vif_plugin, port, subnets)
def test_get_ovs_hybrid_bridge_name(self):
port_id = str(uuid.uuid4())
port = {'id': port_id}
port = fake.get_port_obj(port_id=str(uuid.uuid4()))
self.assertEqual("qbr" + port_id[:11],
self.assertEqual("qbr" + port.id[:11],
ovu._get_ovs_hybrid_bridge_name(port))
def test_is_port_active(self):
port = {'status': 'ACTIVE'}
port = fake.get_port_obj(port_id=str(uuid.uuid4()))
port.status = 'ACTIVE'
self.assertTrue(ovu._is_port_active(port))
def test_is_port_inactive(self):
port = {'status': 'DOWN'}
port = fake.get_port_obj(port_id=str(uuid.uuid4()))
self.assertFalse(ovu._is_port_active(port))
@mock.patch('kuryr.lib.binding.drivers.utils.get_veth_pair_names')
def test_get_vif_name(self, m_get_veth_pair_names):
port_id = mock.sentinel.port_id
vif_name = mock.sentinel.vif_name
port = {'id': port_id}
port = fake.get_port_obj(port_id=str(uuid.uuid4()))
m_get_veth_pair_names.return_value = (vif_name, mock.sentinel.any)
self.assertEqual(vif_name, ovu._get_vif_name(port))
m_get_veth_pair_names.assert_called_once_with(port_id)
m_get_veth_pair_names.assert_called_once_with(port.id)
@mock.patch('kuryr_kubernetes.os_vif_util._make_vif_subnets')
@mock.patch('os_vif.objects.subnet.SubnetList')
@ -477,22 +463,6 @@ class TestOSVIFUtils(test_base.TestCase):
self.assertRaises(k_exc.IntegrityError, ovu._make_vif_subnet,
subnets, subnet_id)
def test_osvif_to_neutron_network_ids(self):
id_a = mock.sentinel.id_a
id_b = mock.sentinel.id_b
net1 = mock.Mock()
net1.id = id_a
net2 = mock.Mock()
net2.id = id_b
net3 = mock.Mock()
net3.id = id_a
subnets = {1: net1, 2: net2, 3: net3}
ret = ovu.osvif_to_neutron_network_ids(subnets)
self.assertEqual(2, len(ret))
self.assertIn(id_a, ret)
self.assertIn(id_b, ret)
def test_osvif_to_neutron_fixed_ips(self):
ip11 = '1.1.1.1'
ip12 = '2.2.2.2'