openstack/neutron
Code Issues Proposed changes

Implements agent for Quantum Networking testing

Browse Source 
This agent client plugs itself into each network.
Then tries to ping each fixed_ips.
Implements blueprint test-agent

Change-Id: I3908e37401272e9e091ceae66e72cfcdd13b7898
This commit is contained in:
Nachi Ueno 2012-08-10 22:15:32 +00:00
parent fce952e8fa
commit 762cc5d830
9 changed files with 777 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
bin/quantum-debug Executable file
View File

@ -0,0 +1,20 @@
#!/usr/bin/env python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2012 Openstack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from quantum.debug.shell import main
main()

33
quantum/debug/README Normal file
View File

@ -0,0 +1,33 @@
Debug Helper Script for Quantum
- Configure
export TEST_CONFIG_FILE=/etc/quantum/dhcp_agent.ini
or
export TEST_CONFIG_FILE=/etc/quantum/l3_agent.ini
you can also specify config file by --config-file option
- Usage
quantum-debug commands
probe-create <net-id> Create probe port - create port and interface, then plug it in.
This commands returns a port id of a probe port. A probe port is a port which is used to test.
The port id is probe id.
We can have multiple probe probes in a network, in order to check connectivity between ports.
quantum-debug probe-exec probe_id_1 'nc -l 192.168.100.3 22'
quantum-debug probe-exec probe_id_2 'nc -vz 192.168.100.4 22'
probe-delete <port-id> Delete probe - delete port then uplug
probe-exec <port-id> 'command' Exec commands on the namespace of the probe
`probe-exec <port-id>` 'interactive command' Exec interactive command (eg, ssh)
probe-list List probes
probe-clear Clear All probes
ping-all --id <network_id> --timeout 1 (optional)
ping-all is all-in-one command to ping all fixed ip's in all network or a specified network.
In the command probe is automatically created if needed.
quantum-debug extends the shell of quantumclient, so you can use all the commands of quantum

17
quantum/debug/__init__.py Normal file
View File

@ -0,0 +1,17 @@
#!/bin/python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

156
quantum/debug/commands.py Normal file
View File

