x/vmware-nsx
Code Issues Proposed changes

Merge "Add unit test for ryu-agent"

Browse Source
This commit is contained in:
Jenkins 2013-02-19 13:15:23 +00:00 committed by Gerrit Code Review
commit f52faafdc5
5 changed files with 549 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
quantum/tests/unit/ryu/fake_ryu.py Normal file
View File

@ -0,0 +1,42 @@
# Copyright (c) 2013 OpenStack, LLC.
#
# 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 mock
def patch_fake_ryu_client():
ryu_mod = mock.Mock()
ryu_app_mod = ryu_mod.app
ryu_app_client = ryu_app_mod.client
conf_switch_key = ryu_app_mod.conf_switch_key
conf_switch_key.OVSDB_ADDR = 'ovsdb_addr'
conf_switch_key.OVS_TUNNEL_ADDR = 'ovs_tunnel_addr'
rest_nw_id = ryu_app_mod.rest_nw_id
rest_nw_id.NW_ID_EXTERNAL = '__NW_ID_EXTERNAL__'
rest_nw_id.NW_ID_RESERVED = '__NW_ID_RESERVED__'
rest_nw_id.NW_ID_VPORT_GRE = '__NW_ID_VPORT_GRE__'
rest_nw_id.NW_ID_UNKNOWN = '__NW_ID_UNKNOWN__'
rest_nw_id.RESERVED_NETWORK_IDS = [
rest_nw_id.NW_ID_EXTERNAL,
rest_nw_id.NW_ID_RESERVED,
rest_nw_id.NW_ID_VPORT_GRE,
rest_nw_id.NW_ID_UNKNOWN,
]
return mock.patch.dict('sys.modules',
{'ryu': ryu_mod,
'ryu.app': ryu_app_mod,
'ryu.app.client': ryu_app_client,
'ryu.app.conf_switch_key': conf_switch_key,
'ryu.app.rest_nw_id': rest_nw_id})

4
quantum/tests/unit/ryu/test_defaults.py
View File

@ -28,6 +28,8 @@ class ConfigurationTest(unittest2.TestCase):
self.assertEqual('br-int', cfg.CONF.OVS.integration_bridge)
self.assertEqual(-1, cfg.CONF.DATABASE.sql_max_retries)
self.assertEqual(2, cfg.CONF.DATABASE.reconnect_interval)
self.assertEqual(2, cfg.CONF.AGENT.polling_interval)
self.assertEqual('sudo', cfg.CONF.AGENT.root_helper)
self.assertEqual('127.0.0.1:8080', cfg.CONF.OVS.openflow_rest_api)
self.assertEqual(1, cfg.CONF.OVS.tunnel_key_min)
self.assertEqual(0xffffff, cfg.CONF.OVS.tunnel_key_max)
self.assertEqual(6634, cfg.CONF.OVS.ovsdb_port)

497
quantum/tests/unit/ryu/test_ryu_agent.py Normal file
View File

