本文整理汇总了Python中neutron.tests.common.helpers.register_dhcp_agent函数的典型用法代码示例。如果您正苦于以下问题：Python register_dhcp_agent函数的具体用法？Python register_dhcp_agent怎么用？Python register_dhcp_agent使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了register_dhcp_agent函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_auto_schedule_one_network_two_segments_with_one_subnet_each
def test_auto_schedule_one_network_two_segments_with_one_subnet_each(self):
net_id = self._create_network()
seg1_id = self._create_segment(net_id)
self._create_subnet(seg1_id, net_id)
helpers.register_dhcp_agent(HOST_D)
segments_service_db.update_segment_host_mapping(
self.ctx, HOST_D, {seg1_id})
scheduler = dhcp_agent_scheduler.ChanceScheduler()
observed_val_first_segment = scheduler.auto_schedule_networks(
self.plugin, self.ctx, HOST_D)
self.assertTrue(observed_val_first_segment)
agents = self.plugin.get_dhcp_agents_hosting_networks(
self.ctx, [net_id])
self.assertEqual(1, len(agents))
seg2_id = self._create_segment(net_id)
self._create_subnet(seg2_id, net_id, '192.168.11.0/24')
helpers.register_dhcp_agent(HOST_C)
segments_service_db.update_segment_host_mapping(
self.ctx, HOST_C, {seg2_id})
observed_val_second_segment = scheduler.auto_schedule_networks(
self.plugin, self.ctx, HOST_C)
self.assertTrue(observed_val_second_segment)
agents = self.plugin.get_dhcp_agents_hosting_networks(
self.ctx, [net_id])
self.assertEqual(2, len(agents))
开发者ID:AradhanaSingh，项目名称:neutron，代码行数:26，代码来源:test_dhcp_agent_scheduler.py
示例2: test_get_agent_by_host_no_candidate
def test_get_agent_by_host_no_candidate(self):
# Register a bunch of non-L2 agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
agent = self.db_mixin.get_agent_by_host(
self.ctx.session, helpers.HOST)
self.assertIsNone(agent)
开发者ID:abhilabh，项目名称:neutron，代码行数:7，代码来源:test_db.py
示例3: test_get_agent_by_host
def test_get_agent_by_host(self):
helpers.register_l3_agent()
helpers.register_dhcp_agent()
helpers.register_ovs_agent()
agent = l2pop_db.get_agent_by_host(
self.ctx.session, helpers.HOST)
self.assertEqual(constants.AGENT_TYPE_OVS, agent.agent_type)
开发者ID:sebrandon1，项目名称:neutron，代码行数:7，代码来源:test_db.py
示例4: _register_agent_states
def _register_agent_states(self, lbaas_agents=False):
"""Register two L3 agents and two DHCP agents."""
l3_hosta = helpers._get_l3_agent_dict(
L3_HOSTA, constants.L3_AGENT_MODE_LEGACY)
l3_hostb = helpers._get_l3_agent_dict(
L3_HOSTB, constants.L3_AGENT_MODE_LEGACY)
dhcp_hosta = helpers._get_dhcp_agent_dict(DHCP_HOSTA)
dhcp_hostc = helpers._get_dhcp_agent_dict(DHCP_HOSTC)
helpers.register_l3_agent(host=L3_HOSTA)
helpers.register_l3_agent(host=L3_HOSTB)
helpers.register_dhcp_agent(host=DHCP_HOSTA)
helpers.register_dhcp_agent(host=DHCP_HOSTC)
res = [l3_hosta, l3_hostb, dhcp_hosta, dhcp_hostc]
if lbaas_agents:
lbaas_hosta = {
'binary': 'neutron-loadbalancer-agent',
'host': LBAAS_HOSTA,
'topic': 'LOADBALANCER_AGENT',
'configurations': {'device_drivers': ['haproxy_ns']},
'agent_type': constants.AGENT_TYPE_LOADBALANCER}
lbaas_hostb = copy.deepcopy(lbaas_hosta)
lbaas_hostb['host'] = LBAAS_HOSTB
callback = agents_db.AgentExtRpcCallback()
callback.report_state(
self.adminContext,
agent_state={'agent_state': lbaas_hosta},
time=datetime.utcnow().strftime(constants.ISO8601_TIME_FORMAT))
callback.report_state(
self.adminContext,
agent_state={'agent_state': lbaas_hostb},
time=datetime.utcnow().strftime(constants.ISO8601_TIME_FORMAT))
res += [lbaas_hosta, lbaas_hostb]
return res
开发者ID:openstack，项目名称:neutron，代码行数:35，代码来源:test_agent.py
示例5: _test_host_changed
def _test_host_changed(self, twice):
self._register_ml2_agents()
helpers.register_dhcp_agent()
with self.subnet(network=self._network) as subnet:
host_arg = {portbindings.HOST_ID: HOST}
with self.port(subnet=subnet,
device_owner=DEVICE_OWNER_COMPUTE,
arg_list=(portbindings.HOST_ID,),
**host_arg) as port1:
p1 = port1['port']
s1 = subnet['subnet']
device1 = 'tap' + p1['id']
p1_ips = [p['ip_address'] for p in p1['fixed_ips']]
s1_cidr = s1['cidr'][s1['cidr'].index('/'):]
expected1 = {'id': p1['id'],
'network_id': p1['network_id'],
'mac_address': p1['mac_address'],
'ip_address': p1_ips[0] + s1_cidr,
'gateway_ip': s1['gateway_ip'],
'evpn': {'import_rt': '64512:101',
'export_rt': '64512:101'
}
}
self.callbacks.update_device_up(self.adminContext,
agent_id=HOST,
device=device1)
self.mock_cast.assert_called_once_with(
mock.ANY, 'attach_port_on_bagpipe_network',
expected1, HOST)
开发者ID:ythomas1，项目名称:networking-bagpipe，代码行数:33，代码来源:test_mech_bagpipe.py
示例6: test_get_agent_by_host
def test_get_agent_by_host(self):
# Register a L2 agent + A bunch of other agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
helpers.register_ovs_agent()
agent = self.db_mixin.get_agent_by_host(
self.ctx.session, helpers.HOST)
self.assertEqual(constants.AGENT_TYPE_OVS, agent.agent_type)
开发者ID:abhilabh，项目名称:neutron，代码行数:8，代码来源:test_db.py
示例7: test_get_dvr_active_network_ports_no_candidate
def test_get_dvr_active_network_ports_no_candidate(self):
self._setup_port_binding()
# Register a bunch of non-L2 agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
tunnel_network_ports = self.db_mixin.get_dvr_active_network_ports(
self.ctx.session, 'network_id')
self.assertEqual(0, len(tunnel_network_ports))
开发者ID:abhilabh，项目名称:neutron，代码行数:8，代码来源:test_db.py
示例8: test_scheduler_one_agents_per_network
def test_scheduler_one_agents_per_network(self):
self._save_networks(['1111'])
helpers.register_dhcp_agent(HOST_C)
self.plugin.network_scheduler.schedule(self.plugin, self.ctx,
{'id': '1111'})
agents = self.plugin.get_dhcp_agents_hosting_networks(self.ctx,
['1111'])
self.assertEqual(1, len(agents))
开发者ID:cisco-openstack，项目名称:neutron，代码行数:8，代码来源:test_dhcp_agent_scheduler.py
示例9: test_get_nondistributed_active_network_ports_no_candidate
def test_get_nondistributed_active_network_ports_no_candidate(self):
self._setup_port_binding(dvr=False)
# Register a bunch of non-L2 agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
fdb_network_ports = l2pop_db.get_nondistributed_active_network_ports(
self.ctx.session, TEST_NETWORK_ID)
self.assertEqual(0, len(fdb_network_ports))
开发者ID:sebrandon1，项目名称:neutron，代码行数:8，代码来源:test_db.py
示例10: test_scheduler_one_agents_per_network
def test_scheduler_one_agents_per_network(self):
net_id = self._create_network()
helpers.register_dhcp_agent(HOST_C)
self.plugin.network_scheduler.schedule(self.plugin, self.ctx,
{'id': net_id})
agents = self.plugin.get_dhcp_agents_hosting_networks(self.ctx,
[net_id])
self.assertEqual(1, len(agents))
开发者ID:AradhanaSingh，项目名称:neutron，代码行数:8，代码来源:test_dhcp_agent_scheduler.py
示例11: test_get_distributed_active_network_ports_no_candidate
def test_get_distributed_active_network_ports_no_candidate(self):
self._setup_port_binding(
device_owner=constants.DEVICE_OWNER_DVR_INTERFACE)
# Register a bunch of non-L2 agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
tunnel_network_ports = l2pop_db.get_distributed_active_network_ports(
self.ctx.session, TEST_NETWORK_ID)
self.assertEqual(0, len(tunnel_network_ports))
开发者ID:sebrandon1，项目名称:neutron，代码行数:9，代码来源:test_db.py
示例12: test_scheduler_two_agents_per_network
def test_scheduler_two_agents_per_network(self):
cfg.CONF.set_override('dhcp_agents_per_network', 2)
self._save_networks(['1111'])
helpers.register_dhcp_agent(HOST_C)
helpers.register_dhcp_agent(HOST_D)
self.plugin.network_scheduler.schedule(self.plugin, self.ctx,
{'id': '1111'})
agents = self.plugin.get_dhcp_agents_hosting_networks(self.ctx,
['1111'])
self.assertEqual(2, len(agents))
开发者ID:cisco-openstack，项目名称:neutron，代码行数:10，代码来源:test_dhcp_agent_scheduler.py
示例13: test_schedule_segment_no_host_mapping
def test_schedule_segment_no_host_mapping(self):
net_id = self._create_network()
seg_id = self._create_segment(net_id)
helpers.register_dhcp_agent(HOST_C)
helpers.register_dhcp_agent(HOST_D)
net = self.plugin.get_network(self.ctx, net_id)
seg = self.segments_plugin.get_segment(self.ctx, seg_id)
net['candidate_hosts'] = seg['hosts']
agents = self.plugin.network_scheduler.schedule(
self.plugin, self.ctx, net)
self.assertEqual(0, len(agents))
开发者ID:AradhanaSingh，项目名称:neutron，代码行数:11，代码来源:test_dhcp_agent_scheduler.py
示例14: test_get_dvr_active_network_ports
def test_get_dvr_active_network_ports(self):
self._setup_port_binding()
# Register a L2 agent + A bunch of other agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
helpers.register_ovs_agent()
tunnel_network_ports = l2pop_db.get_dvr_active_network_ports(
self.ctx.session, 'network_id')
self.assertEqual(1, len(tunnel_network_ports))
_, agent = tunnel_network_ports[0]
self.assertEqual(constants.AGENT_TYPE_OVS, agent.agent_type)
开发者ID:21atlas，项目名称:neutron，代码行数:11，代码来源:test_db.py
示例15: test__get_ha_router_interface_ids_with_ha_replicated_port
def test__get_ha_router_interface_ids_with_ha_replicated_port(self):
helpers.register_dhcp_agent()
helpers.register_l3_agent()
helpers.register_ovs_agent()
self._create_ha_router()
self._setup_port_binding(
device_owner=constants.DEVICE_OWNER_HA_REPLICATED_INT,
device_id=TEST_ROUTER_ID)
ha_iface_ids = l2pop_db._get_ha_router_interface_ids(
self.ctx.session, TEST_NETWORK_ID)
self.assertEqual(1, len(list(ha_iface_ids)))
开发者ID:sebrandon1，项目名称:neutron，代码行数:11，代码来源:test_db.py
示例16: test_get_nondistributed_active_network_ports
def test_get_nondistributed_active_network_ports(self):
self._setup_port_binding(dvr=False)
# Register a L2 agent + A bunch of other agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
helpers.register_ovs_agent()
fdb_network_ports = l2pop_db.get_nondistributed_active_network_ports(
self.ctx.session, TEST_NETWORK_ID)
self.assertEqual(1, len(fdb_network_ports))
_, agent = fdb_network_ports[0]
self.assertEqual(constants.AGENT_TYPE_OVS, agent.agent_type)
开发者ID:sebrandon1，项目名称:neutron，代码行数:11，代码来源:test_db.py
示例17: test_get_nondvr_active_network_ports
def test_get_nondvr_active_network_ports(self):
self._setup_port_binding(dvr=False)
# Register a L2 agent + A bunch of other agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
helpers.register_ovs_agent()
fdb_network_ports = self.db_mixin.get_nondvr_active_network_ports(
self.ctx.session, 'network_id')
self.assertEqual(1, len(fdb_network_ports))
_, agent = fdb_network_ports[0]
self.assertEqual(constants.AGENT_TYPE_OVS, agent.agent_type)
开发者ID:abhilabh，项目名称:neutron，代码行数:11，代码来源:test_db.py
示例18: test_schedule_segment_one_hostable_agent
def test_schedule_segment_one_hostable_agent(self):
net_id = self._create_network()
seg_id = self._create_segment(net_id)
helpers.register_dhcp_agent(HOST_C)
helpers.register_dhcp_agent(HOST_D)
segments_service_db.update_segment_host_mapping(
self.ctx, HOST_C, {seg_id})
net = self._get_network_with_candidate_hosts(net_id, seg_id)
agents = self.plugin.network_scheduler.schedule(
self.plugin, self.ctx, net)
self.assertEqual(1, len(agents))
self.assertEqual(HOST_C, agents[0].host)
开发者ID:AradhanaSingh，项目名称:neutron，代码行数:12，代码来源:test_dhcp_agent_scheduler.py
示例19: test_get_ha_agents_by_router_id
def test_get_ha_agents_by_router_id(self):
helpers.register_dhcp_agent()
helpers.register_l3_agent()
helpers.register_ovs_agent()
self._create_ha_router()
self._setup_port_binding(
device_owner=constants.DEVICE_OWNER_ROUTER_SNAT,
device_id=TEST_ROUTER_ID)
agents = l2pop_db.get_ha_agents_by_router_id(
self.ctx.session, TEST_ROUTER_ID)
ha_agents = [agent.host for agent in agents]
self.assertEqual(tools.UnorderedList([HOST, HOST_2]), ha_agents)
开发者ID:sebrandon1，项目名称:neutron，代码行数:12，代码来源:test_db.py
示例20: test_get_distributed_active_network_ports
def test_get_distributed_active_network_ports(self):
self._setup_port_binding(
device_owner=constants.DEVICE_OWNER_DVR_INTERFACE)
# Register a L2 agent + A bunch of other agents on the same host
helpers.register_l3_agent()
helpers.register_dhcp_agent()
helpers.register_ovs_agent()
tunnel_network_ports = l2pop_db.get_distributed_active_network_ports(
self.ctx.session, TEST_NETWORK_ID)
self.assertEqual(1, len(tunnel_network_ports))
_, agent = tunnel_network_ports[0]
self.assertEqual(constants.AGENT_TYPE_OVS, agent.agent_type)
开发者ID:sebrandon1，项目名称:neutron，代码行数:12，代码来源:test_db.py
注：本文中的neutron.tests.common.helpers.register_dhcp_agent函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论