@ -0,0 +1,156 @@
#!/bin/python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import logging
from cliff import lister
from quantumclient.common import utils
from quantumclient.quantum.v2_0 import QuantumCommand
from quantumclient.quantum.v2_0.port import _format_fixed_ips
class ProbeCommand(QuantumCommand):
log = logging.getLogger(__name__ + '.ProbeCommand')
def get_debug_agent(self):
return self.app.debug_agent
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
self.app.stdout.write(_('Unimplemented commands') + '\n')
class CreateProbe(ProbeCommand):
"""Create probe port and interface, then plug it in."""
log = logging.getLogger(__name__ + '.CreateProbe')
def get_parser(self, prog_name):
parser = super(CreateProbe, self).get_parser(prog_name)
parser.add_argument(
'id', metavar='network_id',
help='ID of network to probe')
return parser
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
debug_agent = self.get_debug_agent()
port = debug_agent.create_probe(parsed_args.id)
self.app.stdout.write(_('Probe created : %s ') % port.id + '\n')
class DeleteProbe(ProbeCommand):
"""Delete probe - delete port then uplug """
log = logging.getLogger(__name__ + '.DeleteProbe')
def get_parser(self, prog_name):
parser = super(DeleteProbe, self).get_parser(prog_name)
parser.add_argument(
'id', metavar='port_id',
help='ID of probe port to delete')
return parser
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
debug_agent = self.get_debug_agent()
debug_agent.delete_probe(parsed_args.id)
self.app.stdout.write(_('Probe %s deleted') % parsed_args.id + '\n')
class ListProbe(QuantumCommand, lister.Lister):
""" List probes """
log = logging.getLogger(__name__ + '.ListProbe')
_formatters = {'fixed_ips': _format_fixed_ips, }
def get_debug_agent(self):
return self.app.debug_agent
def get_data(self, parsed_args):
debug_agent = self.get_debug_agent()
info = debug_agent.list_probes()
columns = len(info) > 0 and sorted(info[0].keys()) or []
return (columns, (utils.get_item_properties(
s, columns, formatters=self._formatters, )
for s in info), )
class ClearProbe(ProbeCommand):
"""Clear All probes """
log = logging.getLogger(__name__ + '.ClearProbe')
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
debug_agent = self.get_debug_agent()
debug_agent.clear_probe()
self.app.stdout.write(_('All Probes deleted ') + '\n')
class ExecProbe(ProbeCommand):
"""Exec commands on the namespace of the probe
"""
log = logging.getLogger(__name__ + '.ExecProbe')
def get_parser(self, prog_name):
parser = super(ExecProbe, self).get_parser(prog_name)
parser.add_argument(
'id', metavar='port_id',
help='ID of probe port to execute command')
parser.add_argument(
'command', metavar='command',
nargs='?',
default=None,
help='Command to execute')
return parser
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
debug_agent = self.get_debug_agent()
result = debug_agent.exec_command(parsed_args.id, parsed_args.command)
self.app.stdout.write(result + '\n')
class PingAll(ProbeCommand):
"""Ping all fixed_ip
"""
log = logging.getLogger(__name__ + '.ExecProbe')
def get_parser(self, prog_name):
parser = super(PingAll, self).get_parser(prog_name)
parser.add_argument(
'--timeout', metavar='<timeout>',
default=10,
help='Ping timeout')
parser.add_argument(
'--id', metavar='network_id',
default=None,
help='ID of network')
return parser
def run(self, parsed_args):
self.log.debug('run(%s)' % parsed_args)
debug_agent = self.get_debug_agent()
result = debug_agent.ping_all(parsed_args.id,
timeout=parsed_args.timeout)
self.app.stdout.write(result + '\n')

189
quantum/debug/debug_agent.py Normal file
View File

