openstack/charm-neutron-openvswitch
Code Issues Proposed changes

More unit tests

Browse Source
This commit is contained in:
Liam Young 2015-02-25 13:50:38 +00:00
parent 0990ac13b3
commit 8b5d14e345
5 changed files with 178 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
hooks/neutron_ovs_context.py
View File

@ -9,9 +9,6 @@ from charmhelpers.core.hookenv import (
)
from charmhelpers.contrib.network.ip import (
get_address_in_network,
get_ipv4_addr,
get_ipv6_addr,
is_bridge_member,
)
from charmhelpers.contrib.openstack.ip import resolve_address
from charmhelpers.core.host import list_nics, get_nic_hwaddr
@ -175,24 +172,27 @@ class NetworkServiceContext(OSContextGenerator):
return ctxt
SHARED_SECRET = "/etc/neutron/secret.txt"
def get_shared_secret():
secret = None
if not os.path.exists(SHARED_SECRET):
secret = str(uuid.uuid4())
with open(SHARED_SECRET, 'w') as secret_file:
secret_file.write(secret)
else:
with open(SHARED_SECRET, 'r') as secret_file:
secret = secret_file.read().strip()
return secret
class DVRSharedSecretContext(OSContextGenerator):
def get_shared_secret(self):
secret = None
if not os.path.exists(self.SHARED_SECRET):
secret = str(uuid.uuid4())
with open(self.SHARED_SECRET, 'w') as secret_file:
secret_file.write(secret)
else:
with open(self.SHARED_SECRET, 'r') as secret_file:
secret = secret_file.read().strip()
return secret
def __call__(self):
self.SHARED_SECRET = "/etc/neutron/secret.txt"
if use_dvr():
ctxt = {
'shared_secret': self.get_shared_secret(),
'shared_secret': get_shared_secret(),
'local_ip': resolve_address(),
}
else:

2
hooks/neutron_ovs_hooks.py
View File

@ -51,6 +51,7 @@ def config_changed():
configure_ovs()
CONFIGS.write_all()
@hooks.hook('neutron-plugin-api-relation-changed')
@restart_on_change(restart_map())
def neutron_plugin_api_changed():
@ -63,6 +64,7 @@ def neutron_plugin_api_changed():
for rid in relation_ids('neutron-plugin'):
neutron_plugin_joined(relation_id=rid)
@hooks.hook('neutron-plugin-relation-joined')
def neutron_plugin_joined(relation_id=None):
rel_data = {

81
unit_tests/test_neutron_ovs_context.py
View File

@ -1,5 +1,6 @@
from test_utils import CharmTestCase
from test_utils import patch_open
from mock import patch
import neutron_ovs_context as context
import charmhelpers
@ -7,6 +8,7 @@ TO_PATCH = [
'relation_get',
'relation_ids',
'related_units',
'resolve_address',
'config',
'unit_get',
'add_bridge',
@ -169,6 +171,7 @@ class OVSPluginContextTest(CharmTestCase):
self.assertEquals(expect, napi_ctxt())
self.service_start.assertCalled()
class L3AgentContextTest(CharmTestCase):
def setUp(self):
@ -196,3 +199,81 @@ class L3AgentContextTest(CharmTestCase):
'l2-population': 'True',
'overlay-network-type': 'vxlan'})
self.assertEquals(context.L3AgentContext()(), {'agent_mode': 'legacy'})
class NetworkServiceContext(CharmTestCase):
def setUp(self):
super(NetworkServiceContext, self).setUp(context, TO_PATCH)
self.relation_get.side_effect = self.test_relation.get
self.config.side_effect = self.test_config.get
def tearDown(self):
super(NetworkServiceContext, self).tearDown()
def test_network_svc_ctxt(self):
self.related_units.return_value = ['unit1']
self.relation_ids.return_value = ['rid2']
self.test_relation.set({'service_protocol': 'http',
'keystone_host': '10.0.0.10',
'service_port': '8080',
'region': 'region1',
'service_tenant': 'tenant',
'service_username': 'bob',
'service_password': 'reallyhardpass'})
self.assertEquals(context.NetworkServiceContext()(),
{'service_protocol': 'http',
'keystone_host': '10.0.0.10',
'service_port': '8080',
'region': 'region1',
'service_tenant': 'tenant',
'service_username': 'bob',
'service_password': 'reallyhardpass'})
class DVRSharedSecretContext(CharmTestCase):
def setUp(self):
super(DVRSharedSecretContext, self).setUp(context,
TO_PATCH)
self.config.side_effect = self.test_config.get
@patch('os.path')
@patch('uuid.uuid4')
def test_secret_created_stored(self, _uuid4, _path):
_path.exists.return_value = False
_uuid4.return_value = 'secret_thing'
with patch_open() as (_open, _file):
self.assertEquals(context.get_shared_secret(),
'secret_thing')
_open.assert_called_with(
context.SHARED_SECRET.format('quantum'), 'w')
_file.write.assert_called_with('secret_thing')
@patch('os.path')
def test_secret_retrieved(self, _path):
_path.exists.return_value = True
with patch_open() as (_open, _file):
_file.read.return_value = 'secret_thing\n'
self.assertEquals(context.get_shared_secret(),
'secret_thing')
_open.assert_called_with(
context.SHARED_SECRET.format('quantum'), 'r')
@patch.object(context, 'use_dvr')
@patch.object(context, 'get_shared_secret')
def test_shared_secretcontext_dvr(self, _shared_secret, _use_dvr):
_shared_secret.return_value = 'secret_thing'
_use_dvr.return_value = True
self.resolve_address.return_value = '10.0.0.10'
self.assertEquals(context.DVRSharedSecretContext()(),
{'shared_secret': 'secret_thing',
'local_ip': '10.0.0.10'})
@patch.object(context, 'use_dvr')
@patch.object(context, 'get_shared_secret')
def test_shared_secretcontext_nodvr(self, _shared_secret, _use_dvr):
_shared_secret.return_value = 'secret_thing'
_use_dvr.return_value = False
self.resolve_address.return_value = '10.0.0.10'
self.assertEquals(context.DVRSharedSecretContext()(), {})

