def _mock_ipdb_context_manager(self, netns):
mock_ipdb = mock.Mock(
interfaces={
'bridge': mock.Mock(
__enter__=mock.Mock(return_value=self.m_bridge_iface),
__exit__=mock.Mock(),
),
'c_interface': mock.Mock(
__enter__=mock.Mock(return_value=self.m_c_iface),
__exit__=mock.Mock(),
),
'h_interface': mock.Mock(
__enter__=mock.Mock(return_value=self.m_h_iface),
__exit__=mock.Mock(),
),
}
)
mock_exit = mock.Mock()
self.ipdbs[netns] = mock.Mock(
__enter__=mock.Mock(return_value=mock_ipdb),
__exit__=mock_exit)
return mock_ipdb, mock_exit
python类Mock()的实例源码
def test_watch_removed(self):
path = '/test'
events = [{'e': i} for i in range(3)]
def handler(event):
self.assertFalse(watcher_obj._idle[path])
watcher_obj._resources.remove(path)
m_handler = mock.Mock()
m_handler.side_effect = handler
watcher_obj = self._test_watch_create_watcher(path, m_handler)
self._test_watch_mock_events(watcher_obj, events)
watcher_obj._watch(path)
m_handler.assert_called_once_with(events[0])
self.assertNotIn(path, watcher_obj._idle)
self.assertNotIn(path, watcher_obj._watching)
def test_watch_interrupted(self):
path = '/test'
events = [{'e': i} for i in range(3)]
def handler(event):
self.assertFalse(watcher_obj._idle[path])
raise greenlet.GreenletExit()
m_handler = mock.Mock()
m_handler.side_effect = handler
watcher_obj = self._test_watch_create_watcher(path, m_handler)
self._test_watch_mock_events(watcher_obj, events)
self.assertRaises(greenlet.GreenletExit, watcher_obj._watch, path)
m_handler.assert_called_once_with(events[0])
self.assertNotIn(path, watcher_obj._idle)
self.assertNotIn(path, watcher_obj._watching)
def test_neutron_to_osvif_vif_load(self, m_mgrs, m_stv_drm):
vif_plugin = mock.sentinel.vif_plugin
port = mock.sentinel.port
subnets = mock.sentinel.subnets
m_mgr = mock.Mock()
m_mgrs.__getitem__.side_effect = KeyError
m_stv_drm.return_value = m_mgr
ovu.neutron_to_osvif_vif(vif_plugin, port, subnets)
m_stv_drm.assert_called_once_with(
namespace=ovu._VIF_TRANSLATOR_NAMESPACE,
name=vif_plugin,
invoke_on_load=False)
m_mgrs.__setitem__.assert_called_once_with(vif_plugin, m_mgr)
m_mgr.driver.assert_called_once_with(vif_plugin, port, subnets)
def test_make_vif_network(self, m_mk_subnet_list, m_make_vif_subnets):
network_id = mock.sentinel.network_id
network = mock.Mock()
orig_network = mock.Mock()
orig_network.id = network_id
orig_network.obj_clone.return_value = network
subnet_id = mock.sentinel.subnet_id
subnets = {subnet_id: orig_network}
vif_subnets = mock.sentinel.vif_subnets
subnet_list = mock.sentinel.subnet_list
m_make_vif_subnets.return_value = vif_subnets
m_mk_subnet_list.return_value = subnet_list
port = {'network_id': network_id}
self.assertEqual(network, ovu._make_vif_network(port, subnets))
self.assertEqual(subnet_list, network.subnets)
m_make_vif_subnets.assert_called_once_with(port, subnets)
m_mk_subnet_list.assert_called_once_with(objects=vif_subnets)
def test_get_parent_port(self):
cls = nested_vif.NestedPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
node_fixed_ip = mock.sentinel.node_fixed_ip
pod_status = mock.MagicMock()
pod_status.__getitem__.return_value = node_fixed_ip
pod = mock.MagicMock()
pod.__getitem__.return_value = pod_status
parent_port = mock.sentinel.parent_port
m_driver._get_parent_port_by_host_ip.return_value = parent_port
cls._get_parent_port(m_driver, neutron, pod)
m_driver._get_parent_port_by_host_ip.assert_called_once()
def test_get_parent_port_by_host_ip_trunk_not_found(self):
cls = nested_vif.NestedPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
node_subnet_id = mock.sentinel.node_subnet_id
oslo_cfg.CONF.set_override('worker_nodes_subnet',
node_subnet_id,
group='pod_vif_nested')
node_fixed_ip = mock.sentinel.node_fixed_ip
ports = {'ports': []}
neutron.list_ports.return_value = ports
self.assertRaises(kl_exc.NoResourceException,
cls._get_parent_port_by_host_ip, m_driver, neutron,
node_fixed_ip)
fixed_ips = ['subnet_id=%s' % str(node_subnet_id),
'ip_address=%s' % str(node_fixed_ip)]
neutron.list_ports.assert_called_once_with(fixed_ips=fixed_ips)
def test_release_vif(self):
cls = nested_vlan_vif.NestedVlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
parent_port = mock.sentinel.parent_port
trunk_id = mock.sentinel.trunk_id
m_driver._get_parent_port.return_value = parent_port
m_driver._get_trunk_id.return_value = trunk_id
pod = mock.sentinel.pod
vif = mock.Mock()
cls.release_vif(m_driver, pod, vif)
m_driver._get_parent_port.assert_called_once_with(neutron, pod)
m_driver._get_trunk_id.assert_called_once_with(parent_port)
m_driver._remove_subport.assert_called_once_with(
neutron, trunk_id, vif.id)
neutron.delete_port.assert_called_once_with(vif.id)
def test_release_vif_not_found(self):
cls = nested_vlan_vif.NestedVlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
parent_port = mock.sentinel.parent_port
trunk_id = mock.sentinel.trunk_id
m_driver._get_parent_port.return_value = parent_port
m_driver._get_trunk_id.return_value = trunk_id
pod = mock.sentinel.pod
vlan_id = mock.sentinel.vlan_id
vif = mock.Mock()
m_driver._port_vlan_mapping = {vif.id: vlan_id}
self.assertTrue(vif.id in m_driver._port_vlan_mapping.keys())
neutron.delete_port.side_effect = n_exc.PortNotFoundClient
cls.release_vif(m_driver, pod, vif)
m_driver._get_parent_port.assert_called_once_with(neutron, pod)
m_driver._get_trunk_id.assert_called_once_with(parent_port)
m_driver._remove_subport.assert_called_once_with(
neutron, trunk_id, vif.id)
neutron.delete_port.assert_called_once_with(vif.id)
def test_add_subport(self):
cls = nested_vlan_vif.NestedVlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
trunk_id = mock.sentinel.trunk_id
subport = mock.sentinel.subport
vlan_id = mock.sentinel.vlan_id
m_driver._get_vlan_id.return_value = vlan_id
subport_dict = [{'segmentation_id': vlan_id,
'port_id': subport,
'segmentation_type': 'vlan'}]
nested_vlan_vif.DEFAULT_MAX_RETRY_COUNT = 1
self.assertEqual(vlan_id, cls._add_subport(m_driver,
neutron, trunk_id, subport))
m_driver._get_vlan_id.assert_called_once_with(trunk_id)
neutron.trunk_add_subports.assert_called_once_with(
trunk_id, {'sub_ports': subport_dict})
def test_add_subport_with_vlan_id_conflict(self):
cls = nested_vlan_vif.NestedVlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
trunk_id = mock.sentinel.trunk_id
subport = mock.sentinel.subport
vlan_id = mock.sentinel.vlan_id
m_driver._get_vlan_id.return_value = vlan_id
subport_dict = [{'segmentation_id': vlan_id,
'port_id': subport,
'segmentation_type': 'vlan'}]
neutron.trunk_add_subports.side_effect = n_exc.Conflict
nested_vlan_vif.DEFAULT_MAX_RETRY_COUNT = 1
self.assertRaises(n_exc.Conflict, cls._add_subport, m_driver,
neutron, trunk_id, subport)
neutron.trunk_add_subports.assert_called_once_with(
trunk_id, {'sub_ports': subport_dict})
def test_activate_vif(self):
cls = neutron_vif.NeutronPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
pod = mock.sentinel.pod
vif = mock.Mock()
vif.active = False
port = mock.MagicMock()
port.__getitem__.return_value = kl_const.PORT_STATUS_ACTIVE
neutron.show_port.return_value = {'port': port}
cls.activate_vif(m_driver, pod, vif)
neutron.show_port.assert_called_once_with(vif.id)
self.assertTrue(vif.active)
def test_release_vif_parent_not_found(self):
cls = nested_macvlan_vif.NestedMacvlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
port_id = lib_utils.get_hash()
pod = mock.sentinel.pod
vif = mock.Mock()
vif.id = port_id
container_mac = mock.sentinel.mac_address
container_ip = mock.sentinel.ip_address
container_port = self._get_fake_port(port_id, container_ip,
container_mac)
neutron.show_port.return_value = container_port
m_driver.lock = mock.MagicMock(spec=threading.Lock())
m_driver._get_parent_port.side_effect = n_exc.NeutronClientException
self.assertRaises(n_exc.NeutronClientException, cls.release_vif,
m_driver, pod, vif)
neutron.show_port.assert_called_once_with(port_id)
m_driver._get_parent_port.assert_called_once_with(neutron, pod)
m_driver._remove_from_allowed_address_pairs.assert_not_called()
neutron.delete_port.assert_not_called()
def test_add_to_allowed_address_pairs_same_ip(self):
cls = nested_macvlan_vif.NestedMacvlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
port_id = lib_utils.get_hash()
vm_port = self._get_fake_port(port_id)['port']
address_pairs = [
{'ip_address': '10.0.0.30',
'mac_address': 'fa:16:3e:1b:30:00'},
{'ip_address': 'fe80::f816:3eff:fe1c:36a9',
'mac_address': 'fa:16:3e:1b:30:00'},
]
vm_port['allowed_address_pairs'].extend(address_pairs)
mac_addr = 'fa:16:3e:71:cb:80'
ip_addr = '10.0.0.30'
address_pairs.append({'ip_address': ip_addr, 'mac_address': mac_addr})
cls._add_to_allowed_address_pairs(m_driver, neutron, vm_port,
frozenset([ip_addr]), mac_addr)
m_driver._update_port_address_pairs.assert_called_once_with(
neutron, port_id, address_pairs)
def test_add_to_allowed_address_pairs_already_present(self):
cls = nested_macvlan_vif.NestedMacvlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
port_id = lib_utils.get_hash()
vm_port = self._get_fake_port(port_id)['port']
address_pairs = [
{'ip_address': '10.0.0.30',
'mac_address': 'fa:16:3e:1b:30:00'},
{'ip_address': 'fe80::f816:3eff:fe1c:36a9',
'mac_address': 'fa:16:3e:1b:30:00'},
]
vm_port['allowed_address_pairs'].extend(address_pairs)
mac_addr = 'fa:16:3e:1b:30:00'
ip_addr = '10.0.0.30'
self.assertRaises(k_exc.AllowedAddressAlreadyPresent,
cls._add_to_allowed_address_pairs, m_driver, neutron,
vm_port, frozenset([ip_addr]), mac_addr)
def test_remove_from_allowed_address_pairs_no_update(self, m_mac):
cls = nested_macvlan_vif.NestedMacvlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
port_id = lib_utils.get_hash()
vm_port = self._get_fake_port(port_id)['port']
mac_addr = 'fa:16:3e:1b:30:00' if m_mac else vm_port['mac_address']
address_pairs = [
{'ip_address': '10.0.0.30',
'mac_address': mac_addr},
{'ip_address': 'fe80::f816:3eff:fe1c:36a9',
'mac_address': mac_addr},
]
vm_port['allowed_address_pairs'].extend(address_pairs)
ip_addr = ['10.0.0.29']
cls._remove_from_allowed_address_pairs(
m_driver, neutron, vm_port, frozenset(ip_addr), m_mac)
m_driver._update_port_address_pairs.assert_not_called()
def test_update_port_address_pairs_failure(self):
cls = nested_macvlan_vif.NestedMacvlanPodVIFDriver
m_driver = mock.Mock(spec=cls)
neutron = self.useFixture(k_fix.MockNeutronClient()).client
port_id = lib_utils.get_hash()
pairs = mock.sentinel.allowed_address_pairs
neutron.update_port.side_effect = n_exc.NeutronClientException
self.assertRaises(n_exc.NeutronClientException,
cls._update_port_address_pairs, m_driver, neutron,
port_id, pairs)
neutron.update_port.assert_called_with(
port_id,
{'port': {'allowed_address_pairs': pairs}})
# TODO(garyloug) consider exending and moving to a parent class
def test_ensure_loadbalancer(self):
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
expected_resp = mock.sentinel.expected_resp
namespace = 'TEST_NAMESPACE'
name = 'TEST_NAME'
project_id = 'TEST_PROJECT'
subnet_id = 'D3FA400A-F543-4B91-9CD3-047AF0CE42D1'
ip = '1.2.3.4'
# TODO(ivc): handle security groups
sg_ids = []
endpoints = {'metadata': {'namespace': namespace, 'name': name}}
m_driver._ensure.return_value = expected_resp
resp = cls.ensure_loadbalancer(m_driver, endpoints, project_id,
subnet_id, ip, sg_ids)
m_driver._ensure.assert_called_once_with(mock.ANY,
m_driver._create_loadbalancer,
m_driver._find_loadbalancer)
req = m_driver._ensure.call_args[0][0]
self.assertEqual("%s/%s" % (namespace, name), req.name)
self.assertEqual(project_id, req.project_id)
self.assertEqual(subnet_id, req.subnet_id)
self.assertEqual(ip, str(req.ip))
self.assertEqual(expected_resp, resp)
def test_ensure_pool(self):
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
expected_resp = mock.sentinel.expected_resp
endpoints = mock.sentinel.endpoints
loadbalancer = obj_lbaas.LBaaSLoadBalancer(
id='00EE9E11-91C2-41CF-8FD4-7970579E5C4C',
project_id='TEST_PROJECT')
listener = obj_lbaas.LBaaSListener(
id='A57B7771-6050-4CA8-A63C-443493EC98AB',
name='TEST_LISTENER_NAME',
protocol='TCP')
m_driver._ensure_provisioned.return_value = expected_resp
resp = cls.ensure_pool(m_driver, endpoints, loadbalancer, listener)
m_driver._ensure_provisioned.assert_called_once_with(
loadbalancer, mock.ANY, m_driver._create_pool,
m_driver._find_pool)
pool = m_driver._ensure_provisioned.call_args[0][1]
self.assertEqual(listener.name, pool.name)
self.assertEqual(loadbalancer.project_id, pool.project_id)
self.assertEqual(listener.id, pool.listener_id)
self.assertEqual(listener.protocol, pool.protocol)
self.assertEqual(expected_resp, resp)
def test_create_loadbalancer(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
loadbalancer = obj_lbaas.LBaaSLoadBalancer(
name='TEST_NAME', project_id='TEST_PROJECT', ip='1.2.3.4',
subnet_id='D3FA400A-F543-4B91-9CD3-047AF0CE42D1')
loadbalancer_id = '00EE9E11-91C2-41CF-8FD4-7970579E5C4C'
req = {'loadbalancer': {
'name': loadbalancer.name,
'project_id': loadbalancer.project_id,
'tenant_id': loadbalancer.project_id,
'vip_address': str(loadbalancer.ip),
'vip_subnet_id': loadbalancer.subnet_id}}
resp = {'loadbalancer': {'id': loadbalancer_id}}
neutron.create_loadbalancer.return_value = resp
ret = cls._create_loadbalancer(m_driver, loadbalancer)
neutron.create_loadbalancer.assert_called_once_with(req)
for attr in loadbalancer.obj_fields:
self.assertEqual(getattr(loadbalancer, attr),
getattr(ret, attr))
self.assertEqual(loadbalancer_id, ret.id)
def test_find_loadbalancer(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
loadbalancer = obj_lbaas.LBaaSLoadBalancer(
name='TEST_NAME', project_id='TEST_PROJECT', ip='1.2.3.4',
subnet_id='D3FA400A-F543-4B91-9CD3-047AF0CE42D1')
loadbalancer_id = '00EE9E11-91C2-41CF-8FD4-7970579E5C4C'
resp = {'loadbalancers': [{'id': loadbalancer_id}]}
neutron.list_loadbalancers.return_value = resp
ret = cls._find_loadbalancer(m_driver, loadbalancer)
neutron.list_loadbalancers.assert_called_once_with(
name=loadbalancer.name,
project_id=loadbalancer.project_id,
tenant_id=loadbalancer.project_id,
vip_address=str(loadbalancer.ip),
vip_subnet_id=loadbalancer.subnet_id)
for attr in loadbalancer.obj_fields:
self.assertEqual(getattr(loadbalancer, attr),
getattr(ret, attr))
self.assertEqual(loadbalancer_id, ret.id)
def test_find_loadbalancer_not_found(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
loadbalancer = obj_lbaas.LBaaSLoadBalancer(
name='TEST_NAME', project_id='TEST_PROJECT', ip='1.2.3.4',
subnet_id='D3FA400A-F543-4B91-9CD3-047AF0CE42D1')
resp = {'loadbalancers': []}
neutron.list_loadbalancers.return_value = resp
ret = cls._find_loadbalancer(m_driver, loadbalancer)
neutron.list_loadbalancers.assert_called_once_with(
name=loadbalancer.name,
project_id=loadbalancer.project_id,
tenant_id=loadbalancer.project_id,
vip_address=str(loadbalancer.ip),
vip_subnet_id=loadbalancer.subnet_id)
self.assertIsNone(ret)
def test_create_listener(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
listener = obj_lbaas.LBaaSListener(
name='TEST_NAME', project_id='TEST_PROJECT', protocol='TCP',
port=1234, loadbalancer_id='00EE9E11-91C2-41CF-8FD4-7970579E5C4C')
listener_id = 'A57B7771-6050-4CA8-A63C-443493EC98AB'
req = {'listener': {
'name': listener.name,
'project_id': listener.project_id,
'tenant_id': listener.project_id,
'loadbalancer_id': listener.loadbalancer_id,
'protocol': listener.protocol,
'protocol_port': listener.port}}
resp = {'listener': {'id': listener_id}}
neutron.create_listener.return_value = resp
ret = cls._create_listener(m_driver, listener)
neutron.create_listener.assert_called_once_with(req)
for attr in listener.obj_fields:
self.assertEqual(getattr(listener, attr),
getattr(ret, attr))
self.assertEqual(listener_id, ret.id)
def test_find_listener_not_found(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
listener = obj_lbaas.LBaaSListener(
name='TEST_NAME', project_id='TEST_PROJECT', protocol='TCP',
port=1234, loadbalancer_id='00EE9E11-91C2-41CF-8FD4-7970579E5C4C')
resp = {'listeners': []}
neutron.list_listeners.return_value = resp
ret = cls._find_listener(m_driver, listener)
neutron.list_listeners.assert_called_once_with(
name=listener.name,
project_id=listener.project_id,
tenant_id=listener.project_id,
loadbalancer_id=listener.loadbalancer_id,
protocol=listener.protocol,
protocol_port=listener.port)
self.assertIsNone(ret)
def test_create_pool_conflict(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
lb_algorithm = 'ROUND_ROBIN'
pool = obj_lbaas.LBaaSPool(
name='TEST_NAME', project_id='TEST_PROJECT', protocol='TCP',
listener_id='A57B7771-6050-4CA8-A63C-443493EC98AB',
loadbalancer_id='00EE9E11-91C2-41CF-8FD4-7970579E5C4C')
req = {'pool': {
'name': pool.name,
'project_id': pool.project_id,
'tenant_id': pool.project_id,
'listener_id': pool.listener_id,
'loadbalancer_id': pool.loadbalancer_id,
'protocol': pool.protocol,
'lb_algorithm': lb_algorithm}}
neutron.create_lbaas_pool.side_effect = n_exc.StateInvalidClient
self.assertRaises(n_exc.StateInvalidClient, cls._create_pool, m_driver,
pool)
neutron.create_lbaas_pool.assert_called_once_with(req)
m_driver._cleanup_bogus_pool.assert_called_once_with(neutron, pool,
lb_algorithm)
def test_find_pool(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
pool = obj_lbaas.LBaaSPool(
name='TEST_NAME', project_id='TEST_PROJECT', protocol='TCP',
listener_id='A57B7771-6050-4CA8-A63C-443493EC98AB',
loadbalancer_id='00EE9E11-91C2-41CF-8FD4-7970579E5C4C')
pool_id = 'D4F35594-27EB-4F4C-930C-31DD40F53B77'
resp = {'pools': [{'id': pool_id,
'listeners': [{'id': pool.listener_id}]}]}
neutron.list_lbaas_pools.return_value = resp
ret = cls._find_pool(m_driver, pool)
neutron.list_lbaas_pools.assert_called_once_with(
name=pool.name,
project_id=pool.project_id,
tenant_id=pool.project_id,
loadbalancer_id=pool.loadbalancer_id,
protocol=pool.protocol)
for attr in pool.obj_fields:
self.assertEqual(getattr(pool, attr),
getattr(ret, attr))
self.assertEqual(pool_id, ret.id)
def test_find_pool_not_found(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
pool = obj_lbaas.LBaaSPool(
name='TEST_NAME', project_id='TEST_PROJECT', protocol='TCP',
listener_id='A57B7771-6050-4CA8-A63C-443493EC98AB',
loadbalancer_id='00EE9E11-91C2-41CF-8FD4-7970579E5C4C')
resp = {'pools': []}
neutron.list_lbaas_pools.return_value = resp
ret = cls._find_pool(m_driver, pool)
neutron.list_lbaas_pools.assert_called_once_with(
name=pool.name,
project_id=pool.project_id,
tenant_id=pool.project_id,
loadbalancer_id=pool.loadbalancer_id,
protocol=pool.protocol)
self.assertIsNone(ret)
def test_find_member(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
member = obj_lbaas.LBaaSMember(
name='TEST_NAME', project_id='TEST_PROJECT', ip='1.2.3.4',
port=1234, subnet_id='D3FA400A-F543-4B91-9CD3-047AF0CE42D1',
pool_id='D4F35594-27EB-4F4C-930C-31DD40F53B77')
member_id = '3A70CEC0-392D-4BC1-A27C-06E63A0FD54F'
resp = {'members': [{'id': member_id}]}
neutron.list_lbaas_members.return_value = resp
ret = cls._find_member(m_driver, member)
neutron.list_lbaas_members.assert_called_once_with(
member.pool_id,
name=member.name,
project_id=member.project_id,
tenant_id=member.project_id,
subnet_id=member.subnet_id,
address=member.ip,
protocol_port=member.port)
for attr in member.obj_fields:
self.assertEqual(getattr(member, attr),
getattr(ret, attr))
self.assertEqual(member_id, ret.id)
def test_find_member_not_found(self):
neutron = self.useFixture(k_fix.MockNeutronClient()).client
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
member = obj_lbaas.LBaaSMember(
name='TEST_NAME', project_id='TEST_PROJECT', ip='1.2.3.4',
port=1234, subnet_id='D3FA400A-F543-4B91-9CD3-047AF0CE42D1',
pool_id='D4F35594-27EB-4F4C-930C-31DD40F53B77')
resp = {'members': []}
neutron.list_lbaas_members.return_value = resp
ret = cls._find_member(m_driver, member)
neutron.list_lbaas_members.assert_called_once_with(
member.pool_id,
name=member.name,
project_id=member.project_id,
tenant_id=member.project_id,
subnet_id=member.subnet_id,
address=member.ip,
protocol_port=member.port)
self.assertIsNone(ret)
def test_request(self):
cls = d_lbaasv2.LBaaSv2Driver
m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
loadbalancer = mock.sentinel.loadbalancer
obj = mock.sentinel.obj
create = mock.sentinel.create
find = mock.sentinel.find
expected_result = mock.sentinel.expected_result
timer = [mock.sentinel.t0, mock.sentinel.t1]
m_driver._provisioning_timer.return_value = timer
m_driver._ensure.side_effect = [n_exc.StateInvalidClient,
expected_result]
ret = cls._ensure_provisioned(m_driver, loadbalancer, obj, create,
find)
m_driver._wait_for_provisioning.assert_has_calls(
[mock.call(loadbalancer, t) for t in timer])
m_driver._ensure.assert_has_calls(
[mock.call(obj, create, find) for _ in timer])
self.assertEqual(expected_result, ret)