@ -0,0 +1,189 @@
#!/bin/python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import errno
import logging
import os
import shlex
import socket
import sys
import netaddr
from quantum.agent.common import config
from quantum.agent.dhcp_agent import DictModel
from quantum.agent.linux import interface
from quantum.agent.linux import ip_lib
from quantum.agent.linux import utils
from quantum.openstack.common import cfg
from quantum.openstack.common import importutils
from quantumclient.v2_0 import client
LOG = logging.getLogger('test-agent')
DEVICE_OWNER_PROBE = 'network:probe'
class QuantumDebugAgent():
OPTS = [
cfg.StrOpt('root_helper', default='sudo'),
# Needed for drivers
cfg.StrOpt('admin_user'),
cfg.StrOpt('admin_password'),
cfg.StrOpt('admin_tenant_name'),
cfg.StrOpt('auth_url'),
cfg.StrOpt('auth_strategy', default='keystone'),
cfg.StrOpt('auth_region'),
cfg.BoolOpt('use_namespaces', default=True),
cfg.StrOpt('interface_driver',
help="The driver used to manage the virtual interface.")
]
def __init__(self, conf, client, driver):
self.conf = conf
self.client = client
self.driver = driver
def _get_namespace(self, port):
return "qprobe-%s" % port.id
def create_probe(self, network_id):
network = self._get_network(network_id)
port = self._create_port(network)
port.network = network
interface_name = self.driver.get_device_name(port)
namespace = None
if self.conf.use_namespaces:
namespace = self._get_namespace(port)
if ip_lib.device_exists(interface_name,
self.conf.root_helper, namespace):
LOG.debug(_('Reusing existing device: %s.') % interface_name)
else:
self.driver.plug(network.id,
port.id,
interface_name,
port.mac_address,
namespace=namespace)
ip_cidrs = []
for fixed_ip in port.fixed_ips:
subnet = fixed_ip.subnet
net = netaddr.IPNetwork(subnet.cidr)
ip_cidr = '%s/%s' % (fixed_ip.ip_address, net.prefixlen)
ip_cidrs.append(ip_cidr)
self.driver.init_l3(interface_name, ip_cidrs, namespace=namespace)
return port
def _get_subnet(self, subnet_id):
subnet_dict = self.client.show_subnet(subnet_id)['subnet']
return DictModel(subnet_dict)
def _get_network(self, network_id):
network_dict = self.client.show_network(network_id)['network']
network = DictModel(network_dict)
obj_subnet = [self._get_subnet(s_id) for s_id in network.subnets]
network.subnets = obj_subnet
return network
def clear_probe(self):
ports = self.client.list_ports(device_id=socket.gethostname(),
device_owner=DEVICE_OWNER_PROBE)
info = ports['ports']
for port in info:
self.delete_probe(port['id'])
def delete_probe(self, port_id):
port = DictModel(self.client.show_port(port_id)['port'])
ip = ip_lib.IPWrapper(self.conf.root_helper)
namespace = self._get_namespace(port)
if self.conf.use_namespaces and ip.netns.exists(namespace):
self.driver.unplug(self.driver.get_device_name(port),
namespace=namespace)
ip.netns.delete(namespace)
else:
self.driver.unplug(self.driver.get_device_name(port))
self.client.delete_port(port.id)
def list_probes(self):
ports = self.client.list_ports(device_owner=DEVICE_OWNER_PROBE)
info = ports['ports']
for port in info:
port['device_name'] = self.driver.get_device_name(DictModel(port))
return info
def exec_command(self, port_id, command=None):
port = DictModel(self.client.show_port(port_id)['port'])
ip = ip_lib.IPWrapper(self.conf.root_helper)
namespace = self._get_namespace(port)
if self.conf.use_namespaces:
if not command:
return "sudo ip netns exec %s" % self._get_namespace(port)
namespace = ip.ensure_namespace(namespace)
return namespace.netns.execute(shlex.split(command))
else:
return utils.execute(shlex.split(command))
def ensure_probe(self, network_id):
ports = self.client.list_ports(network_id=network_id,
device_id=socket.gethostname(),
device_owner=DEVICE_OWNER_PROBE)
info = ports.get('ports', [])
if info:
return DictModel(info[0])
else:
return self.create_probe(network_id)
def ping_all(self, network_id=None, timeout=1):
if network_id:
ports = self.client.list_ports(network_id=network_id)['ports']
else:
ports = self.client.list_ports()['ports']
result = ""
for port in ports:
probe = self.ensure_probe(port['network_id'])
if port['device_owner'] == DEVICE_OWNER_PROBE:
continue
for fixed_ip in port['fixed_ips']:
address = fixed_ip['ip_address']
subnet = self._get_subnet(fixed_ip['subnet_id'])
if subnet.ip_version == 4:
ping_command = 'ping'
else:
ping_command = 'ping6'
result += self.exec_command(probe.id,
'%s -c 1 -w %s %s' % (ping_command,
timeout,
address))
return result
def _create_port(self, network):
body = dict(port=dict(
admin_state_up=True,
network_id=network.id,
device_id='%s' % socket.gethostname(),
device_owner=DEVICE_OWNER_PROBE,
tenant_id=network.tenant_id,
fixed_ips=[dict(subnet_id=s.id) for s in network.subnets]))
port_dict = self.client.create_port(body)['port']
port = DictModel(port_dict)
port.network = network
for fixed_ip in port.fixed_ips:
fixed_ip.subnet = self._get_subnet(fixed_ip.subnet_id)
return port

86
quantum/debug/shell.py Normal file
View File