@ -0,0 +1,497 @@
# Copyright (c) 2013 OpenStack, LLC.
#
# 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 contextlib import nested
import httplib
import mock
import unittest2 as unittest
from quantum.openstack.common import importutils
from quantum.tests.unit.ryu import fake_ryu
class RyuAgentTestCase(unittest.TestCase):
_AGENT_NAME = 'quantum.plugins.ryu.agent.ryu_quantum_agent'
def setUp(self):
self.addCleanup(mock.patch.stopall)
self.fake_ryu = fake_ryu.patch_fake_ryu_client().start()
self.mod_agent = importutils.import_module(self._AGENT_NAME)
class TestOVSQuantumOFPRyuAgent(RyuAgentTestCase):
def setUp(self):
super(TestOVSQuantumOFPRyuAgent, self).setUp()
self.plugin_api = mock.patch(
self._AGENT_NAME + '.RyuPluginApi').start()
self.ovsbridge = mock.patch(
self._AGENT_NAME + '.OVSBridge').start()
self.vifportset = mock.patch(
self._AGENT_NAME + '.VifPortSet').start()
self.q_ctx = mock.patch(
self._AGENT_NAME + '.q_context').start()
def mock_rest_addr(self, rest_addr):
integ_br = 'integ_br'
tunnel_ip = '192.168.0.1'
ovsdb_ip = '172.16.0.1'
ovsdb_port = 16634
root_helper = 'helper'
self.mod_agent.OVSBridge.return_value.datapath_id = '1234'
mock_context = mock.Mock(return_value='abc')
self.q_ctx.get_admin_context_without_session = mock_context
mock_rest_addr = mock.Mock(return_value=rest_addr)
self.plugin_api.return_value.get_ofp_rest_api_addr = mock_rest_addr
# Instantiate OVSQuantumOFPRyuAgent
self.agent = self.mod_agent.OVSQuantumOFPRyuAgent(
integ_br, tunnel_ip, ovsdb_ip, ovsdb_port, root_helper)
def test_valid_rest_addr(self):
self.mock_rest_addr('192.168.0.1:8080')
# OVSBridge
self.ovsbridge.assert_has_calls([
mock.call('integ_br', 'helper'),
mock.call().find_datapath_id()
])
# RyuPluginRpc
self.plugin_api.assert_has_calls([
mock.call('q-plugin'),
mock.call().get_ofp_rest_api_addr('abc')
])
# OFPClient
self.mod_agent.client.OFPClient.assert_calls([
mock.call('192.168.0.1:8080')
])
# VifPortSet
self.vifportset.assert_has_calls([
mock.call(
self.ovsbridge.return_value,
self.mod_agent.client.OFPClient.return_value),
mock.call().setup()
])
# SwitchConfClient
self.mod_agent.client.SwitchConfClient.assert_has_calls([
mock.call('192.168.0.1:8080'),
mock.call().set_key('1234', 'ovs_tunnel_addr', '192.168.0.1'),
mock.call().set_key('1234', 'ovsdb_addr',
'tcp:%s:%d' % ('172.16.0.1', 16634))
])
# OVSBridge
self.ovsbridge.return_value.set_manager.assert_calls([
mock.call('ptcp:%d' % 16634)
])
def test_invalid_rest_addr(self):
self.assertRaises(self.mod_agent.q_exc.Invalid,
self.mock_rest_addr, (''))
class TestRyuPluginApi(RyuAgentTestCase):
def test_get_ofp_rest_api_addr(self):
with nested(
mock.patch(self._AGENT_NAME + '.RyuPluginApi.make_msg',
return_value='msg'),
mock.patch(self._AGENT_NAME + '.RyuPluginApi.call',
return_value='10.0.0.1')
) as (mock_msg, mock_call):
api = self.mod_agent.RyuPluginApi('topics')
addr = api.get_ofp_rest_api_addr('context')
self.assertEqual(addr, '10.0.0.1')
mock_msg.assert_calls([
mock.call('get_ofp_rest_api')
])
mock_call.assert_calls([
mock.call('context', 'msg', topic='topics')
])
class TestVifPortSet(RyuAgentTestCase):
def test_setup(self):
attrs = {'switch.datapath_id': 'dp1', 'ofport': 'p1'}
p1 = mock.Mock(**attrs)
attrs = {'switch.datapath_id': 'dp2', 'ofport': 'p2'}
p2 = mock.Mock(**attrs)
attrs = {'get_external_ports.return_value': [p1, p2]}
int_br = mock.Mock(**attrs)
with mock.patch(self._AGENT_NAME + '.client.OFPClient') as client:
api = client()
vif = self.mod_agent.VifPortSet(int_br, api)
vif.setup()
client.assert_has_calls([
mock.call().update_port('__NW_ID_EXTERNAL__', 'dp1', 'p1'),
mock.call().update_port('__NW_ID_EXTERNAL__', 'dp2', 'p2')
])
def test_setup_empty(self):
attrs = {'get_external_ports.return_value': []}
int_br = mock.Mock(**attrs)
api = mock.Mock()
vif = self.mod_agent.VifPortSet(int_br, api)
vif.setup()
self.assertEqual(api.update_port.call_count, 0)
class TestOVSBridge(RyuAgentTestCase):
def setUp(self):
super(TestOVSBridge, self).setUp()
self.lib_ovs = mock.patch(
'quantum.agent.linux.ovs_lib.OVSBridge').start()
def test_find_datapath_id(self):
with mock.patch(self._AGENT_NAME + '.OVSBridge.get_datapath_id',
return_value='1234') as mock_get_dpid:
br = self.mod_agent.OVSBridge('br_name', 'helper')
br.find_datapath_id()
mock_get_dpid.assert_calls([
mock.call()
])
self.assertEqual(br.datapath_id, '1234')
def test_set_manager(self):
with mock.patch(
self._AGENT_NAME + '.OVSBridge.run_vsctl') as mock_vsctl:
br = self.mod_agent.OVSBridge('br_name', 'helper')
br.set_manager('target')
mock_vsctl.assert_calls([
mock.call(['set-manager', 'target'])
])
def test_get_ofport(self):
with mock.patch(
self._AGENT_NAME + '.OVSBridge.db_get_val',
return_value=1) as mock_db:
br = self.mod_agent.OVSBridge('br_name', 'helper')
ofport = br.get_ofport('name')
mock_db.assert_calls([
mock.call('Interface', 'name', 'ofport')
])
self.assertEqual(ofport, 1)
def test_get_ports(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.get_port_name_list',
return_value=['p1', 'p2']),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
return_value=1)
) as (mock_name, mock_ofport):
get_port = mock.Mock(side_effect=['port1', 'port2'])
br = self.mod_agent.OVSBridge('br_name', 'helper')
ports = br._get_ports(get_port)
mock_name.assert_calls([
mock.call()
])
mock_ofport.assert_calls([
mock.call('p1'),
mock.call('p2')
])
get_port.assert_calls([
mock.call('p1'),
mock.call('p2')
])
self.assertEqual(len(ports), 2)
self.assertEqual(ports, ['port1', 'port2'])
def test_get_ports_empty(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.get_port_name_list',
return_value=[]),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
return_value=1)
) as (mock_name, mock_ofport):
get_port = mock.Mock(side_effect=['port1', 'port2'])
br = self.mod_agent.OVSBridge('br_name', 'helper')
ports = br._get_ports(get_port)
mock_name.assert_calls([
mock.call()
])
self.assertEqual(mock_ofport.call_count, 0)
self.assertEqual(get_port.call_count, 0)
self.assertEqual(len(ports), 0)
def test_get_ports_invalid_ofport(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.get_port_name_list',
return_value=['p1', 'p2']),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
side_effect=[-1, 1])
) as (mock_name, mock_ofport):
get_port = mock.Mock(side_effect=['port1', 'port2'])
br = self.mod_agent.OVSBridge('br_name', 'helper')
ports = br._get_ports(get_port)
mock_name.assert_calls([
mock.call()
])
mock_ofport.assert_calls([
mock.call('p1'),
mock.call('p2')
])
get_port.assert_calls([
mock.call('p2')
])
self.assertEqual(len(ports), 1)
self.assertEqual(ports, ['port1'])
def test_get_ports_invalid_port(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.get_port_name_list',
return_value=['p1', 'p2']),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
side_effect=[1, 2])
) as (mock_name, mock_ofport):
get_port = mock.Mock(side_effect=[None, 'port2'])
br = self.mod_agent.OVSBridge('br_name', 'helper')
ports = br._get_ports(get_port)
mock_name.assert_calls([
mock.call()
])
mock_ofport.assert_calls([
mock.call('p1'),
mock.call('p2')
])
get_port.assert_calls([
mock.call('p1'),
mock.call('p2')
])
self.assertEqual(len(ports), 1)
self.assertEqual(ports, ['port2'])
def test_get_external_port(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.db_get_map',
side_effect=[None, {'opts': 'opts_val'}]),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
return_value=1),
mock.patch(self._AGENT_NAME + '.VifPort')
) as (mock_db, mock_ofport, mock_vif):
br = self.mod_agent.OVSBridge('br_name', 'helper')
vifport = br._get_external_port('iface')
mock_db.assert_calls([
mock.call('Interface', 'iface', 'external_ids'),
mock.call('Interface', 'iface', 'options'),
])
mock_ofport.assert_calls([
mock.call('iface')
])
mock_vif.assert_calls([
mock.call('iface', 1, None, None, br)
])
self.assertEqual(vifport, mock_vif.return_value)
def test_get_external_port_vmport(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.db_get_map',
side_effect=[{'extids': 'extid_val'},
{'opts': 'opts_val'}]),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
return_value=1),
mock.patch(self._AGENT_NAME + '.VifPort')
) as (mock_db, mock_ofport, mock_vif):
br = self.mod_agent.OVSBridge('br_name', 'helper')
vifport = br._get_external_port('iface')
mock_db.assert_calls([
mock.call('Interface', 'iface', 'external_ids'),
])
self.assertEqual(mock_ofport.call_count, 0)
self.assertEqual(mock_vif.call_count, 0)
self.assertEqual(vifport, None)
def test_get_external_port_tunnel(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge.db_get_map',
side_effect=[None, {'remote_ip': '0.0.0.0'}]),
mock.patch(self._AGENT_NAME + '.OVSBridge.get_ofport',
return_value=1),
mock.patch(self._AGENT_NAME + '.VifPort')
) as (mock_db, mock_ofport, mock_vif):
br = self.mod_agent.OVSBridge('br_name', 'helper')
vifport = br._get_external_port('iface')
mock_db.assert_calls([
mock.call('Interface', 'iface', 'external_ids'),
mock.call('Interface', 'iface', 'options'),
])
self.assertEqual(mock_ofport.call_count, 0)
self.assertEqual(mock_vif.call_count, 0)
self.assertEqual(vifport, None)
def test_get_external_ports(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSBridge._get_external_port'),
mock.patch(self._AGENT_NAME + '.OVSBridge._get_ports')
) as (mock_extport, mock_port):
br = self.mod_agent.OVSBridge('br_name', 'helper')
br.get_external_ports()
mock_port.assert_calls([
mock.call(mock_extport)
])
class TestRyuQuantumAgent(RyuAgentTestCase):
def test_get_my_ip(self):
sock_attrs = {
'return_value.getsockname.return_value': ['1.2.3.4', '']}
with mock.patch('socket.socket', **sock_attrs):
addr = self.mod_agent._get_my_ip()
self.assertEqual(addr, '1.2.3.4')
def test_get_ip_ip(self):
cfg_attrs = {'cfg_ip': '1.2.3.4',
'cfg_iface': 'eth0'}
netifs_attrs = {'AF_INET': 0,
'ifaddresses.return_value': [[{'addr': '10.0.0.1'}]]}
with nested(
mock.patch('quantum.openstack.common.cfg.CONF.OVS', **cfg_attrs),
mock.patch(self._AGENT_NAME + '.netifaces', **netifs_attrs),
mock.patch(self._AGENT_NAME + '._get_my_ip',
return_value='172.16.0.1')
) as (_cfg, mock_netifs, mock_myip):
ip = self.mod_agent._get_ip('cfg_ip', 'cfg_iface')
self.assertEqual(mock_netifs.ifaddresses.call_count, 0)
self.assertEqual(mock_myip.call_count, 0)
self.assertEqual(ip, '1.2.3.4')
def test_get_ip_iface(self):
cfg_attrs = {'cfg_ip': None,
'cfg_iface': 'eth0'}
netifs_attrs = {'AF_INET': 0,
'ifaddresses.return_value': [[{'addr': '10.0.0.1'}]]}
with nested(
mock.patch('quantum.openstack.common.cfg.CONF.OVS', **cfg_attrs),
mock.patch(self._AGENT_NAME + '.netifaces', **netifs_attrs),
mock.patch(self._AGENT_NAME + '._get_my_ip',
return_value='172.16.0.1')
) as (_cfg, mock_netifs, mock_myip):
ip = self.mod_agent._get_ip('cfg_ip', 'cfg_iface')
mock_netifs.ifaddresses.assert_calls([
mock.call('eth0')
])
self.assertEqual(mock_myip.call_count, 0)
self.assertEqual(ip, '10.0.0.1')
def test_get_ip_myip(self):
cfg_attrs = {'cfg_ip': None,
'cfg_iface': None}
netifs_attrs = {'AF_INET': 0,
'ifaddresses.return_value': [[{'addr': '10.0.0.1'}]]}
with nested(
mock.patch('quantum.openstack.common.cfg.CONF.OVS', **cfg_attrs),
mock.patch(self._AGENT_NAME + '.netifaces', **netifs_attrs),
mock.patch(self._AGENT_NAME + '._get_my_ip',
return_value='172.16.0.1')
) as (_cfg, mock_netifs, mock_myip):
ip = self.mod_agent._get_ip('cfg_ip', 'cfg_iface')
self.assertEqual(mock_netifs.ifaddresses.call_count, 0)
mock_myip.assert_calls([
mock.call()
])
self.assertEqual(ip, '172.16.0.1')
def test_get_tunnel_ip(self):
with mock.patch(self._AGENT_NAME + '._get_ip',
return_value='1.2.3.4') as mock_getip:
ip = self.mod_agent._get_tunnel_ip()
mock_getip.assert_calls([
mock.call('tunnel_ip', 'tunnel_interface')
])
self.assertEqual(ip, '1.2.3.4')
def test_get_ovsdb_ip(self):
with mock.patch(self._AGENT_NAME + '._get_ip',
return_value='1.2.3.4') as mock_getip:
ip = self.mod_agent._get_ovsdb_ip()
mock_getip.assert_calls([
mock.call('ovsdb_ip', 'ovsdb_interface')
])
self.assertEqual(ip, '1.2.3.4')
def mock_main(self):
cfg_attrs = {'OVS.integration_bridge': 'integ_br',
'OVS.ovsdb_port': 16634,
'AGENT.root_helper': 'helper'}
with nested(
mock.patch('quantum.openstack.common.cfg.CONF', **cfg_attrs),
mock.patch(self._AGENT_NAME + '.logging_config'),
mock.patch(self._AGENT_NAME + '._get_tunnel_ip',
return_value='10.0.0.1'),
mock.patch(self._AGENT_NAME + '._get_ovsdb_ip',
return_value='172.16.0.1'),
) as (mock_conf, mock_log_conf, _tun, _ovsdb):
self.mod_agent.main()
mock_log_conf.assert_calls([
mock.call(mock_conf)
])
def test_main(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSQuantumOFPRyuAgent'),
mock.patch('sys.exit', side_effect=SystemExit(0))
) as (mock_agent, mock_exit):
self.assertRaises(SystemExit, self.mock_main)
mock_agent.assert_calls([
mock.call('integ_br', '10.0.0.1', '172.16.0.1', 16634, 'helper')
])
mock_exit.assert_calls([
mock.call(0)
])
def test_main_raise(self):
with nested(
mock.patch(self._AGENT_NAME + '.OVSQuantumOFPRyuAgent',
side_effect=httplib.HTTPException('boom')),
mock.patch('sys.exit', side_effect=SystemExit(0))
) as (mock_agent, mock_exit):
self.assertRaises(SystemExit, self.mock_main)
mock_agent.assert_calls([
mock.call('integ_br', '10.0.0.1', '172.16.0.1', 16634, 'helper')
])
mock_exit.assert_calls([
mock.call(1)
])

