Improve DHCP RPC handler

Remove unnecessary DB retrieval operations from "get_network_info"
method.

Partial-Bug: #1950662
Change-Id: If4b33c8437dba411fed913e7e1c7f06d899c08f7
This commit is contained in:
Rodolfo Alonso Hernandez 2021-11-05 14:02:48 +00:00
parent 90b5456b8c
commit c686a2b555
7 changed files with 143 additions and 106 deletions

View File

@ -170,6 +170,14 @@ class NetModel(DictModel):
def namespace(self):
return self._ns_name
# TODO(ralonsoh): remove in Z+.
@property
def project_id(self):
try:
return self['project_id']
except KeyError:
return self['tenant_id']
class DhcpBase(object, metaclass=abc.ABCMeta):
@ -1584,7 +1592,7 @@ class DeviceManager(object):
admin_state_up=True,
device_id=device_id,
network_id=network.id,
tenant_id=network.tenant_id,
project_id=network.project_id,
fixed_ips=unique_ip_subnets)
return self.plugin.create_dhcp_port({'port': port_dict})

View File

@ -36,6 +36,7 @@ from neutron._i18n import _
from neutron.common import utils
from neutron.db import provisioning_blocks
from neutron.extensions import segment as segment_ext
from neutron.objects import network as network_obj
from neutron.quota import resource_registry
@ -218,43 +219,53 @@ class DhcpRpcCallback(object):
LOG.debug('Network %(network_id)s requested from '
'%(host)s', {'network_id': network_id,
'host': host})
plugin = directory.get_plugin()
try:
network = plugin.get_network(context, network_id)
except exceptions.NetworkNotFound:
LOG.debug("Network %s could not be found, it might have "
"been deleted concurrently.", network_id)
network = network_obj.Network.get_object(context, id=network_id)
if not network:
LOG.debug('Network %s could not be found, it might have '
'been deleted concurrently.', network_id)
return
subnet_filters = {'network_id': [network_id], 'enable_dhcp': [True]}
subnets = plugin.get_subnets(context, filters=subnet_filters)
seg_plug = directory.get_plugin(
segment_ext.SegmentPluginBase.get_plugin_type())
plugin = directory.get_plugin()
# Only subnets with DHCP enabled.
subnets = [plugin._make_subnet_dict(subnet) for subnet in
network.db_obj.subnets if subnet.enable_dhcp]
seg_subnets = [subnet for subnet in subnets if
subnet.get('segment_id')]
nonlocal_subnets = []
if seg_plug and subnets:
seg_subnets = [subnet for subnet in subnets
if subnet.get('segment_id')]
# If there are no subnets with segments, then this is not a routed
# network and no filtering should take place.
if seg_subnets:
segment_ids = seg_plug.get_segments_by_hosts(context, [host])
# There might be something to do if no segment_ids exist that
# are mapped to this host. However, it seems that if this
# host is not mapped to any segments and this is a routed
# network, then this host shouldn't have even been scheduled
# to.
nonlocal_subnets = [subnet for subnet in seg_subnets
if subnet['segment_id'] not in segment_ids]
subnets = [subnet for subnet in seg_subnets
if subnet['segment_id'] in segment_ids]
# If there are no subnets with segments, then this is not a routed
# network and no filtering should take place.
if seg_plug and seg_subnets:
# Network subnets can be associated only to network segments.
# From those segments, we retrieve only those ones mapped to
# 'host'.
segment_ids = {ns.id for ns in network.segments if
host in ns.hosts}
# There might be something to do if no segment_ids exist that
# are mapped to this host. However, it seems that if this
# host is not mapped to any segments and this is a routed
# network, then this host shouldn't have even been scheduled
# to.
nonlocal_subnets = [subnet for subnet in seg_subnets
if subnet.get('segment_id') not in segment_ids]
subnets = [subnet for subnet in seg_subnets
if subnet.get('segment_id') in segment_ids]
# NOTE(kevinbenton): we sort these because the agent builds tags
# based on position in the list and has to restart the process if
# the order changes.
network['subnets'] = sorted(subnets, key=operator.itemgetter('id'))
network['non_local_subnets'] = sorted(nonlocal_subnets,
key=operator.itemgetter('id'))
port_filters = {'network_id': [network_id]}
network['ports'] = plugin.get_ports(context, filters=port_filters)
return network
# TODO(ralonsoh): in Z+, remove "tenant_id" parameter. DHCP agents
# should read only "project_id".
return {'id': network.id,
'project_id': network.project_id,
'tenant_id': network.project_id,
'admin_state_up': network.admin_state_up,
'subnets': sorted(subnets, key=operator.itemgetter('id')),
'non_local_subnets': sorted(nonlocal_subnets,
key=operator.itemgetter('id')),
'ports': plugin.get_ports(context, filters=port_filters),
'mtu': network.mtu}
@db_api.retry_db_errors
def release_dhcp_port(self, context, **kwargs):