@ -0,0 +1,86 @@
#!/bin/python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import itertools
import sys
from quantum.agent.common import config
from quantum.agent.linux import interface
import quantum.debug.commands
from quantum.debug.debug_agent import QuantumDebugAgent
from quantum.openstack.common import cfg
from quantum.openstack.common import importutils
from quantumclient.common import exceptions as exc
from quantumclient.common import utils
from quantumclient.shell import env, QuantumShell, QUANTUM_API_VERSION
COMMAND_V2 = {
'probe-create': utils.import_class(
'quantum.debug.commands.CreateProbe'),
'probe-delete': utils.import_class(
'quantum.debug.commands.DeleteProbe'),
'probe-list': utils.import_class(
'quantum.debug.commands.ListProbe'),
'probe-clear': utils.import_class(
'quantum.debug.commands.ClearProbe'),
'probe-exec': utils.import_class(
'quantum.debug.commands.ExecProbe'),
'ping-all': utils.import_class(
'quantum.debug.commands.PingAll'),
#TODO(nati) ping, netcat , nmap, bench
}
COMMANDS = {'2.0': COMMAND_V2}
class QuantumDebugShell(QuantumShell):
def __init__(self, api_version):
super(QuantumDebugShell, self).__init__(api_version)
for k, v in COMMANDS[api_version].items():
self.command_manager.add_command(k, v)
def build_option_parser(self, description, version):
parser = super(QuantumDebugShell, self).build_option_parser(
description, version)
parser.add_argument(
'--config-file',
default=env('TEST_CONFIG_FILE'),
help='Config file for interface driver '
'(You may also use either the '
'l3_agent.ini or the dhcp_agent.ini)')
return parser
def initialize_app(self, argv):
super(QuantumDebugShell, self).initialize_app(argv)
if not self.options.config_file:
raise exc.CommandError(
"You must provide a config file for bridge -"
" either --config-file or env[TEST_CONFIG_FILE]")
client = self.client_manager.quantum
cfg.CONF.register_opts(interface.OPTS)
cfg.CONF.register_opts(QuantumDebugAgent.OPTS)
cfg.CONF(['--config-file', self.options.config_file])
config.setup_logging(cfg.CONF)
driver = importutils.import_object(cfg.CONF.interface_driver, cfg.CONF)
self.debug_agent = QuantumDebugAgent(cfg.CONF, client, driver)
def main(argv=None):
return QuantumDebugShell(QUANTUM_API_VERSION).run(argv or sys.argv[1:])
if __name__ == "__main__":
sys.exit(main())

271
quantum/tests/unit/test_debug_commands.py Normal file
View File

