# Copyright 2012 OpenStack Foundation # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import mock from neutron.agent.common import config from neutron.agent.common import ovs_lib from neutron.agent.linux import interface from neutron.agent.linux import ip_lib from neutron.agent.linux import utils from neutron.common import constants from neutron.extensions import flavor from neutron.openstack.common import uuidutils from neutron.tests import base class BaseChild(interface.LinuxInterfaceDriver): def plug_new(*args): pass def unplug(*args): pass class FakeNetwork(object): id = '12345678-1234-5678-90ab-ba0987654321' class FakeSubnet(object): cidr = '192.168.1.1/24' class FakeAllocation(object): subnet = FakeSubnet() ip_address = '192.168.1.2' ip_version = 4 class FakePort(object): id = 'abcdef01-1234-5678-90ab-ba0987654321' fixed_ips = [FakeAllocation] device_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc' network = FakeNetwork() network_id = network.id class TestBase(base.BaseTestCase): def setUp(self): super(TestBase, self).setUp() self.conf = config.setup_conf() self.conf.register_opts(interface.OPTS) self.ip_dev_p = mock.patch.object(ip_lib, 'IPDevice') self.ip_dev = self.ip_dev_p.start() self.ip_p = mock.patch.object(ip_lib, 'IPWrapper') self.ip = self.ip_p.start() self.device_exists_p = mock.patch.object(ip_lib, 'device_exists') self.device_exists = self.device_exists_p.start() class TestABCDriver(TestBase): def setUp(self): super(TestABCDriver, self).setUp() mock_link_addr = mock.PropertyMock(return_value='aa:bb:cc:dd:ee:ff') type(self.ip_dev().link).address = mock_link_addr def test_get_device_name(self): bc = BaseChild(self.conf) device_name = bc.get_device_name(FakePort()) self.assertEqual('tapabcdef01-12', device_name) def test_l3_init(self): addresses = [dict(scope='global', dynamic=False, cidr='172.16.77.240/24')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) self.ip_dev().route.list_onlink_routes.return_value = [] bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns, extra_subnets=[{'cidr': '172.20.0.0/24'}]) self.ip_dev.assert_has_calls( [mock.call('tap0', namespace=ns), mock.call().addr.list(filters=['permanent']), mock.call().addr.add('192.168.1.2/24'), mock.call().addr.delete('172.16.77.240/24'), mock.call().route.list_onlink_routes(constants.IP_VERSION_4), mock.call().route.list_onlink_routes(constants.IP_VERSION_6), mock.call().route.add_onlink_route('172.20.0.0/24')]) def test_l3_init_delete_onlink_routes(self): addresses = [dict(scope='global', dynamic=False, cidr='172.16.77.240/24')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) self.ip_dev().route.list_onlink_routes.return_value = ['172.20.0.0/24'] bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns) self.ip_dev.assert_has_calls( [mock.call().route.list_onlink_routes(constants.IP_VERSION_4), mock.call().route.list_onlink_routes(constants.IP_VERSION_6), mock.call().route.delete_onlink_route('172.20.0.0/24')]) def test_l3_init_with_preserve(self): addresses = [dict(scope='global', dynamic=False, cidr='192.168.1.3/32')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns, preserve_ips=['192.168.1.3/32']) self.ip_dev.assert_has_calls( [mock.call('tap0', namespace=ns), mock.call().addr.list(filters=['permanent']), mock.call().addr.add('192.168.1.2/24')]) self.assertFalse(self.ip_dev().addr.delete.called) self.assertFalse(self.ip_dev().delete_addr_and_conntrack_state.called) def _test_l3_init_clean_connections(self, clean_connections): addresses = [ dict(scope='global', dynamic=False, cidr='10.0.0.1/24'), dict(scope='global', dynamic=False, cidr='10.0.0.3/32')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['10.0.0.1/24'], namespace=ns, clean_connections=clean_connections) delete = self.ip_dev().delete_addr_and_conntrack_state if clean_connections: delete.assert_called_once_with('10.0.0.3/32') else: self.assertFalse(delete.called) def test_l3_init_with_clean_connections(self): self._test_l3_init_clean_connections(True) def test_l3_init_without_clean_connections(self): self._test_l3_init_clean_connections(False) def _test_l3_init_with_ipv6(self, include_gw_ip): addresses = [dict(scope='global', dynamic=False, cidr='2001:db8:a::123/64')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) self.ip_dev().route.list_onlink_routes.return_value = [] bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' new_cidr = '2001:db8:a::124/64' kwargs = {'namespace': ns, 'extra_subnets': [{'cidr': '2001:db8:b::/64'}]} if include_gw_ip: kwargs['gateway_ips'] = ['2001:db8:a::1'] bc.init_l3('tap0', [new_cidr], **kwargs) expected_calls = ( [mock.call('tap0', namespace=ns), mock.call().addr.list(filters=['permanent']), mock.call().addr.add('2001:db8:a::124/64'), mock.call().addr.delete('2001:db8:a::123/64')]) if include_gw_ip: expected_calls += ( [mock.call().route.add_gateway('2001:db8:a::1')]) expected_calls += ( [mock.call().route.list_onlink_routes(constants.IP_VERSION_4), mock.call().route.list_onlink_routes(constants.IP_VERSION_6), mock.call().route.add_onlink_route('2001:db8:b::/64')]) self.ip_dev.assert_has_calls(expected_calls) def test_l3_init_ipv6_with_gw_ip(self): self._test_l3_init_with_ipv6(include_gw_ip=True) def test_l3_init_ipv6_without_gw_ip(self): self._test_l3_init_with_ipv6(include_gw_ip=False) def test_l3_init_ext_gw_with_dual_stack(self): old_addrs = [dict(ip_version=4, scope='global', dynamic=False, cidr='172.16.77.240/24'), dict(ip_version=6, scope='global', dynamic=False, cidr='2001:db8:a::123/64')] self.ip_dev().addr.list = mock.Mock(return_value=old_addrs) self.ip_dev().route.list_onlink_routes.return_value = [] bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' new_cidrs = ['192.168.1.2/24', '2001:db8:a::124/64'] bc.init_l3('tap0', new_cidrs, namespace=ns, extra_subnets=[{'cidr': '172.20.0.0/24'}]) self.ip_dev.assert_has_calls( [mock.call('tap0', namespace=ns), mock.call().addr.list(filters=['permanent']), mock.call().addr.add('192.168.1.2/24'), mock.call().addr.add('2001:db8:a::124/64'), mock.call().addr.delete('172.16.77.240/24'), mock.call().addr.delete('2001:db8:a::123/64'), mock.call().route.list_onlink_routes(constants.IP_VERSION_4), mock.call().route.list_onlink_routes(constants.IP_VERSION_6), mock.call().route.add_onlink_route('172.20.0.0/24')], any_order=True) def test_l3_init_with_ipv6_delete_onlink_routes(self): addresses = [dict(scope='global', dynamic=False, cidr='2001:db8:a::123/64')] route = '2001:db8:a::/64' self.ip_dev().addr.list = mock.Mock(return_value=addresses) self.ip_dev().route.list_onlink_routes.return_value = [route] bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['2001:db8:a::124/64'], namespace=ns) self.ip_dev.assert_has_calls( [mock.call().route.list_onlink_routes(constants.IP_VERSION_4), mock.call().route.list_onlink_routes(constants.IP_VERSION_6), mock.call().route.delete_onlink_route(route)]) def test_l3_init_with_duplicated_ipv6(self): addresses = [dict(scope='global', dynamic=False, cidr='2001:db8:a::123/64')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['2001:db8:a::123/64'], namespace=ns) self.assertFalse(self.ip_dev().addr.add.called) def test_l3_init_with_duplicated_ipv6_uncompact(self): addresses = [dict(scope='global', dynamic=False, cidr='2001:db8:a::123/64')] self.ip_dev().addr.list = mock.Mock(return_value=addresses) bc = BaseChild(self.conf) ns = '12345678-1234-5678-90ab-ba0987654321' bc.init_l3('tap0', ['2001:db8:a:0000:0000:0000:0000:0123/64'], namespace=ns) self.assertFalse(self.ip_dev().addr.add.called) class TestOVSInterfaceDriver(TestBase): def test_get_device_name(self): br = interface.OVSInterfaceDriver(self.conf) device_name = br.get_device_name(FakePort()) self.assertEqual('tapabcdef01-12', device_name) def test_plug_no_ns(self): self._test_plug() def test_plug_with_ns(self): self._test_plug(namespace='01234567-1234-1234-99') def test_plug_alt_bridge(self): self._test_plug(bridge='br-foo') def test_plug_configured_bridge(self): br = 'br-v' self.conf.set_override('ovs_use_veth', False) self.conf.set_override('ovs_integration_bridge', br) self.assertEqual(self.conf.ovs_integration_bridge, br) def device_exists(dev, namespace=None): return dev == br ovs = interface.OVSInterfaceDriver(self.conf) with mock.patch.object(ovs, '_ovs_add_port') as add_port: self.device_exists.side_effect = device_exists ovs.plug('01234567-1234-1234-99', 'port-1234', 'tap0', 'aa:bb:cc:dd:ee:ff', bridge=None, namespace=None) add_port.assert_called_once_with('br-v', 'tap0', 'port-1234', 'aa:bb:cc:dd:ee:ff', internal=True) def _test_plug(self, additional_expectation=[], bridge=None, namespace=None): if not bridge: bridge = 'br-int' def device_exists(dev, namespace=None): return dev == bridge with mock.patch.object(ovs_lib.OVSBridge, 'replace_port') as replace: ovs = interface.OVSInterfaceDriver(self.conf) self.device_exists.side_effect = device_exists ovs.plug('01234567-1234-1234-99', 'port-1234', 'tap0', 'aa:bb:cc:dd:ee:ff', bridge=bridge, namespace=namespace) replace.assert_called_once_with( 'tap0', ('type', 'internal'), ('external_ids', { 'iface-id': 'port-1234', 'iface-status': 'active', 'attached-mac': 'aa:bb:cc:dd:ee:ff'})) expected = [mock.call(), mock.call().device('tap0'), mock.call().device().link.set_address('aa:bb:cc:dd:ee:ff')] expected.extend(additional_expectation) if namespace: expected.extend( [mock.call().ensure_namespace(namespace), mock.call().ensure_namespace().add_device_to_namespace( mock.ANY)]) expected.extend([mock.call().device().link.set_up()]) self.ip.assert_has_calls(expected) def test_mtu_int(self): self.assertIsNone(self.conf.network_device_mtu) self.conf.set_override('network_device_mtu', 9000) self.assertEqual(self.conf.network_device_mtu, 9000) def test_plug_mtu(self): self.conf.set_override('network_device_mtu', 9000) self._test_plug([mock.call().device().link.set_mtu(9000)]) def test_unplug(self, bridge=None): if not bridge: bridge = 'br-int' with mock.patch('neutron.agent.common.ovs_lib.OVSBridge') as ovs_br: ovs = interface.OVSInterfaceDriver(self.conf) ovs.unplug('tap0') ovs_br.assert_has_calls([mock.call(bridge), mock.call().delete_port('tap0')]) class TestOVSInterfaceDriverWithVeth(TestOVSInterfaceDriver): def setUp(self): super(TestOVSInterfaceDriverWithVeth, self).setUp() self.conf.set_override('ovs_use_veth', True) def test_get_device_name(self): br = interface.OVSInterfaceDriver(self.conf) device_name = br.get_device_name(FakePort()) self.assertEqual('ns-abcdef01-12', device_name) def test_plug_with_prefix(self): self._test_plug(devname='qr-0', prefix='qr-') def _test_plug(self, devname=None, bridge=None, namespace=None, prefix=None, mtu=None): if not devname: devname = 'ns-0' if not bridge: bridge = 'br-int' def device_exists(dev, namespace=None): return dev == bridge ovs = interface.OVSInterfaceDriver(self.conf) self.device_exists.side_effect = device_exists root_dev = mock.Mock() ns_dev = mock.Mock() self.ip().add_veth = mock.Mock(return_value=(root_dev, ns_dev)) expected = [mock.call(), mock.call().add_veth('tap0', devname, namespace2=namespace)] with mock.patch.object(ovs_lib.OVSBridge, 'replace_port') as replace: ovs.plug('01234567-1234-1234-99', 'port-1234', devname, 'aa:bb:cc:dd:ee:ff', bridge=bridge, namespace=namespace, prefix=prefix) replace.assert_called_once_with( 'tap0', ('external_ids', { 'iface-id': 'port-1234', 'iface-status': 'active', 'attached-mac': 'aa:bb:cc:dd:ee:ff'})) ns_dev.assert_has_calls( [mock.call.link.set_address('aa:bb:cc:dd:ee:ff')]) if mtu: ns_dev.assert_has_calls([mock.call.link.set_mtu(mtu)]) root_dev.assert_has_calls([mock.call.link.set_mtu(mtu)]) self.ip.assert_has_calls(expected) root_dev.assert_has_calls([mock.call.link.set_up()]) ns_dev.assert_has_calls([mock.call.link.set_up()]) def test_plug_mtu(self): self.conf.set_override('network_device_mtu', 9000) self._test_plug(mtu=9000) def test_unplug(self, bridge=None): if not bridge: bridge = 'br-int' with mock.patch('neutron.agent.common.ovs_lib.OVSBridge') as ovs_br: ovs = interface.OVSInterfaceDriver(self.conf) ovs.unplug('ns-0', bridge=bridge) ovs_br.assert_has_calls([mock.call(bridge), mock.call().delete_port('tap0')]) self.ip_dev.assert_has_calls([mock.call('ns-0', namespace=None), mock.call().link.delete()]) class TestBridgeInterfaceDriver(TestBase): def test_get_device_name(self): br = interface.BridgeInterfaceDriver(self.conf) device_name = br.get_device_name(FakePort()) self.assertEqual('ns-abcdef01-12', device_name) def test_plug_no_ns(self): self._test_plug() def test_plug_with_ns(self): self._test_plug(namespace='01234567-1234-1234-99') def _test_plug(self, namespace=None, mtu=None): def device_exists(device, namespace=None): return device.startswith('brq') root_veth = mock.Mock() ns_veth = mock.Mock() self.ip().add_veth = mock.Mock(return_value=(root_veth, ns_veth)) self.device_exists.side_effect = device_exists br = interface.BridgeInterfaceDriver(self.conf) mac_address = 'aa:bb:cc:dd:ee:ff' br.plug('01234567-1234-1234-99', 'port-1234', 'ns-0', mac_address, namespace=namespace) ip_calls = [mock.call(), mock.call().add_veth('tap0', 'ns-0', namespace2=namespace)] ns_veth.assert_has_calls([mock.call.link.set_address(mac_address)]) if mtu: ns_veth.assert_has_calls([mock.call.link.set_mtu(mtu)]) root_veth.assert_has_calls([mock.call.link.set_mtu(mtu)]) self.ip.assert_has_calls(ip_calls) root_veth.assert_has_calls([mock.call.link.set_up()]) ns_veth.assert_has_calls([mock.call.link.set_up()]) def test_plug_dev_exists(self): self.device_exists.return_value = True with mock.patch('neutron.agent.linux.interface.LOG.info') as log: br = interface.BridgeInterfaceDriver(self.conf) br.plug('01234567-1234-1234-99', 'port-1234', 'tap0', 'aa:bb:cc:dd:ee:ff') self.assertFalse(self.ip_dev.called) self.assertEqual(log.call_count, 1) def test_plug_mtu(self): self.device_exists.return_value = False self.conf.set_override('network_device_mtu', 9000) self._test_plug(mtu=9000) def test_unplug_no_device(self): self.device_exists.return_value = False self.ip_dev().link.delete.side_effect = RuntimeError with mock.patch('neutron.agent.linux.interface.LOG') as log: br = interface.BridgeInterfaceDriver(self.conf) br.unplug('tap0') [mock.call(), mock.call('tap0'), mock.call().link.delete()] self.assertEqual(log.error.call_count, 1) def test_unplug(self): self.device_exists.return_value = True with mock.patch('neutron.agent.linux.interface.LOG.debug') as log: br = interface.BridgeInterfaceDriver(self.conf) br.unplug('tap0') self.assertEqual(log.call_count, 1) self.ip_dev.assert_has_calls([mock.call('tap0', namespace=None), mock.call().link.delete()]) class TestMetaInterfaceDriver(TestBase): def setUp(self): super(TestMetaInterfaceDriver, self).setUp() config.register_interface_driver_opts_helper(self.conf) self.client_cls_p = mock.patch('neutronclient.v2_0.client.Client') client_cls = self.client_cls_p.start() self.client_inst = mock.Mock() client_cls.return_value = self.client_inst fake_network = {'network': {flavor.FLAVOR_NETWORK: 'fake1'}} fake_port = {'ports': [{'mac_address': 'aa:bb:cc:dd:ee:ffa', 'network_id': 'test'}]} self.client_inst.list_ports.return_value = fake_port self.client_inst.show_network.return_value = fake_network self.conf.set_override('auth_url', 'http://localhost:35357/v2.0') self.conf.set_override('auth_region', 'RegionOne') self.conf.set_override('admin_user', 'neutron') self.conf.set_override('admin_password', 'password') self.conf.set_override('admin_tenant_name', 'service') self.conf.set_override( 'meta_flavor_driver_mappings', 'fake1:neutron.agent.linux.interface.OVSInterfaceDriver,' 'fake2:neutron.agent.linux.interface.BridgeInterfaceDriver') self.conf.set_override('endpoint_type', 'internalURL') def test_get_driver_by_network_id(self): meta_interface = interface.MetaInterfaceDriver(self.conf) driver = meta_interface._get_driver_by_network_id('test') self.assertIsInstance(driver, interface.OVSInterfaceDriver) def test_set_device_plugin_tag(self): meta_interface = interface.MetaInterfaceDriver(self.conf) driver = meta_interface._get_driver_by_network_id('test') meta_interface._set_device_plugin_tag(driver, 'tap0', namespace=None) expected = [mock.call('tap0', namespace=None), mock.call().link.set_alias('fake1')] self.ip_dev.assert_has_calls(expected) namespace = '01234567-1234-1234-99' meta_interface._set_device_plugin_tag(driver, 'tap1', namespace=namespace) expected = [mock.call('tap1', namespace='01234567-1234-1234-99'), mock.call().link.set_alias('fake1')] self.ip_dev.assert_has_calls(expected) def test_get_device_plugin_tag(self): meta_interface = interface.MetaInterfaceDriver(self.conf) self.ip_dev().link.alias = 'fake1' plugin_tag0 = meta_interface._get_device_plugin_tag('tap0', namespace=None) expected = [mock.call('tap0', namespace=None)] self.ip_dev.assert_has_calls(expected) self.assertEqual('fake1', plugin_tag0) namespace = '01234567-1234-1234-99' expected = [mock.call('tap1', namespace='01234567-1234-1234-99')] plugin_tag1 = meta_interface._get_device_plugin_tag( 'tap1', namespace=namespace) self.ip_dev.assert_has_calls(expected) self.assertEqual('fake1', plugin_tag1) class TestIVSInterfaceDriver(TestBase): def setUp(self): super(TestIVSInterfaceDriver, self).setUp() def test_get_device_name(self): br = interface.IVSInterfaceDriver(self.conf) device_name = br.get_device_name(FakePort()) self.assertEqual('ns-abcdef01-12', device_name) def test_plug_with_prefix(self): self._test_plug(devname='qr-0', prefix='qr-') def _test_plug(self, devname=None, namespace=None, prefix=None, mtu=None): if not devname: devname = 'ns-0' def device_exists(dev, namespace=None): return dev == 'indigo' ivs = interface.IVSInterfaceDriver(self.conf) self.device_exists.side_effect = device_exists root_dev = mock.Mock() _ns_dev = mock.Mock() ns_dev = mock.Mock() self.ip().add_veth = mock.Mock(return_value=(root_dev, _ns_dev)) self.ip().device = mock.Mock(return_value=(ns_dev)) expected = [mock.call(), mock.call().add_veth('tap0', devname), mock.call().device(devname)] ivsctl_cmd = ['ivs-ctl', 'add-port', 'tap0'] with mock.patch.object(utils, 'execute') as execute: ivs.plug('01234567-1234-1234-99', 'port-1234', devname, 'aa:bb:cc:dd:ee:ff', namespace=namespace, prefix=prefix) execute.assert_called_once_with(ivsctl_cmd, run_as_root=True) ns_dev.assert_has_calls( [mock.call.link.set_address('aa:bb:cc:dd:ee:ff')]) if mtu: ns_dev.assert_has_calls([mock.call.link.set_mtu(mtu)]) root_dev.assert_has_calls([mock.call.link.set_mtu(mtu)]) if namespace: expected.extend( [mock.call().ensure_namespace(namespace), mock.call().ensure_namespace().add_device_to_namespace( mock.ANY)]) self.ip.assert_has_calls(expected) root_dev.assert_has_calls([mock.call.link.set_up()]) ns_dev.assert_has_calls([mock.call.link.set_up()]) def test_plug_mtu(self): self.conf.set_override('network_device_mtu', 9000) self._test_plug(mtu=9000) def test_plug_namespace(self): self._test_plug(namespace='mynamespace') def test_unplug(self): ivs = interface.IVSInterfaceDriver(self.conf) ivsctl_cmd = ['ivs-ctl', 'del-port', 'tap0'] with mock.patch.object(utils, 'execute') as execute: ivs.unplug('ns-0') execute.assert_called_once_with(ivsctl_cmd, run_as_root=True) self.ip_dev.assert_has_calls([mock.call('ns-0', namespace=None), mock.call().link.delete()]) class TestMidonetInterfaceDriver(TestBase): def setUp(self): self.conf = config.setup_conf() self.conf.register_opts(interface.OPTS) self.driver = interface.MidonetInterfaceDriver(self.conf) self.network_id = uuidutils.generate_uuid() self.port_id = uuidutils.generate_uuid() self.device_name = "tap0" self.mac_address = "aa:bb:cc:dd:ee:ff" self.bridge = "br-test" self.namespace = "ns-test" super(TestMidonetInterfaceDriver, self).setUp() def test_plug(self): cmd = ['mm-ctl', '--bind-port', self.port_id, 'tap0'] self.device_exists.return_value = False root_dev = mock.Mock() ns_dev = mock.Mock() self.ip().add_veth = mock.Mock(return_value=(root_dev, ns_dev)) with mock.patch.object(utils, 'execute') as execute: self.driver.plug( self.network_id, self.port_id, self.device_name, self.mac_address, self.bridge, self.namespace) execute.assert_called_once_with(cmd, run_as_root=True) expected = [mock.call(), mock.call(), mock.call().add_veth(self.device_name, self.device_name, namespace2=self.namespace), mock.call().ensure_namespace(self.namespace), mock.call().ensure_namespace().add_device_to_namespace( mock.ANY)] ns_dev.assert_has_calls( [mock.call.link.set_address(self.mac_address)]) root_dev.assert_has_calls([mock.call.link.set_up()]) ns_dev.assert_has_calls([mock.call.link.set_up()]) self.ip.assert_has_calls(expected, True) def test_unplug(self): self.driver.unplug(self.device_name, self.bridge, self.namespace) self.ip_dev.assert_has_calls([ mock.call(self.device_name, namespace=self.namespace), mock.call().link.delete()]) self.ip.assert_has_calls(mock.call().garbage_collect_namespace())