View File

@ -49,7 +49,7 @@ class TestDhcp(functional_base.BaseSudoTestCase):
dev_mgr = dhcp.DeviceManager(self.conf, plugin)
network = {
'id': 'foo_id',
'tenant_id': 'foo_tenant',
'project_id': 'foo_project',
'namespace': 'qdhcp-foo_id',
'ports': [],
'subnets': [tests_base.AttributeDict({'id': 'subnet_foo_id',

View File

@ -147,7 +147,7 @@ class DHCPAgentOVSTestFramework(base.BaseSudoTestCase):
non_local_subnets=non_local_subnets,
ports=ports,
admin_state_up=True,
tenant_id=uuidutils.generate_uuid())
project_id=uuidutils.generate_uuid())
return net_dict
def get_interface_name(self, network, port):

View File

@ -51,7 +51,7 @@ DEVICE_MANAGER = '%s.%s' % (dev_man.__module__, dev_man.__name__)
DHCP_PLUGIN = '%s.%s' % (rpc_api.__module__, rpc_api.__name__)
FAKE_NETWORK_UUID = '12345678-1234-5678-1234567890ab'
FAKE_NETWORK_DHCP_NS = "qdhcp-%s" % FAKE_NETWORK_UUID
FAKE_TENANT_ID = 'aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'
FAKE_PROJECT_ID = 'aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'
FAKE_PRIORITY = 6
@ -60,7 +60,7 @@ fake_subnet1_allocation_pools = dhcp.DictModel(id='', start='172.9.9.2',
fake_subnet1 = dhcp.DictModel(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='172.9.9.0/24', enable_dhcp=True, name='',
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
gateway_ip='172.9.9.1', host_routes=[],
dns_nameservers=[],
ip_version=const.IP_VERSION_4,
@ -72,7 +72,8 @@ fake_subnet2_allocation_pools = dhcp.DictModel(id='', start='172.9.8.2',
fake_subnet2 = dhcp.DictModel(id='dddddddd-dddd-dddd-dddddddddddd',
network_id=FAKE_NETWORK_UUID,
cidr='172.9.8.0/24', enable_dhcp=False, name='',
tenant_id=FAKE_TENANT_ID, gateway_ip='172.9.8.1',
project_id=FAKE_PROJECT_ID,
gateway_ip='172.9.8.1',
host_routes=[], dns_nameservers=[],
ip_version=const.IP_VERSION_4,
allocation_pools=fake_subnet2_allocation_pools)
@ -85,7 +86,7 @@ fake_subnet3 = dhcp.DictModel(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
fake_ipv6_subnet = dhcp.DictModel(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='2001:0db8::0/64', enable_dhcp=True,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
gateway_ip='2001:0db8::1',
ip_version=const.IP_VERSION_6,
ipv6_ra_mode='slaac', ipv6_address_mode=None)
@ -160,44 +161,44 @@ fake_dist_port = dhcp.DictModel(id='12345678-1234-aaaa-1234567890ab',
fixed_ips=[fake_meta_fixed_ip])
fake_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_subnet1, fake_subnet2],
ports=[fake_port1])
fake_network_ipv6 = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_ipv6_subnet],
ports=[fake_ipv6_port])
fake_network_ipv6_ipv4 = dhcp.NetModel(
id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_ipv6_subnet, fake_subnet1],
ports=[fake_port1])
isolated_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_subnet1],
ports=[fake_port1])
nonisolated_dist_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_subnet1],
ports=[fake_port1, fake_port2])
empty_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_subnet1],
ports=[])
fake_meta_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_meta_subnet],
ports=[fake_meta_port])
@ -206,13 +207,13 @@ fake_meta_dvr_network = dhcp.NetModel(fake_meta_network)
fake_meta_dvr_network['ports'] = [fake_meta_dvr_port]
fake_dist_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[fake_meta_subnet],
ports=[fake_meta_port, fake_dist_port])
fake_down_network = dhcp.NetModel(id='12345678-dddd-dddd-1234567890ab',
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=False,
subnets=[],
ports=[])
@ -1113,7 +1114,7 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
def test_refresh_dhcp_helper_no_dhcp_enabled_networks(self):
network = dhcp.NetModel(dict(id='net-id',
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[],
ports=[]))
@ -1130,7 +1131,7 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
def test_refresh_dhcp_helper_exception_during_rpc(self):
network = dhcp.NetModel(dict(id='net-id',
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
admin_state_up=True,
subnets=[],
ports=[]))
@ -1221,7 +1222,7 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
def test_subnet_update_end_restart(self):
new_state = dhcp.NetModel(dict(id=fake_network.id,
tenant_id=fake_network.tenant_id,
project_id=fake_network.project_id,
admin_state_up=True,
subnets=[fake_subnet1, fake_subnet3],
ports=[fake_port1]))
@ -1240,7 +1241,7 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
def test_subnet_delete_end_no_network_id(self):
prev_state = dhcp.NetModel(dict(id=fake_network.id,
tenant_id=fake_network.tenant_id,
project_id=fake_network.project_id,
admin_state_up=True,
subnets=[fake_subnet1, fake_subnet3],
ports=[fake_port1]))
@ -1264,7 +1265,7 @@ class TestDhcpAgentEventHandler(base.BaseTestCase):
def test_subnet_update_end_delete_payload(self):
prev_state = dhcp.NetModel(dict(id=fake_network.id,
tenant_id=fake_network.tenant_id,
project_id=fake_network.project_id,
admin_state_up=True,
subnets=[fake_subnet1, fake_subnet3],
ports=[fake_port1]))
@ -1593,7 +1594,7 @@ class TestNetworkCache(base.BaseTestCase):
def test_get_port_ids(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
subnets=[fake_subnet1],
ports=[fake_port1]))
self.nc.put(fake_net)
@ -1604,7 +1605,7 @@ class TestNetworkCache(base.BaseTestCase):
def test_get_port_ids_limited_nets(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
subnets=[fake_subnet1],
ports=[fake_port1]))
fake_port2 = copy.deepcopy(fake_port1)
@ -1612,7 +1613,7 @@ class TestNetworkCache(base.BaseTestCase):
fake_port2['network_id'] = '12345678-1234-5678-1234567890ac'
fake_net2 = dhcp.NetModel(
dict(id='12345678-1234-5678-1234567890ac',
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
subnets=[fake_subnet1],
ports=[fake_port2]))
self.nc.put(fake_net)
@ -1628,7 +1629,7 @@ class TestNetworkCache(base.BaseTestCase):
def test_put_port(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
subnets=[fake_subnet1],
ports=[fake_port1]))
self.nc.put(fake_net)
@ -1639,7 +1640,7 @@ class TestNetworkCache(base.BaseTestCase):
def test_put_port_existing(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
subnets=[fake_subnet1],
ports=[fake_port1, fake_port2]))
self.nc.put(fake_net)
@ -1651,7 +1652,7 @@ class TestNetworkCache(base.BaseTestCase):
def test_remove_port_existing(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
project_id=FAKE_PROJECT_ID,
subnets=[fake_subnet1],
ports=[fake_port1, fake_port2]))
self.nc.put(fake_net)
@ -1891,7 +1892,7 @@ class TestDeviceManager(base.BaseTestCase):
plugin.assert_has_calls([
mock.call.create_dhcp_port(
{'port': {'name': '', 'admin_state_up': True,
'network_id': net.id, 'tenant_id': net.tenant_id,
'network_id': net.id, 'project_id': net.project_id,
'fixed_ips':
[{'subnet_id': port.fixed_ips[0].subnet_id}],
'device_id': mock.ANY}})])
@ -1981,7 +1982,7 @@ class TestDeviceManager(base.BaseTestCase):
mock.call.create_dhcp_port(
{'port': {'name': '', 'admin_state_up': True,
'network_id': net.id,
'tenant_id': net.tenant_id,
'project_id': net.project_id,
'fixed_ips': [{'subnet_id':
fake_dhcp_port.fixed_ips[0].subnet_id}],
'device_id': mock.ANY}})])
@ -2027,8 +2028,8 @@ class TestDeviceManager(base.BaseTestCase):
plugin.assert_has_calls([
mock.call.create_dhcp_port(
{'port': {'name': '', 'admin_state_up': True,
'network_id':
fake_network.id, 'tenant_id': fake_network.tenant_id,
'network_id': fake_network.id,
'project_id': fake_network.project_id,
'fixed_ips':
[{'subnet_id': fake_fixed_ip1.subnet_id}],
'device_id': mock.ANY}})])
@ -2111,7 +2112,7 @@ class TestDeviceManager(base.BaseTestCase):
def test_destroy(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID))
project_uid=FAKE_PROJECT_ID))
with mock.patch('neutron.agent.linux.interface.NullDriver') as dvr_cls:
mock_driver = mock.MagicMock()
@ -2134,7 +2135,7 @@ class TestDeviceManager(base.BaseTestCase):
def test_destroy_with_none(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID))
project_id=FAKE_PROJECT_ID))
with mock.patch('neutron.agent.linux.interface.NullDriver') as dvr_cls:
mock_driver = mock.MagicMock()
@ -2155,7 +2156,7 @@ class TestDeviceManager(base.BaseTestCase):
def test_get_interface_name(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID))
project_id=FAKE_PROJECT_ID))
fake_port = dhcp.DictModel(
dict(id='12345678-1234-aaaa-1234567890ab',
@ -2181,7 +2182,7 @@ class TestDeviceManager(base.BaseTestCase):
def test_get_device_id(self):
fake_net = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID))
project_id=FAKE_PROJECT_ID))
expected = ('dhcp1ae5f96c-c527-5079-82ea-371a01645457-12345678-1234-'
'5678-1234567890ab')
# the DHCP port name only contains the hostname and not the domain name

