我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用mock.Mock()。
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
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)
def test_release_not_ready(self): cls = d_lbaasv2.LBaaSv2Driver m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver) loadbalancer = mock.sentinel.loadbalancer obj = mock.sentinel.obj m_delete = mock.Mock() timer = [mock.sentinel.t0, mock.sentinel.t1] m_driver._provisioning_timer.return_value = timer m_delete.side_effect = n_exc.StateInvalidClient self.assertRaises(k_exc.ResourceNotReady, cls._release, m_driver, loadbalancer, obj, m_delete) call_count = len(timer) self.assertEqual(call_count, m_driver._wait_for_provisioning.call_count) self.assertEqual(call_count, m_delete.call_count)
def test_acquire_service_pub_ip_info_usr_specified_ip(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client floating_ip = {'floating_ip_address': '1.2.3.4', 'port_id': None, 'id': 'a2a62ea7-e3bf-40df-8c09-aa0c29876a6b'} neutron.list_floatingips.return_value = {'floatingips': [floating_ip]} project_id = mock.sentinel.project_id spec_type = 'LoadBalancer' spec_lb_ip = '1.2.3.4' expected_resp = \ obj_lbaas.LBaaSPubIp(ip_id=floating_ip['id'], ip_addr=floating_ip['floating_ip_address'], alloc_method='user') self.assertEqual(expected_resp, cls.acquire_service_pub_ip_info (m_driver, spec_type, spec_lb_ip, project_id))
def test_acquire_service_pub_ip_info_user_specified_non_exist_fip(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client floating_ip = {'floating_ip_address': '1.2.3.5', 'port_id': None} neutron.list_floatingips.return_value = {'floatingips': [floating_ip]} project_id = mock.sentinel.project_id spec_type = 'LoadBalancer' spec_lb_ip = '1.2.3.4' self.assertIsNone(cls.acquire_service_pub_ip_info (m_driver, spec_type, spec_lb_ip, project_id))
def test_acquire_service_pub_ip_info_user_specified_occupied_fip(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client floating_ip = {'floating_ip_address': '1.2.3.4', 'port_id': 'ec29d641-fec4-4f67-928a-124a76b3a8e6'} neutron.list_floatingips.return_value = {'floatingips': [floating_ip]} project_id = mock.sentinel.project_id spec_type = 'LoadBalancer' spec_lb_ip = '1.2.3.4' self.assertIsNone(cls.acquire_service_pub_ip_info (m_driver, spec_type, spec_lb_ip, project_id))
def test_acquire_service_pub_ip_info_alloc_from_pool(self, m_cfg): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client m_cfg.neutron_defaults.external_svc_subnet =\ mock.sentinel.external_svc_subnet neutron.show_subnet.return_value =\ {'subnet': {'network_id': 'ec29d641-fec4-4f67-928a-124a76b3a8e6'}} floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} neutron.create_floatingip.return_value = {'floatingip': floating_ip} project_id = mock.sentinel.project_id spec_type = 'LoadBalancer' spec_lb_ip = None expected_resp = \ obj_lbaas.LBaaSPubIp(ip_id=floating_ip['id'], ip_addr=floating_ip['floating_ip_address'], alloc_method='pool') self.assertEqual(expected_resp, cls.acquire_service_pub_ip_info (m_driver, spec_type, spec_lb_ip, project_id))
def test_release_pub_ip_alloc_method_pool_neutron_exception(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client neutron.delete_floatingip.side_effect = n_exc.NeutronClientException floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} service_pub_ip_info = \ obj_lbaas.LBaaSPubIp(ip_id=floating_ip['id'], ip_addr=floating_ip['floating_ip_address'], alloc_method='pool') rc = cls.release_pub_ip(m_driver, service_pub_ip_info) self.assertEqual(rc, False)
def test_release_pub_ip_alloc_method_pool_neutron_succeeded(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client neutron.delete_floatingip.return_value = None floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} service_pub_ip_info = \ obj_lbaas.LBaaSPubIp(ip_id=floating_ip['id'], ip_addr=floating_ip['floating_ip_address'], alloc_method='pool') rc = cls.release_pub_ip(m_driver, service_pub_ip_info) self.assertEqual(rc, True)
def test_associate_lb_fip_id_not_exist_neutron_exception(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client neutron.update_floatingip.side_effect = n_exc.NeutronClientException floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} service_pub_ip_info = \ obj_lbaas.LBaaSPubIp(ip_id=floating_ip['id'], ip_addr=floating_ip['floating_ip_address'], alloc_method='pool') vip_port_id = 'ec29d641-fec4-4f67-928a-124a76b3a777' self.assertRaises( n_exc.NeutronClientException, cls.associate_pub_ip, m_driver, service_pub_ip_info, vip_port_id)
def test_disassociate_pub_ip_fip_id_not_exist(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client neutron.update_floatingip.return_value = None floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} service_pub_ip_info = \ obj_lbaas.LBaaSPubIp(ip_id=0, ip_addr=floating_ip['floating_ip_address'], alloc_method='pool') self.assertIsNone(cls.disassociate_pub_ip (m_driver, service_pub_ip_info))
def test_disassociate_pub_ip_neutron_exception(self): cls = d_lb_public_ip.FloatingIpServicePubIPDriver m_driver = mock.Mock(spec=cls) m_driver._drv_pub_ip = public_ip.FipPubIpDriver() neutron = self.useFixture(k_fix.MockNeutronClient()).client neutron.update_floatingip.side_effect = n_exc.NeutronClientException floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} service_pub_ip_info = \ obj_lbaas.LBaaSPubIp(ip_id=floating_ip['id'], ip_addr=floating_ip['floating_ip_address'], alloc_method='pool') self.assertRaises( n_exc.NeutronClientException, cls.disassociate_pub_ip, m_driver, service_pub_ip_info)
def test_allocate_ip_all_green(self): cls = d_public_ip.FipPubIpDriver m_driver = mock.Mock(spec=cls) pub_net_id = mock.sentinel.pub_net_id pub_subnet_id = mock.sentinel.pub_subnet_id project_id = mock.sentinel.project_id description = mock.sentinel.description neutron = self.useFixture(k_fix.MockNeutronClient()).client floating_ip = {'floating_ip_address': '1.2.3.5', 'id': 'ec29d641-fec4-4f67-928a-124a76b3a888'} neutron.create_floatingip.return_value = {'floatingip': floating_ip} fip_id, fip_addr = cls.allocate_ip( m_driver, pub_net_id, pub_subnet_id, project_id, description) self.assertEqual(fip_id, floating_ip['id']) self.assertEqual(fip_addr, floating_ip['floating_ip_address'])
def test_on_present(self): svc_event = mock.sentinel.svc_event old_spec = mock.sentinel.old_spec new_spec = mock.sentinel.new_spec project_id = mock.sentinel.project_id m_drv_project = mock.Mock() m_drv_project.get_project.return_value = project_id m_handler = mock.Mock(spec=h_lbaas.LBaaSSpecHandler) m_handler._get_lbaas_spec.return_value = old_spec m_handler._has_lbaas_spec_changes.return_value = True m_handler._generate_lbaas_spec.return_value = new_spec m_handler._should_ignore.return_value = False m_handler._drv_project = m_drv_project h_lbaas.LBaaSSpecHandler.on_present(m_handler, svc_event) m_handler._get_lbaas_spec.assert_called_once_with(svc_event) m_handler._has_lbaas_spec_changes.assert_called_once_with(svc_event, old_spec) m_handler._generate_lbaas_spec.assert_called_once_with(svc_event) m_handler._set_lbaas_spec.assert_called_once_with(svc_event, new_spec)
def test_on_present_no_changes(self): svc_event = mock.sentinel.svc_event old_spec = mock.sentinel.old_spec m_handler = mock.Mock(spec=h_lbaas.LBaaSSpecHandler) m_handler._get_lbaas_spec.return_value = old_spec m_handler._has_lbaas_spec_changes.return_value = False m_handler._should_ignore.return_value = False h_lbaas.LBaaSSpecHandler.on_present(m_handler, svc_event) m_handler._get_lbaas_spec.assert_called_once_with(svc_event) m_handler._has_lbaas_spec_changes.assert_called_once_with(svc_event, old_spec) m_handler._generate_lbaas_spec.assert_not_called() m_handler._set_lbaas_spec.assert_not_called()
def test_has_port_changes(self): m_handler = mock.Mock(spec=h_lbaas.LBaaSSpecHandler) m_service = mock.MagicMock() m_handler._get_service_ports.return_value = [ {'port': 1, 'name': 'X', 'protocol': 'TCP'}, ] m_lbaas_spec = mock.MagicMock() m_lbaas_spec.ports = [ obj_lbaas.LBaaSPortSpec(name='X', protocol='TCP', port=1), obj_lbaas.LBaaSPortSpec(name='Y', protocol='TCP', port=2), ] ret = h_lbaas.LBaaSSpecHandler._has_port_changes( m_handler, m_service, m_lbaas_spec) self.assertTrue(ret)
def test_generate_lbaas_port_specs(self): m_handler = mock.Mock(spec=h_lbaas.LBaaSSpecHandler) m_handler._get_service_ports.return_value = [ {'port': 1, 'name': 'X', 'protocol': 'TCP'}, {'port': 2, 'name': 'Y', 'protocol': 'TCP'} ] expected_ports = [ obj_lbaas.LBaaSPortSpec(name='X', protocol='TCP', port=1), obj_lbaas.LBaaSPortSpec(name='Y', protocol='TCP', port=2), ] ret = h_lbaas.LBaaSSpecHandler._generate_lbaas_port_specs( m_handler, mock.sentinel.service) self.assertEqual(expected_ports, ret) m_handler._get_service_ports.assert_called_once_with( mock.sentinel.service)
def test_on_present(self): lbaas_spec = mock.sentinel.lbaas_spec lbaas_state = mock.sentinel.lbaas_state endpoints = mock.sentinel.endpoints m_handler = mock.Mock(spec=h_lbaas.LoadBalancerHandler) m_handler._get_lbaas_spec.return_value = lbaas_spec m_handler._should_ignore.return_value = False m_handler._get_lbaas_state.return_value = lbaas_state m_handler._sync_lbaas_members.return_value = True h_lbaas.LoadBalancerHandler.on_present(m_handler, endpoints) m_handler._get_lbaas_spec.assert_called_once_with(endpoints) m_handler._should_ignore.assert_called_once_with(endpoints, lbaas_spec) m_handler._get_lbaas_state.assert_called_once_with(endpoints) m_handler._sync_lbaas_members.assert_called_once_with( endpoints, lbaas_state, lbaas_spec) m_handler._set_lbaas_state.assert_called_once_with( endpoints, lbaas_state)
def test_should_ignore(self): endpoints = mock.sentinel.endpoints lbaas_spec = mock.sentinel.lbaas_spec # REVISIT(ivc): ddt? m_handler = mock.Mock(spec=h_lbaas.LoadBalancerHandler) m_handler._has_pods.return_value = True m_handler._is_lbaas_spec_in_sync.return_value = True ret = h_lbaas.LoadBalancerHandler._should_ignore( m_handler, endpoints, lbaas_spec) self.assertEqual(False, ret) m_handler._has_pods.assert_called_once_with(endpoints) m_handler._is_lbaas_spec_in_sync.assert_called_once_with( endpoints, lbaas_spec)
def test_get_pod_subnet(self): subnet_id = mock.sentinel.subnet_id project_id = mock.sentinel.project_id target_ref = {'kind': k_const.K8S_OBJ_POD, 'name': 'pod-name', 'namespace': 'default'} ip = '1.2.3.4' m_handler = mock.Mock(spec=h_lbaas.LoadBalancerHandler) m_drv_pod_project = mock.Mock() m_drv_pod_project.get_project.return_value = project_id m_handler._drv_pod_project = m_drv_pod_project m_drv_pod_subnets = mock.Mock() m_drv_pod_subnets.get_subnets.return_value = { subnet_id: osv_network.Network(subnets=osv_subnet.SubnetList( objects=[osv_subnet.Subnet(cidr='1.2.3.0/24')]))} m_handler._drv_pod_subnets = m_drv_pod_subnets observed_subnet_id = h_lbaas.LoadBalancerHandler._get_pod_subnet( m_handler, target_ref, ip) self.assertEqual(subnet_id, observed_subnet_id)
def test_init(self, m_get_project_driver, m_get_subnets_driver, m_get_sg_driver, m_get_vif_driver, m_get_vif_pool_driver, m_set_vif_driver): project_driver = mock.sentinel.project_driver subnets_driver = mock.sentinel.subnets_driver sg_driver = mock.sentinel.sg_driver vif_driver = mock.sentinel.vif_driver vif_pool_driver = mock.Mock(spec=drivers.VIFPoolDriver) m_get_project_driver.return_value = project_driver m_get_subnets_driver.return_value = subnets_driver m_get_sg_driver.return_value = sg_driver m_get_vif_driver.return_value = vif_driver m_get_vif_pool_driver.return_value = vif_pool_driver handler = h_vif.VIFHandler() self.assertEqual(project_driver, handler._drv_project) self.assertEqual(subnets_driver, handler._drv_subnets) self.assertEqual(sg_driver, handler._drv_sg) self.assertEqual(vif_pool_driver, handler._drv_vif_pool)
def test_register(self, m_init, m_wrap_consumer): consumes = {mock.sentinel.key_fn1: mock.sentinel.key1, mock.sentinel.key_fn2: mock.sentinel.key2, mock.sentinel.key_fn3: mock.sentinel.key3} m_dispatcher = mock.Mock() m_consumer = mock.Mock() m_consumer.consumes = consumes m_wrap_consumer.return_value = mock.sentinel.handler m_init.return_value = None pipeline = _TestEventPipeline() pipeline._dispatcher = m_dispatcher pipeline.register(m_consumer) m_wrap_consumer.assert_called_once_with(m_consumer) m_dispatcher.register.assert_has_calls([ mock.call(key_fn, key, mock.sentinel.handler) for key_fn, key in consumes.items()], any_order=True)