24
quantum/tests/unit/ryu/test_ryu_plugin.py
View File

@ -15,6 +15,7 @@
import mock
from quantum.tests.unit.ryu import fake_ryu
from quantum.tests.unit import test_db_plugin as test_plugin
@ -22,29 +23,8 @@ class RyuPluginV2TestCase(test_plugin.QuantumDbPluginV2TestCase):
_plugin_name = 'quantum.plugins.ryu.ryu_quantum_plugin.RyuQuantumPluginV2'
def _patch_fake_ryu_client(self):
ryu_mod = mock.Mock()
ryu_app_mod = ryu_mod.app
ryu_app_client = ryu_app_mod.client
rest_nw_id = ryu_app_mod.rest_nw_id
rest_nw_id.NW_ID_EXTERNAL = '__NW_ID_EXTERNAL__'
rest_nw_id.NW_ID_RESERVED = '__NW_ID_RESERVED__'
rest_nw_id.NW_ID_VPORT_GRE = '__NW_ID_VPORT_GRE__'
rest_nw_id.NW_ID_UNKNOWN = '__NW_ID_UNKNOWN__'
rest_nw_id.RESERVED_NETWORK_IDS = [
rest_nw_id.NW_ID_EXTERNAL,
rest_nw_id.NW_ID_RESERVED,
rest_nw_id.NW_ID_VPORT_GRE,
rest_nw_id.NW_ID_UNKNOWN,
]
return mock.patch.dict('sys.modules',
{'ryu': ryu_mod,
'ryu.app': ryu_app_mod,
'ryu.app.client': ryu_app_client,
'ryu.app.rest_nw_id': rest_nw_id})
def setUp(self):
self.ryu_patcher = self._patch_fake_ryu_client()
self.ryu_patcher = fake_ryu.patch_fake_ryu_client()
self.ryu_patcher.start()
super(RyuPluginV2TestCase, self).setUp(self._plugin_name)

5
tools/test-requires
View File

@ -16,3 +16,8 @@ webtest==1.3.3
###############################
configobj
###############################
# Packages for the Ryu Plugin
###############################
netifaces
###############################