View File

@ -3148,7 +3148,7 @@ class TestDeviceManager(TestConfBase):
# Setup with no existing DHCP port - expect a new DHCP port to
# be created.
network = FakeDeviceManagerNetwork()
network.tenant_id = 'Tenant A'
network.project_id = 'Project A'
def mock_create(dict):
port = dhcp.DictModel(dict['port'])
@ -3231,7 +3231,7 @@ class TestDeviceManager(TestConfBase):
# Setup with a reserved DHCP port.
network = FakeDualNetworkReserved()
network.tenant_id = 'Tenant A'
network.project_id = 'Project A'
reserved_port = network.ports[-1]
def mock_update(port_id, dict):
@ -3302,7 +3302,7 @@ class TestDeviceManager(TestConfBase):
# Setup with a reserved DHCP port.
network = FakeDualNetworkReserved2()
network.tenant_id = 'Tenant A'
network.project_id = 'Project A'
reserved_port_1 = network.ports[-2]
reserved_port_2 = network.ports[-1]
@ -3333,7 +3333,7 @@ class TestDeviceManager(TestConfBase):
"""
# Setup with a reserved DHCP port.
fake_network = FakeDualNetworkReserved2()
fake_network.tenant_id = 'Tenant A'
fake_network.project_id = 'Project A'
reserved_port_2 = fake_network.ports[-1]
mock_plugin = mock.Mock()

View File

@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import operator
from unittest import mock
from neutron_lib.api.definitions import portbindings
@ -23,10 +24,12 @@ from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from oslo_db import exception as db_exc
from oslo_messaging.rpc import dispatcher as rpc_dispatcher
from oslo_utils import uuidutils
from neutron.api.rpc.handlers import dhcp_rpc
from neutron.common import utils
from neutron.db import provisioning_blocks
from neutron.objects import network as network_obj
from neutron.tests import base
@ -207,50 +210,64 @@ class TestDhcpRpcCallback(base.BaseTestCase):
context='ctx', host='host', port_id='66',
port={'port': {'network_id': 'a'}}))
def test_get_network_info_return_none_on_not_found(self):
self.plugin.get_network.side_effect = exceptions.NetworkNotFound(
net_id='a')
@mock.patch.object(network_obj.Network, 'get_object', return_value=None)
def test_get_network_info_return_none_on_not_found(self, *args):
retval = self.callbacks.get_network_info(mock.Mock(), network_id='a')
self.assertIsNone(retval)
def _test_get_network_info(self, segmented_network=False,
routed_network=False):
network_retval = dict(id='a')
if not routed_network:
subnet_retval = [dict(id='a'), dict(id='c'), dict(id='b')]
else:
subnet_retval = [dict(id='c', segment_id='1'),
dict(id='b', segment_id='2'),
dict(id='a', segment_id='1')]
port_retval = mock.Mock()
@mock.patch.object(network_obj.Network, 'get_object')
def _test_get_network_info(self, mock_net_get_object,
segmented_network=False, routed_network=False):
def _network_to_dict(network, ports):
segment_ids = ['1']
subnets = [_make_subnet_dict(sn) for sn in network.db_obj.subnets]
if routed_network:
non_local_subnets = [subnet for subnet in subnets
if subnet.get('segment_id') not in
segment_ids]
subnets = [subnet for subnet in subnets
if subnet.get('segment_id') in segment_ids]
else:
non_local_subnets = []
return {'id': network.id,
'project_id': network.project_id,
'tenant_id': network.project_id,
'admin_state_up': network.admin_state_up,
'ports': ports,
'subnets': sorted(subnets, key=operator.itemgetter('id')),
'non_local_subnets': sorted(non_local_subnets,
key=operator.itemgetter('id')),
'mtu': network.mtu}
def _make_subnet_dict(subnet):
ret = {'id': subnet.id}
if type(subnet.segment_id) == str:
ret['segment_id'] = subnet.segment_id
return ret
if not routed_network:
subnets = [mock.Mock(id='a'), mock.Mock(id='c'), mock.Mock(id='b')]
else:
subnets = [mock.Mock(id='a', segment_id='1'),
mock.Mock(id='c', segment_id='2'),
mock.Mock(id='b', segment_id='1')]
db_obj = mock.Mock(subnets=subnets)
project_id = uuidutils.generate_uuid()
network = mock.Mock(id='a', admin_state_up=True, db_obj=db_obj,
project_id=project_id, mtu=1234)
ports = mock.Mock()
mock_net_get_object.return_value = network
self.plugin.get_ports.return_value = ports
self.plugin._make_subnet_dict = _make_subnet_dict
self.plugin.get_network.return_value = network_retval
self.plugin.get_subnets.return_value = subnet_retval
self.plugin.get_ports.return_value = port_retval
if segmented_network:
self.segment_plugin.get_segments.return_value = [dict(id='1'),
dict(id='2')]
self.segment_plugin.get_segments_by_hosts.return_value = ['1']
network.segments = [mock.Mock(id='1', hosts=['host1']),
mock.Mock(id='2', hosts=['host2'])]
retval = self.callbacks.get_network_info(mock.Mock(), network_id='a')
self.assertEqual(retval, network_retval)
sorted_nonlocal_subnet_retval = []
if not routed_network:
sorted_subnet_retval = [dict(id='a'), dict(id='b'), dict(id='c')]
else:
sorted_subnet_retval = [dict(id='a', segment_id='1'),
dict(id='c', segment_id='1')]
sorted_nonlocal_subnet_retval = [dict(id='b', segment_id='2')]
self.assertEqual(retval['subnets'], sorted_subnet_retval)
self.assertEqual(retval['non_local_subnets'],
sorted_nonlocal_subnet_retval)
self.assertEqual(retval['ports'], port_retval)
subnet_filters = {'network_id': [network_retval['id']],
'enable_dhcp': [True]}
self.plugin.assert_has_calls(
[mock.call.get_network(mock.ANY, 'a'),
mock.call.get_subnets(mock.ANY, filters=subnet_filters),
mock.call.get_ports(mock.ANY, filters={'network_id': ['a']})])
retval = self.callbacks.get_network_info(mock.Mock(), network_id='a',
host='host1')
reference = _network_to_dict(network, ports)
self.assertEqual(reference, retval)
def test_get_network_info(self):
self._test_get_network_info()