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)
python类Mock()的实例源码
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)
