本文整理汇总了Python中neutron_lbaas.tests.nested函数的典型用法代码示例。如果您正苦于以下问题：Python nested函数的具体用法？Python nested怎么用？Python nested使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了nested函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_create_member_on_different_subnets
def test_create_member_on_different_subnets(self):
with nested(
self.subnet(),
self.subnet(cidr='20.0.0.0/24'),
self.subnet(cidr='30.0.0.0/24')
) as (vip_sub, pool_sub, member_sub):
with self.pool(provider='radware',
subnet_id=pool_sub['subnet']['id']) as pool:
with nested(
self.port(subnet=vip_sub,
fixed_ips=[{'ip_address': '10.0.0.2'}]),
self.port(subnet=pool_sub,
fixed_ips=[{'ip_address': '20.0.0.2'}]),
self.port(subnet=member_sub,
fixed_ips=[{'ip_address': '30.0.0.2'}])
):
with nested(
self.member(pool_id=pool['pool']['id'],
address='10.0.0.2'),
self.member(pool_id=pool['pool']['id'],
address='20.0.0.2'),
self.member(pool_id=pool['pool']['id'],
address='30.0.0.2')
) as (member_vip, member_pool, member_out):
with self.vip(pool=pool, subnet=vip_sub):
calls = [
mock.call(
'POST', '/api/workflow/' +
pool['pool']['id'] +
'/action/BaseCreate',
mock.ANY, driver.TEMPLATE_HEADER
)
]
self.driver_rest_call_mock.assert_has_calls(
calls, any_order=True)
mock_calls = self.driver_rest_call_mock.mock_calls
params = mock_calls[-2][1][2]['parameters']
member_subnet_array = params['member_subnet_array']
member_mask_array = params['member_mask_array']
member_gw_array = params['member_gw_array']
self.assertEqual(['10.0.0.0',
'255.255.255.255',
'30.0.0.0'],
member_subnet_array)
self.assertEqual(['255.255.255.0',
'255.255.255.255',
'255.255.255.0'],
member_mask_array)
self.assertEqual(
[pool_sub['subnet']['gateway_ip'],
'255.255.255.255',
pool_sub['subnet']['gateway_ip']],
member_gw_array)
开发者ID:hthompson6，项目名称:neutron-lbaas，代码行数:54，代码来源:test_plugin_driver.py
示例2: test_build_objects_with_l7
def test_build_objects_with_l7(self):
with self.subnet(cidr='10.0.0.0/24') as vip_sub:
with self.loadbalancer(
subnet=vip_sub,
vip_address=WF_APPLY_PARAMS['parameters']['vip_address']
) as lb:
lb_id = lb['loadbalancer']['id']
with self.listener(
protocol=lb_con.PROTOCOL_HTTP,
loadbalancer_id=lb_id) as listener:
listener_id = listener['listener']['id']
with nested(
self.pool(
protocol=lb_con.PROTOCOL_HTTP,
listener_id=listener_id),
self.pool(
protocol=lb_con.PROTOCOL_HTTP,
loadbalancer_id=lb_id)) as (def_pool, pol_pool):
def_pool_id = def_pool['pool']['id']
pol_pool_id = pol_pool['pool']['id']
with self.l7policy(
listener_id,
action=lb_con.L7_POLICY_ACTION_REDIRECT_TO_POOL,
redirect_pool_id=pol_pool_id) as policy:
policy_id = policy['l7policy']['id']
self.driver_rest_call_mock.reset_mock()
with nested(
self.l7policy_rule(
l7policy_id=policy_id,
key=u'key1', value=u'val1'),
self.l7policy_rule(
l7policy_id=policy_id,
key=u'key2', value=u'val2'),
self.member(
pool_id=def_pool_id,
subnet=vip_sub,
address=u'10.0.1.10')) as (
rule1, rule2,
def_m):
self.driver_rest_call_mock.reset_mock()
rest_call_function_mock.__dict__.update(
{'WORKFLOW_MISSING': False})
with self.member(
pool_id=pol_pool_id,
subnet=vip_sub,
address=u'10.0.1.20'):
self.compare_apply_call()
开发者ID:mdurrant-b3，项目名称:neutron-lbaas，代码行数:51，代码来源:test_v2_plugin_driver.py
示例3: test_plug_no_gw
def test_plug_no_gw(self):
test_port = {'id': 'port_id',
'network_id': 'net_id',
'mac_address': 'mac_addr',
'fixed_ips': [{'ip_address': '10.0.0.2',
'subnet': {'cidr': '10.0.0.0/24'}}]}
test_address = '10.0.0.2'
with nested(
mock.patch('neutron.agent.linux.ip_lib.device_exists'),
mock.patch('netaddr.IPNetwork'),
mock.patch('neutron.agent.linux.ip_lib.IPWrapper'),
) as (dev_exists, ip_net, ip_wrap):
self.vif_driver.get_device_name.return_value = 'test_interface'
dev_exists.return_value = False
ip_net.return_value = ip_net
ip_net.prefixlen = 24
self.driver._plug('test_ns', test_port, test_address)
self.rpc_mock.plug_vip_port.assert_called_once_with(
test_port['id'])
self.assertTrue(dev_exists.called)
self.vif_driver.plug.assert_called_once_with('net_id', 'port_id',
'test_interface',
'mac_addr',
namespace='test_ns')
self.vif_driver.init_l3.assert_called_once_with(
'test_interface',
['10.0.0.2/24'],
namespace='test_ns'
)
self.assertFalse(ip_wrap.called)
dev_exists.return_value = True
self.assertRaises(exceptions.PreexistingDeviceFailure,
self.driver._plug, 'test_ns', test_port,
test_address, False)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:35，代码来源:test_namespace_driver.py
示例4: test_update_pool_health_monitor
def test_update_pool_health_monitor(self):
from_hmon = {
'admin_state_up': True, 'tenant_id': TENANT_ID, 'delay': 5,
'max_retries': 5, 'timeout': 5, 'pools': [
{'status': 'PENDING_UPDATE', 'status_description': None,
'pool_id': POOL_ID}],
'type': 'PING', 'id': HEALTHMON_ID}
to_hmon = {
'admin_state_up': True, 'tenant_id': TENANT_ID, 'delay': 5,
'max_retries': 10, 'timeout': 5, 'pools': [
{'status': 'ACTIVE', 'status_description': None,
'pool_id': POOL_ID}],
'type': 'PING', 'id': HEALTHMON_ID}
pool_mapping = {'edge_id': EDGE_ID, 'edge_pool_id': EDGE_POOL_ID}
mon_mapping = {'edge_id': EDGE_ID, 'edge_monitor_id': EDGE_MON_ID}
with nested(
mock.patch.object(db, 'get_nsxv_edge_pool_mapping'),
mock.patch.object(db, 'get_nsxv_edge_monitor_mapping'),
mock.patch.object(self.service_plugin._core_plugin.nsx_v,
'update_pool_health_monitor')
) as (mock_get_pool_mapping, mock_get_mon_mapping, mock_upd_pool_hm):
mock_get_pool_mapping.return_value = pool_mapping
mock_get_mon_mapping.return_value = mon_mapping
self.edge_driver.update_pool_health_monitor(
self.context, from_hmon, to_hmon, POOL_ID)
mock_upd_pool_hm.assert_called_with(
self.context, from_hmon, to_hmon, POOL_ID, mon_mapping)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:30，代码来源:test_edge_driver.py
示例5: test_build_objects_graph_lb_pool
def test_build_objects_graph_lb_pool(self):
with self.subnet(cidr='10.0.0.0/24') as vip_sub:
with self.loadbalancer(
subnet=vip_sub,
vip_address=WF_APPLY_PARAMS['parameters']['vip_address']
) as lb:
lb_id = lb['loadbalancer']['id']
with self.listener(loadbalancer_id=lb_id) as listener:
listener_id = listener['listener']['id']
with self.pool(
protocol=lb_con.PROTOCOL_HTTP,
listener_id=listener_id) as pool:
with nested(
self.member(pool_id=pool['pool']['id'],
subnet=vip_sub, address='10.0.1.10'),
self.member(pool_id=pool['pool']['id'],
subnet=vip_sub, address='10.0.1.20')):
self.driver_rest_call_mock.reset_mock()
rest_call_function_mock.__dict__.update(
{'WORKFLOW_MISSING': False})
with self.pool(
protocol=lb_con.PROTOCOL_HTTP,
loadbalancer_id=lb_id):
self.compare_apply_call()
开发者ID:mdurrant-b3，项目名称:neutron-lbaas，代码行数:26，代码来源:test_v2_plugin_driver.py
示例6: test_update_vip
def test_update_vip(self):
vip_from = {
'status': 'ACTIVE', 'protocol': 'HTTP', 'description': '',
'address': '10.0.0.8', 'protocol_port': 555,
'port_id': VIP_PORT_ID, 'id': VIP_ID, 'status_description': None,
'name': 'testvip1', 'admin_state_up': True,
'subnet_id': SUBNET_ID, 'tenant_id': TENANT_ID,
'connection_limit': -1, 'pool_id': POOL_ID,
'session_persistence': {'type': 'SOURCE_IP'}}
vip_to = {
'status': 'PENDING_UPDATE', 'protocol': 'HTTP',
'description': '', 'address': '10.0.0.8', 'protocol_port': 555,
'port_id': VIP_PORT_ID, 'id': VIP_ID, 'status_description': None,
'name': 'testvip1', 'admin_state_up': True,
'subnet_id': SUBNET_ID, 'tenant_id': TENANT_ID,
'connection_limit': -1, 'pool_id': POOL_ID,
'session_persistence': {'type': 'HTTP_COOKIE'}}
pool_mapping = {'edge_id': EDGE_ID, 'edge_pool_id': EDGE_POOL_ID}
vip_mapping = {'edge_id': EDGE_ID, 'edge_vse_id': EDGE_VSE_ID,
'edge_app_profile_id': APP_PROFILE_ID}
with nested(
mock.patch.object(db, 'get_nsxv_edge_pool_mapping'),
mock.patch.object(db, 'get_nsxv_edge_vip_mapping'),
mock.patch.object(self.service_plugin._core_plugin.nsx_v,
'update_vip')
) as (mock_get_pool_mapping, mock_get_vip_mapping, mock_upd_vip):
mock_get_pool_mapping.return_value = pool_mapping
mock_get_vip_mapping.return_value = vip_mapping
self.edge_driver.update_vip(self.context, vip_from, vip_to)
mock_upd_vip.assert_called_with(self.context, vip_from, vip_to,
pool_mapping, vip_mapping)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:33，代码来源:test_edge_driver.py
示例7: test_build_objects_with_tls
def test_build_objects_with_tls(self):
with self.subnet(cidr='10.0.0.0/24') as vip_sub:
with self.loadbalancer(subnet=vip_sub) as lb:
lb_id = lb['loadbalancer']['id']
with nested(
mock.patch('neutron_lbaas.services.loadbalancer.plugin.'
'cert_parser', autospec=True),
mock.patch('neutron_lbaas.services.loadbalancer.plugin.'
'CERT_MANAGER_PLUGIN.CertManager',
autospec=True)
) as (cert_parser_mock, cert_manager_mock):
cert_mock = mock.Mock(spec=cert_manager.Cert)
cert_mock.get_certificate.return_value = 'certificate'
cert_mock.get_intermediates.return_value = 'intermediates'
cert_mock.get_private_key.return_value = 'private_key'
cert_mock.get_private_key_passphrase.return_value = \
'private_key_passphrase'
cert_manager_mock().get_cert.return_value = cert_mock
cert_parser_mock.validate_cert.return_value = True
with self.listener(
protocol=lb_con.PROTOCOL_TERMINATED_HTTPS,
loadbalancer_id=lb_id,
default_tls_container_ref='def1',
sni_container_refs=['sni1', 'sni2']) as listener:
with self.pool(
protocol=lb_con.PROTOCOL_HTTP,
listener_id=listener['listener']['id']) as pool:
with self.member(pool_id=pool['pool']['id'],
subnet=vip_sub,
address='10.0.1.10'):
self.compare_apply_call()
开发者ID:mercadolibre，项目名称:neutron-lbaas，代码行数:32，代码来源:test_v2_plugin_driver.py
示例8: test_plug_not_send_gratuitous_arp
def test_plug_not_send_gratuitous_arp(self):
self.conf.haproxy.send_gratuitous_arp = 0
test_port = {'id': 'port_id',
'network_id': 'net_id',
'mac_address': 'mac_addr',
'fixed_ips': [{'ip_address': '10.0.0.2',
'subnet': {'cidr': '10.0.0.0/24',
'gateway_ip': '10.0.0.1'}}]}
test_address = '10.0.0.2'
with nested(
mock.patch('neutron.agent.linux.ip_lib.device_exists'),
mock.patch('netaddr.IPNetwork'),
mock.patch('neutron.agent.linux.ip_lib.IPWrapper'),
) as (dev_exists, ip_net, ip_wrap):
self.vif_driver.get_device_name.return_value = 'test_interface'
dev_exists.return_value = False
ip_net.return_value = ip_net
ip_net.prefixlen = 24
self.driver._plug('test_ns', test_port, test_address)
cmd = ['route', 'add', 'default', 'gw', '10.0.0.1']
expected = [
self._ip_mock_call('test_ns'),
mock.call().netns.execute(cmd, check_exit_code=False)]
self.assertEqual(expected, ip_wrap.mock_calls)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:25，代码来源:test_namespace_driver.py
示例9: test_update_pool
def test_update_pool(self):
from_pool = {
'status': 'ACTIVE', 'lb_method': 'ROUND_ROBIN',
'protocol': 'HTTP', 'description': '', 'health_monitors': [],
'members': [], 'status_description': None, 'id': POOL_ID,
'vip_id': None, 'name': 'testpool2', 'admin_state_up': True,
'subnet_id': SUBNET_ID, 'tenant_id': TENANT_ID,
'health_monitors_status': [], 'provider': 'vmwareedge'}
to_pool = {
'status': 'PENDING_UPDATE', 'lb_method': 'LEAST_CONNECTIONS',
'protocol': 'HTTP', 'description': '', 'health_monitors': [],
'members': [], 'status_description': None, 'id': POOL_ID,
'vip_id': None, 'name': 'testpool2', 'admin_state_up': True,
'subnet_id': SUBNET_ID, 'tenant_id': TENANT_ID,
'health_monitors_status': [], 'provider': 'vmwareedge'}
mapping = {'edge_id': EDGE_ID, 'edge_pool_id': EDGE_POOL_ID}
with nested(
mock.patch.object(db, 'get_nsxv_edge_pool_mapping'),
mock.patch.object(self.service_plugin._core_plugin.nsx_v,
'update_pool')
) as (mock_get_mapping, mock_update_pool):
mock_get_mapping.return_value = mapping
self.edge_driver.update_pool(self.context, from_pool, to_pool)
mock_update_pool.assert_called_with(self.context, from_pool,
to_pool, mapping)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:29，代码来源:test_edge_driver.py
示例10: test_delete_pool_existing
def test_delete_pool_existing(self):
with nested(
mock.patch.object(self.driver, 'undeploy_instance'),
mock.patch.object(self.driver, 'exists'),
) as (undeploy, exists):
exists.return_value = True
self.driver.delete_pool({'id': '1'})
undeploy.assert_called_once_with('1', delete_namespace=True)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:8，代码来源:test_namespace_driver.py
示例11: test_delete_pool_non_existing
def test_delete_pool_non_existing(self):
with nested(
mock.patch.object(self.driver, 'undeploy_instance'),
mock.patch.object(self.driver, 'exists'),
) as (undeploy, exists):
exists.return_value = False
self.driver.delete_pool({'id': '1'})
self.assertFalse(undeploy.called)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:8，代码来源:test_namespace_driver.py
示例12: test_delete_pool_successful
def test_delete_pool_successful(self):
pool = {'id': POOL_ID}
with nested(
mock.patch.object(self.service_plugin, '_delete_db_pool'),
mock.patch.object(db, 'delete_nsxv_edge_pool_mapping')
) as (mock_del_db_pool, mock_del_mapping):
self.edge_driver.delete_pool_successful(self.context, pool)
mock_del_db_pool.assert_called_with(self.context, POOL_ID)
mock_del_mapping.assert_called_with(self.context, POOL_ID)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:10，代码来源:test_edge_driver.py
示例13: test_delete_vip_successful
def test_delete_vip_successful(self):
vip = {'pool_id': POOL_ID, 'id': VIP_ID}
with nested(
mock.patch.object(db, 'delete_nsxv_edge_vip_mapping'),
mock.patch.object(self.service_plugin, '_delete_db_vip')
) as (mock_del_vip_mapping, mock_del_vip):
self.edge_driver.delete_vip_successful(self.context, vip)
mock_del_vip_mapping.assert_called_with(self.context, POOL_ID)
mock_del_vip.assert_called_with(self.context, VIP_ID)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:10，代码来源:test_edge_driver.py
示例14: test_create_member_on_different_subnet_multiple_ports
def test_create_member_on_different_subnet_multiple_ports(self):
cfg.CONF.set_override("allow_overlapping_ips", 'true')
with self.network() as other_net:
with nested(
self.subnet(),
self.subnet(cidr='20.0.0.0/24'),
self.subnet(cidr='30.0.0.0/24'),
self.subnet(network=other_net, cidr='30.0.0.0/24')
) as (vip_sub, pool_sub, member_sub1, member_sub2):
with self.pool(provider='radware',
subnet_id=pool_sub['subnet']['id']) as pool:
with nested(
self.port(subnet=member_sub1,
fixed_ips=[{'ip_address': '30.0.0.2'}]),
self.port(subnet=member_sub2,
fixed_ips=[{'ip_address': '30.0.0.2'}])):
with self.member(pool_id=pool['pool']['id'],
address='30.0.0.2'):
with self.vip(pool=pool, subnet=vip_sub):
calls = [
mock.call(
'POST', '/api/workflow/' +
pool['pool']['id'] +
'/action/BaseCreate',
mock.ANY, driver.TEMPLATE_HEADER
)
]
self.driver_rest_call_mock.assert_has_calls(
calls, any_order=True)
calls = self.driver_rest_call_mock.mock_calls
params = calls[-2][1][2]['parameters']
m_sub_array = params['member_subnet_array']
m_mask_array = params['member_mask_array']
m_gw_array = params['member_gw_array']
self.assertEqual(['30.0.0.2'],
m_sub_array)
self.assertEqual(['255.255.255.255'],
m_mask_array)
self.assertEqual(
[pool_sub['subnet']['gateway_ip']],
m_gw_array)
开发者ID:hthompson6，项目名称:neutron-lbaas，代码行数:42，代码来源:test_plugin_driver.py
示例15: test_refresh_device_non_existing
def test_refresh_device_non_existing(self):
with nested(
mock.patch.object(self.driver, 'deploy_instance'),
mock.patch.object(self.driver, 'exists'),
mock.patch.object(self.driver, 'undeploy_instance')
) as (deploy, exists, undeploy):
pool_id = 'pool_id1'
deploy.return_value = False
exists.return_value = False
self.driver._refresh_device(pool_id)
self.assertFalse(undeploy.called)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:11，代码来源:test_namespace_driver.py
示例16: test_refresh_device_not_deployed
def test_refresh_device_not_deployed(self):
with nested(
mock.patch.object(self.driver, 'deploy_instance'),
mock.patch.object(self.driver, 'exists'),
mock.patch.object(self.driver, 'undeploy_instance')
) as (deploy, exists, undeploy):
pool_id = 'pool_id1'
deploy.return_value = False
exists.return_value = True
self.driver._refresh_device(pool_id)
undeploy.assert_called_once_with(pool_id)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:11，代码来源:test_namespace_driver.py
示例17: test_refresh_device
def test_refresh_device(self):
with nested(
mock.patch.object(self.driver, 'deploy_instance'),
mock.patch.object(self.driver, 'undeploy_instance')
) as (deploy, undeploy):
pool_id = 'pool_id1'
self.driver._refresh_device(pool_id)
self.rpc_mock.get_logical_device.assert_called_once_with(pool_id)
deploy.assert_called_once_with(
self.rpc_mock.get_logical_device.return_value)
self.assertFalse(undeploy.called)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:11，代码来源:test_namespace_driver.py
示例18: test_update
def test_update(self):
with nested(
mock.patch.object(self.driver, '_get_state_file_path'),
mock.patch.object(self.driver, '_spawn'),
mock.patch.object(six.moves.builtins, 'open')
) as (gsp, spawn, mock_open):
mock_open.return_value = ['5']
self.driver.update(self.fake_config)
mock_open.assert_called_once_with(gsp.return_value, 'r')
spawn.assert_called_once_with(self.fake_config, ['-sf', '5'])
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:12，代码来源:test_namespace_driver.py
示例19: test_main
def test_main(self):
logging_str = 'neutron.agent.common.config.setup_logging'
with nested(
mock.patch(logging_str),
mock.patch.object(agent.service, 'launch'),
mock.patch('sys.argv'),
mock.patch.object(agent.manager, 'LbaasAgentManager'),
mock.patch.object(cfg.CONF, 'register_opts')
) as (mock_logging, mock_launch, sys_argv, mgr_cls, ro):
agent.main()
mock_launch.assert_called_once_with(mock.ANY, mock.ANY)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:12，代码来源:test_agent.py
示例20: test_create_pool_health_monitor_successful
def test_create_pool_health_monitor_successful(self):
hmon = {'id': HEALTHMON_ID}
with nested(
mock.patch.object(db, 'add_nsxv_edge_monitor_mapping'),
mock.patch.object(self.edge_driver,
'pool_health_monitor_successful')
) as (mock_add_pool_mon_mapping, mock_pool_hmon_successful):
self.edge_driver.create_pool_health_monitor_successful(
self.context, hmon, POOL_ID, EDGE_ID, EDGE_MON_ID)
mock_add_pool_mon_mapping.assert_called_with(
self.context, HEALTHMON_ID, EDGE_ID, EDGE_MON_ID)
mock_pool_hmon_successful.assert_called_with(self.context,
hmon, POOL_ID)
开发者ID:fnaval，项目名称:neutron-lbaas，代码行数:13，代码来源:test_edge_driver.py
注：本文中的neutron_lbaas.tests.nested函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论