36
unit_tests/test_neutron_ovs_hooks.py
View File

@ -24,7 +24,10 @@ TO_PATCH = [
'config',
'CONFIGS',
'determine_packages',
'determine_dvr_packages',
'get_shared_secret',
'log',
'relation_ids',
'relation_set',
'configure_ovs',
]
@ -61,6 +64,39 @@ class NeutronOVSHooksTests(CharmTestCase):
self.assertTrue(self.CONFIGS.write_all.called)
self.configure_ovs.assert_called_with()
@patch.object(neutron_ovs_context, 'use_dvr')
def test_config_changed_dvr(self, _use_dvr):
_use_dvr.return_value = True
self.determine_dvr_packages.return_value = ['dvr']
self._call_hook('config-changed')
self.apt_update.assert_called_with()
self.assertTrue(self.CONFIGS.write_all.called)
self.apt_install.assert_has_calls([
call(['dvr'], fatal=True),
])
self.configure_ovs.assert_called_with()
@patch.object(hooks, 'neutron_plugin_joined')
@patch.object(neutron_ovs_context, 'use_dvr')
def test_neutron_plugin_api(self, _use_dvr, _plugin_joined):
_use_dvr.return_value = False
self.relation_ids.return_value = ['rid']
self._call_hook('neutron-plugin-api-relation-changed')
self.configure_ovs.assert_called_with()
self.assertTrue(self.CONFIGS.write_all.called)
_plugin_joined.assert_called_with(relation_id='rid')
def test_neutron_plugin_joined(self):
self.get_shared_secret.return_value = 'secret'
self._call_hook('neutron-plugin-relation-joined')
rel_data = {
'metadata-shared-secret': 'secret',
}
self.relation_set.assert_called_with(
relation_id=None,
**rel_data
)
def test_amqp_joined(self):
self._call_hook('amqp-relation-joined')
self.relation_set.assert_called_with(

44
unit_tests/test_neutron_ovs_utils.py
View File

@ -1,7 +1,8 @@
from mock import MagicMock, patch
from mock import MagicMock, patch, call
from collections import OrderedDict
import charmhelpers.contrib.openstack.templating as templating
from charmhelpers.contrib.openstack import context
templating.OSConfigRenderer = MagicMock()
@ -16,6 +17,8 @@ import charmhelpers.core.hookenv as hookenv
TO_PATCH = [
'add_bridge',
'add_bridge_port',
'os_release',
'neutron_plugin_attribute',
]
@ -39,6 +42,15 @@ def _mock_npa(plugin, attr, net_manager=None):
return plugins[plugin][attr]
class DummyContext():
def __init__(self, return_value):
self.return_value = return_value
def __call__(self):
return self.return_value
class TestNeutronOVSUtils(CharmTestCase):
def setUp(self):
@ -83,8 +95,20 @@ class TestNeutronOVSUtils(CharmTestCase):
def test_resource_map(self, _use_dvr):
_use_dvr.return_value = False
_map = nutils.resource_map()
svcs = ['neutron-plugin-openvswitch-agent']
confs = [nutils.NEUTRON_CONF]
[self.assertIn(q_conf, _map.keys()) for q_conf in confs]
self.assertEqual(_map[nutils.NEUTRON_CONF]['services'], svcs)
@patch.object(neutron_ovs_context, 'use_dvr')
def test_resource_map_dvr(self, _use_dvr):
_use_dvr.return_value = True
_map = nutils.resource_map()
svcs = ['neutron-plugin-openvswitch-agent', 'neutron-metadata-agent',
'neutron-vpn-agent']
confs = [nutils.NEUTRON_CONF]
[self.assertIn(q_conf, _map.keys()) for q_conf in confs]
self.assertEqual(_map[nutils.NEUTRON_CONF]['services'], svcs)
@patch.object(neutron_ovs_context, 'use_dvr')
def test_restart_map(self, _use_dvr):
@ -99,3 +123,21 @@ class TestNeutronOVSUtils(CharmTestCase):
for item in _restart_map:
self.assertTrue(item in _restart_map)
self.assertTrue(expect[item] == _restart_map[item])
@patch.object(context, 'ExternalPortContext')
def test_configure_ovs_ovs_ext_port(self, _ext_port_ctxt):
_ext_port_ctxt.return_value = \
DummyContext(return_value={'ext_port': 'eth0'})
nutils.configure_ovs()
self.add_bridge.assert_has_calls([
call('br-int'),
call('br-ex'),
call('br-data')
])
self.add_bridge_port.assert_called_with('br-ex', 'eth0')
@patch.object(neutron_ovs_context, 'DVRSharedSecretContext')
def test_get_shared_secret(self, _dvr_secret_ctxt):
_dvr_secret_ctxt.return_value = \
DummyContext(return_value={'shared_secret': 'supersecret'})
self.assertEqual(nutils.get_shared_secret(), 'supersecret')