@ -0,0 +1,271 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import socket
import sys
import uuid
import mock
import unittest2 as unittest
from quantum.agent.common import config
from quantum.agent.linux import interface
from quantum.agent.linux import utils
from quantum.common import exceptions
from quantum.debug import commands
from quantum.debug.debug_agent import DEVICE_OWNER_PROBE, QuantumDebugAgent
from quantum.openstack.common import cfg
class MyApp(object):
def __init__(self, _stdout):
self.stdout = _stdout
class TestDebugCommands(unittest.TestCase):
def setUp(self):
cfg.CONF.register_opts(interface.OPTS)
cfg.CONF.register_opts(QuantumDebugAgent.OPTS)
cfg.CONF(args=sys.argv, project='quantum')
cfg.CONF.set_override('use_namespaces', True)
cfg.CONF.root_helper = 'sudo'
self.addCleanup(mock.patch.stopall)
device_exists_p = mock.patch(
'quantum.agent.linux.ip_lib.device_exists')
device_exists_p.start()
namespace_p = mock.patch(
'quantum.agent.linux.ip_lib.IpNetnsCommand')
namespace_p.start()
ensure_namespace_p = mock.patch(
'quantum.agent.linux.ip_lib.IPWrapper.ensure_namespace')
ensure_namespace_p.start()
dvr_cls_p = mock.patch('quantum.agent.linux.interface.NullDriver')
driver_cls = dvr_cls_p.start()
mock_driver = mock.MagicMock()
mock_driver.DEV_NAME_LEN = (
interface.LinuxInterfaceDriver.DEV_NAME_LEN)
mock_driver.get_device_name.return_value = 'tap12345678-12'
driver_cls.return_value = mock_driver
self.driver = mock_driver
client_cls_p = mock.patch('quantumclient.v2_0.client.Client')
client_cls = client_cls_p.start()
client_inst = mock.Mock()
client_cls.return_value = client_inst
fake_network = {'network': {'id': 'fake_net',
'tenant_id': 'fake_tenant',
'subnets': ['fake_subnet']}}
fake_port = {'port':
{'id': 'fake_port',
'device_owner': 'fake_device',
'mac_address': 'aa:bb:cc:dd:ee:ffa',
'network_id': 'fake_net',
'fixed_ips':
[{'subnet_id': 'fake_subnet', 'ip_address':'10.0.0.3'}]
}}
fake_ports = {'ports': [fake_port['port']]}
self.fake_ports = fake_ports
allocation_pools = [{'start': '10.0.0.2',
'end': '10.0.0.254'}]
fake_subnet_v4 = {'subnet': {'name': 'fake_subnet_v4',
'id': 'fake_subnet',
'network_id': 'fake_net',
'gateway_ip': '10.0.0.1',
'dns_nameservers': ['10.0.0.2'],
'host_routes': [],
'cidr': '10.0.0.0/24',
'allocation_pools': allocation_pools,
'enable_dhcp': True,
'ip_version': 4}}
client_inst.list_ports.return_value = fake_ports
client_inst.create_port.return_value = fake_port
client_inst.show_port.return_value = fake_port
client_inst.show_network.return_value = fake_network
client_inst.show_subnet.return_value = fake_subnet_v4
self.client = client_inst
mock_std = mock.Mock()
self.app = MyApp(mock_std)
self.app.debug_agent = QuantumDebugAgent(cfg.CONF,
client_inst,
mock_driver)
def test_create_probe(self):
cmd = commands.CreateProbe(self.app, None)
cmd_parser = cmd.get_parser('create_probe')
args = ['fake_net']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
fake_port = {'port':
{'device_owner': DEVICE_OWNER_PROBE,
'admin_state_up': True,
'network_id': 'fake_net',
'tenant_id': 'fake_tenant',
'fixed_ips': [{'subnet_id': 'fake_subnet'}],
'device_id': socket.gethostname()}}
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.show_network('fake_net'),
mock.call.show_subnet('fake_subnet'),
mock.call.create_port(fake_port),
mock.call.show_subnet('fake_subnet')])
self.driver.assert_has_calls([mock.call.init_l3('tap12345678-12',
['10.0.0.3/24'],
namespace=namespace
)])
def test_delete_probe(self):
cmd = commands.DeleteProbe(self.app, None)
cmd_parser = cmd.get_parser('delete_probe')
args = ['fake_port']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.show_port('fake_port'),
mock.call.delete_port('fake_port')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.unplug('tap12345678-12',
namespace=namespace)])
def test_delete_probe_without_namespace(self):
cfg.CONF.set_override('use_namespaces', False)
cmd = commands.DeleteProbe(self.app, None)
cmd_parser = cmd.get_parser('delete_probe')
args = ['fake_port']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
self.client.assert_has_calls([mock.call.show_port('fake_port'),
mock.call.delete_port('fake_port')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.unplug('tap12345678-12')])
def test_list_probe(self):
cmd = commands.ListProbe(self.app, None)
cmd_parser = cmd.get_parser('list_probe')
args = []
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
self.client.assert_has_calls(
[mock.call.list_ports(device_owner=DEVICE_OWNER_PROBE)])
def test_exec_command(self):
cmd = commands.ExecProbe(self.app, None)
cmd_parser = cmd.get_parser('exec_command')
args = ['fake_port', 'fake_command']
parsed_args = cmd_parser.parse_args(args)
with mock.patch('quantum.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
self.client.assert_has_calls([mock.call.show_port('fake_port')])
def test_exec_command_without_namespace(self):
cfg.CONF.set_override('use_namespaces', False)
cmd = commands.ExecProbe(self.app, None)
cmd_parser = cmd.get_parser('exec_command')
args = ['fake_port', 'fake_command']
parsed_args = cmd_parser.parse_args(args)
with mock.patch('quantum.agent.linux.utils.execute') as exe:
cmd.run(parsed_args)
exe.assert_has_calls([mock.call.execute(mock.ANY)])
self.client.assert_has_calls([mock.call.show_port('fake_port')])
def test_clear_probe(self):
cmd = commands.ClearProbe(self.app, None)
cmd_parser = cmd.get_parser('clear_probe')
args = []
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.list_ports(
device_id=socket.gethostname(),
device_owner=DEVICE_OWNER_PROBE),
mock.call.show_port('fake_port'),
mock.call.delete_port('fake_port')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.unplug('tap12345678-12',
namespace=namespace)])
def test_ping_all_with_ensure_port(self):
fake_ports = self.fake_ports
def fake_port_list(network_id=None, device_owner=None, device_id=None):
if network_id:
# In order to test ensure_port, return []
return {'ports': []}
return fake_ports
self.client.list_ports.side_effect = fake_port_list
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')
args = []
parsed_args = cmd_parser.parse_args(args)
namespace = 'qprobe-fake_port'
with mock.patch('quantum.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
fake_port = {'port':
{'device_owner': DEVICE_OWNER_PROBE,
'admin_state_up': True,
'network_id': 'fake_net',
'tenant_id': 'fake_tenant',
'fixed_ips': [{'subnet_id': 'fake_subnet'}],
'device_id': socket.gethostname()}}
expected = [mock.call.show_network('fake_net'),
mock.call.show_subnet('fake_subnet'),
mock.call.create_port(fake_port),
mock.call.show_subnet('fake_subnet')]
self.client.assert_has_calls(expected)
self.driver.assert_has_calls([mock.call.init_l3('tap12345678-12',
['10.0.0.3/24'],
namespace=namespace
)])
def test_ping_all(self):
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')
args = []
parsed_args = cmd_parser.parse_args(args)
with mock.patch('quantum.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
fake_port = {'port':
{'device_owner': DEVICE_OWNER_PROBE,
'admin_state_up': True,
'network_id': 'fake_net',
'tenant_id': 'fake_tenant',
'fixed_ips': [{'subnet_id': 'fake_subnet'}],
'device_id': socket.gethostname()}}
expected = [mock.call.list_ports(),
mock.call.list_ports(network_id='fake_net',
device_owner=DEVICE_OWNER_PROBE,
device_id=socket.gethostname()),
mock.call.show_subnet('fake_subnet'),
mock.call.show_port('fake_port')]
self.client.assert_has_calls(expected)
def test_ping_all_v6(self):
fake_subnet_v6 = {'subnet': {'name': 'fake_v6',
'ip_version': 6}}
self.client.show_subnet.return_value = fake_subnet_v6
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')
args = []
parsed_args = cmd_parser.parse_args(args)
with mock.patch('quantum.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
self.client.assert_has_calls([mock.call.list_ports()])

1
setup.py
View File

@ -112,6 +112,7 @@ setuptools.setup(
'quantum-nec-agent = '
'quantum.plugins.nec.agent.nec_quantum_agent:main',
'quantum-server = quantum.server:main',
'quantum-debug = quantum.debug.shell:main',
]
},
)

7
tools/test-requires
View File

@ -1,11 +1,12 @@
distribute>=0.6.24
cliff
coverage
mock>=0.8
distribute>=0.6.24
mock>=1.0b1
mox==0.5.3
nose
nosehtmloutput
nosexcover
openstack.nose_plugin
nosehtmloutput
pep8
sphinx>=1.1.2
unittest2