openstack/charm-neutron-gateway
Code Issues Proposed changes

Added unit tests

Browse Source
This commit is contained in:
James Page 2013-07-22 16:39:07 +01:00
parent 753896ba44
commit c72636d3c1
8 changed files with 719 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
hooks/quantum_contexts.py
View File

@ -1,15 +1,39 @@
# vim: set ts=4:et
import os
import uuid
import socket
from charmhelpers.core.hookenv import (
config,
relation_ids,
related_units,
relation_get,
unit_get,
cached,
)
from charmhelpers.core.host import (
apt_install,
)
from charmhelpers.contrib.openstack.context import (
OSContextGenerator,
context_complete
)
import quantum_utils as qutils
DB_USER = "quantum"
QUANTUM_DB = "quantum"
NOVA_DB_USER = "nova"
NOVA_DB = "nova"
OVS = "ovs"
NVP = "nvp"
OVS_PLUGIN = \
"quantum.plugins.openvswitch.ovs_quantum_plugin.OVSQuantumPluginV2"
NVP_PLUGIN = \
"quantum.plugins.nicira.nicira_nvp_plugin.QuantumPlugin.NvpPluginV2"
CORE_PLUGIN = {
OVS: OVS_PLUGIN,
NVP: NVP_PLUGIN
}
class NetworkServiceContext(OSContextGenerator):
@ -58,9 +82,9 @@ class ExternalPortContext(OSContextGenerator):
class QuantumGatewayContext(OSContextGenerator):
def __call__(self):
ctxt = {
'shared_secret': qutils.get_shared_secret(),
'local_ip': qutils.get_host_ip(),
'core_plugin': qutils.CORE_PLUGIN[config('plugin')],
'shared_secret': get_shared_secret(),
'local_ip': get_host_ip(),
'core_plugin': CORE_PLUGIN[config('plugin')],
'plugin': config('plugin')
}
return ctxt
@ -75,15 +99,49 @@ class QuantumSharedDBContext(OSContextGenerator):
ctxt = {
'database_host': relation_get('db_host', rid=rid,
unit=unit),
'quantum_database': qutils.QUANTUM_DB,
'quantum_user': qutils.DB_USER,
'quantum_database': QUANTUM_DB,
'quantum_user': DB_USER,
'quantum_password': relation_get('quantum_password',
rid=rid, unit=unit),
'nova_database': qutils.NOVA_DB,
'nova_user': qutils.NOVA_DB_USER,
'nova_database': NOVA_DB,
'nova_user': NOVA_DB_USER,
'nova_password': relation_get('nova_password', rid=rid,
unit=unit)
}
print ctxt
if context_complete(ctxt):
return ctxt
return {}
@cached
def get_host_ip(hostname=None):
try:
import dns.resolver
except ImportError:
apt_install('python-dnspython', fatal=True)
import dns.resolver
hostname = hostname or unit_get('private-address')
try:
# Test to see if already an IPv4 address
socket.inet_aton(hostname)
return hostname
except socket.error:
answers = dns.resolver.query(hostname, 'A')
if answers:
return answers[0].address
SHARED_SECRET = "/etc/quantum/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

6
hooks/quantum_hooks.py
View File

@ -33,11 +33,13 @@ from quantum_utils import (
get_packages,
get_early_packages,
valid_plugin,
DB_USER, QUANTUM_DB,
NOVA_DB_USER, NOVA_DB,
configure_ovs,
reassign_agent_resources,
)
from quantum_contexts import (
DB_USER, QUANTUM_DB,
NOVA_DB_USER, NOVA_DB,
)
hooks = Hooks()
CONFIGS = register_configs()

90
hooks/quantum_utils.py
View File

@ -1,11 +1,6 @@
import os
import uuid
import socket
from charmhelpers.core.hookenv import (
log,
config,
unit_get,
cached
)
from charmhelpers.core.host import (
apt_install,
@ -22,23 +17,19 @@ from charmhelpers.contrib.openstack.utils import (
)
import charmhelpers.contrib.openstack.context as context
import charmhelpers.contrib.openstack.templating as templating
import quantum_contexts
from quantum_contexts import (
CORE_PLUGIN,
OVS, NVP,
QuantumGatewayContext,
NetworkServiceContext,
QuantumSharedDBContext,
ExternalPortContext,
)
from collections import OrderedDict
OVS = "ovs"
NVP = "nvp"
OVS_PLUGIN = \
"quantum.plugins.openvswitch.ovs_quantum_plugin.OVSQuantumPluginV2"
NVP_PLUGIN = \
"quantum.plugins.nicira.nicira_nvp_plugin.QuantumPlugin.NvpPluginV2"
CORE_PLUGIN = {
OVS: OVS_PLUGIN,
NVP: NVP_PLUGIN
}
def valid_plugin():
print config('plugin')
return config('plugin') in CORE_PLUGIN
OVS_PLUGIN_CONF = \
@ -94,18 +85,18 @@ NOVA_CONF = "/etc/nova/nova.conf"
SHARED_CONFIG_FILES = {
DHCP_AGENT_CONF: {
'hook_contexts': [quantum_contexts.QuantumGatewayContext()],
'hook_contexts': [QuantumGatewayContext()],
'services': ['quantum-dhcp-agent']
},
METADATA_AGENT_CONF: {
'hook_contexts': [quantum_contexts.NetworkServiceContext()],
'hook_contexts': [NetworkServiceContext()],
'services': ['quantum-metadata-agent']
},
NOVA_CONF: {
'hook_contexts': [context.AMQPContext(),
quantum_contexts.QuantumSharedDBContext(),
quantum_contexts.NetworkServiceContext(),
quantum_contexts.QuantumGatewayContext()],
QuantumSharedDBContext(),
NetworkServiceContext(),
QuantumGatewayContext()],
'services': ['nova-api-metadata']
},
}
@ -113,24 +104,24 @@ SHARED_CONFIG_FILES = {
OVS_CONFIG_FILES = {
QUANTUM_CONF: {
'hook_contexts': [context.AMQPContext(),
quantum_contexts.QuantumGatewayContext()],
QuantumGatewayContext()],
'services': ['quantum-l3-agent',
'quantum-dhcp-agent',
'quantum-metadata-agent',
'quantum-plugin-openvswitch-agent']
},
L3_AGENT_CONF: {
'hook_contexts': [quantum_contexts.NetworkServiceContext()],
'hook_contexts': [NetworkServiceContext()],
'services': ['quantum-l3-agent']
},
# TODO: Check to see if this is actually required
OVS_PLUGIN_CONF: {
'hook_contexts': [quantum_contexts.QuantumSharedDBContext(),
quantum_contexts.QuantumGatewayContext()],
'hook_contexts': [QuantumSharedDBContext(),
QuantumGatewayContext()],
'services': ['quantum-plugin-openvswitch-agent']
},
EXT_PORT_CONF: {
'hook_contexts': [quantum_contexts.ExternalPortContext()],
'hook_contexts': [ExternalPortContext()],
'services': []
}
}
@ -182,39 +173,16 @@ def restart_map():
return OrderedDict(_map)
DB_USER = "quantum"
QUANTUM_DB = "quantum"
KEYSTONE_SERVICE = "quantum"
NOVA_DB_USER = "nova"
NOVA_DB = "nova"
RABBIT_USER = "nova"
RABBIT_VHOST = "nova"
INT_BRIDGE = "br-int"
EXT_BRIDGE = "br-ex"
SHARED_SECRET = "/etc/quantum/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
DHCP_AGENT = "DHCP Agent"
L3_AGENT = "L3 Agent"
def reassign_agent_resources():
''' Use agent scheduler API to detect down agents and re-schedule '''
env = quantum_contexts.NetworkServiceContext()()
env = NetworkServiceContext()()
if not env:
log('Unable to re-assign resources at this time')
return
@ -305,24 +273,6 @@ def do_openstack_upgrade(configs):
configs.set_release(openstack_release=new_os_rel)
@cached
def get_host_ip(hostname=None):
try:
import dns.resolver
except ImportError:
apt_install('python-dnspython', fatal=True)
import dns.resolver
hostname = hostname or unit_get('private-address')
try:
# Test to see if already an IPv4 address
socket.inet_aton(hostname)
return hostname
except socket.error:
answers = dns.resolver.query(hostname, 'A')
if answers:
return answers[0].address
def configure_ovs():
if config('plugin') == OVS:
add_bridge(INT_BRIDGE)

2
unit_tests/__init__.py
View File

@ -0,0 +1,2 @@
import sys
sys.path.append('hooks')

218
unit_tests/test_quantum_contexts.py Normal file
View File

@ -0,0 +1,218 @@
from mock import MagicMock, patch
import quantum_contexts
from contextlib import contextmanager
from test_utils import (
CharmTestCase
)
TO_PATCH = [
'config',
'relation_get',
'relation_ids',
'related_units',
'context_complete',
'unit_get',
'apt_install',
]
@contextmanager
def patch_open():
'''Patch open() to allow mocking both open() itself and the file that is
yielded.
Yields the mock for "open" and "file", respectively.'''
mock_open = MagicMock(spec=open)
mock_file = MagicMock(spec=file)
@contextmanager
def stub_open(*args, **kwargs):
mock_open(*args, **kwargs)
yield mock_file
with patch('__builtin__.open', stub_open):
yield mock_open, mock_file
class _TestQuantumContext(CharmTestCase):
def setUp(self):
super(_TestQuantumContext, self).setUp(quantum_contexts, TO_PATCH)
self.config.side_effect = self.test_config.get
def test_not_related(self):
self.relation_ids.return_value = []
self.assertEquals(self.context(), {})
def test_no_units(self):
self.relation_ids.return_value = []
self.relation_ids.return_value = ['foo']
self.related_units.return_value = []
self.assertEquals(self.context(), {})
def test_no_data(self):
self.relation_ids.return_value = ['foo']
self.related_units.return_value = ['bar']
self.relation_get.side_effect = self.test_relation.get
self.context_complete.return_value = False
self.assertEquals(self.context(), {})
def test_data_multi_unit(self):
self.relation_ids.return_value = ['foo']
self.related_units.return_value = ['bar', 'baz']
self.context_complete.return_value = True
self.relation_get.side_effect = self.test_relation.get
self.assertEquals(self.context(), self.data_result)
def test_data_single_unit(self):
self.relation_ids.return_value = ['foo']
self.related_units.return_value = ['bar']
self.context_complete.return_value = True
self.relation_get.side_effect = self.test_relation.get
self.assertEquals(self.context(), self.data_result)
class TestQuantumSharedDBContext(_TestQuantumContext):
def setUp(self):
super(TestQuantumSharedDBContext, self).setUp()
self.context = quantum_contexts.QuantumSharedDBContext()
self.test_relation.set(
{'db_host': '10.5.0.1',
'nova_password': 'novapass',
'quantum_password': 'quantumpass'}
)
self.data_result = {
'database_host': '10.5.0.1',
'nova_user': 'nova',
'nova_password': 'novapass',
'nova_database': 'nova',
'quantum_user': 'quantum',
'quantum_password': 'quantumpass',
'quantum_database': 'quantum'
}
class TestNetworkServiceContext(_TestQuantumContext):
def setUp(self):
super(TestNetworkServiceContext, self).setUp()
self.context = quantum_contexts.NetworkServiceContext()
self.test_relation.set(
{'keystone_host': '10.5.0.1',
'service_port': '5000',
'auth_port': '20000',
'service_tenant': 'tenant',
'service_username': 'username',
'service_password': 'password',
'quantum_host': '10.5.0.2',
'quantum_port': '9696',
'quantum_url': 'http://10.5.0.2:9696/v2',
'region': 'aregion'}
)
self.data_result = {
'keystone_host': '10.5.0.1',
'service_port': '5000',
'auth_port': '20000',
'service_tenant': 'tenant',
'service_username': 'username',
'service_password': 'password',
'quantum_host': '10.5.0.2',
'quantum_port': '9696',
'quantum_url': 'http://10.5.0.2:9696/v2',
'region': 'aregion',
'service_protocol': 'http',
'auth_protocol': 'http',
}
class TestExternalPortContext(CharmTestCase):
def setUp(self):
super(TestExternalPortContext, self).setUp(quantum_contexts,
TO_PATCH)
def test_no_ext_port(self):
self.config.return_value = None
self.assertEquals(quantum_contexts.ExternalPortContext()(),
None)
def test_ext_port(self):
self.config.return_value = 'eth1010'
self.assertEquals(quantum_contexts.ExternalPortContext()(),
{'ext_port': 'eth1010'})
class TestQuantumGatewayContext(CharmTestCase):
def setUp(self):
super(TestQuantumGatewayContext, self).setUp(quantum_contexts,
TO_PATCH)
@patch.object(quantum_contexts, 'get_shared_secret')
@patch.object(quantum_contexts, 'get_host_ip')
def test_all(self, _host_ip, _secret):
self.config.return_value = 'ovs'
_host_ip.return_value = '10.5.0.1'
_secret.return_value = 'testsecret'
self.assertEquals(quantum_contexts.QuantumGatewayContext()(),
{'shared_secret': 'testsecret',
'local_ip': '10.5.0.1',
'core_plugin':
"quantum.plugins.openvswitch.ovs_quantum_plugin."
"OVSQuantumPluginV2",
'plugin': 'ovs'})
class TestSharedSecret(CharmTestCase):
def setUp(self):
super(TestSharedSecret, self).setUp(quantum_contexts,
TO_PATCH)
@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(quantum_contexts.get_shared_secret(),
'secret_thing')
_open.assert_called_with(quantum_contexts.SHARED_SECRET, '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(quantum_contexts.get_shared_secret(),
'secret_thing')
_open.assert_called_with(quantum_contexts.SHARED_SECRET, 'r')
class TestHostIP(CharmTestCase):
def setUp(self):
super(TestHostIP, self).setUp(quantum_contexts,
TO_PATCH)
def test_get_host_ip_already_ip(self):
self.assertEquals(quantum_contexts.get_host_ip('10.5.0.1'),
'10.5.0.1')
def test_get_host_ip_noarg(self):
self.unit_get.return_value = "10.5.0.1"
self.assertEquals(quantum_contexts.get_host_ip(),
'10.5.0.1')
@patch('dns.resolver.query')
def test_get_host_ip_hostname_unresolvable(self, _query):
class NXDOMAIN(Exception):
pass
_query.side_effect = NXDOMAIN()
self.assertRaises(NXDOMAIN, quantum_contexts.get_host_ip,
'missing.example.com')
@patch('dns.resolver.query')
def test_get_host_ip_hostname_resolvable(self, _query):
data = MagicMock()
data.address = '10.5.0.1'
_query.return_value = [data]
self.assertEquals(quantum_contexts.get_host_ip('myhost.example.com'),
'10.5.0.1')
_query.assert_called_with('myhost.example.com', 'A')

143
unit_tests/test_quantum_hooks.py Normal file
View File

@ -0,0 +1,143 @@
from mock import MagicMock, patch, call
import quantum_utils as utils
_register_configs = utils.register_configs
_restart_map = utils.restart_map
utils.register_configs = MagicMock()
utils.restart_map = MagicMock()
import quantum_hooks as hooks
utils.register_configs = _register_configs
utils.restart_map = _restart_map
from charmhelpers.contrib.hahelpers.cluster import HAIncompleteConfig
from test_utils import CharmTestCase
TO_PATCH = [
'config',
'configure_installation_source',
'valid_plugin',
'apt_update',
'apt_install',
'filter_installed_packages',
'get_early_packages',
'get_packages',
'log',
'do_openstack_upgrade',
'openstack_upgrade_available',
'CONFIGS',
'configure_ovs',
'relation_set',
'unit_get',
'relation_get',
'install_ca_cert',
'eligible_leader',
'reassign_agent_resources',
]
class TestQuantumHooks(CharmTestCase):
def setUp(self):
super(TestQuantumHooks, self).setUp(hooks, TO_PATCH)
self.config.side_effect = self.test_config.get
self.test_config.set('openstack-origin', 'cloud:precise-havana')
self.test_config.set('plugin', 'ovs')
def _call_hook(self, hookname):
hooks.hooks.execute([
'hooks/{}'.format(hookname)])
def test_install_hook(self):
self.valid_plugin.return_value = True
_pkgs = ['foo', 'bar']
self.filter_installed_packages.return_value = _pkgs
self._call_hook('install')
self.configure_installation_source.assert_called_with(
'cloud:precise-havana'
)
self.apt_update.assert_called_with(fatal=True)
self.apt_install.assert_has_calls([
call(_pkgs, fatal=True),
call(_pkgs, fatal=True),
])
self.get_early_packages.assert_called()
self.get_packages.assert_called()
@patch('sys.exit')
def test_install_hook_invalid_plugin(self, _exit):
self.valid_plugin.return_value = False
self._call_hook('install')
self.log.assert_called()
_exit.assert_called_with(1)
def test_config_changed_upgrade(self):
self.openstack_upgrade_available.return_value = True
self.valid_plugin.return_value = True
self._call_hook('config-changed')
self.do_openstack_upgrade.assert_called_with(self.CONFIGS)
self.CONFIGS.write_all.assert_called()
self.configure_ovs.assert_called()
@patch('sys.exit')
def test_config_changed_invalid_plugin(self, _exit):
self.valid_plugin.return_value = False
self._call_hook('config-changed')
self.log.assert_called()
_exit.assert_called_with(1)
def test_upgrade_charm(self):
_install = self.patch('install')
_config_changed = self.patch('config_changed')
self._call_hook('upgrade-charm')
_install.assert_called()
_config_changed.assert_called()
def test_db_joined(self):
self.unit_get.return_value = 'myhostname'
self._call_hook('shared-db-relation-joined')
self.relation_set.assert_called_with(
quantum_username='quantum',
quantum_database='quantum',
quantum_hostname='myhostname',
nova_username='nova',
nova_database='nova',
nova_hostname='myhostname',
)
def test_amqp_joined(self):
self._call_hook('amqp-relation-joined')
self.relation_set.assert_called_with(
username='nova',
vhost='nova',
)
def test_amqp_changed(self):
self._call_hook('amqp-relation-changed')
self.CONFIGS.write_all.assert_called()
def test_shared_db_changed(self):
self._call_hook('shared-db-relation-changed')
self.CONFIGS.write_all.assert_called()
def test_nm_changed(self):
self.relation_get.return_value = "cert"
self._call_hook('quantum-network-service-relation-changed')
self.CONFIGS.write_all.assert_called()
self.install_ca_cert.assert_called_with('cert')
def test_cluster_departed_nvp(self):
self.test_config.set('plugin', 'nvp')
self._call_hook('cluster-relation-departed')
self.log.assert_called()
self.eligible_leader.assert_not_called()
self.reassign_agent_resources.assert_not_called()
def test_cluster_departed_ovs_not_leader(self):
self.eligible_leader.return_value = False
self._call_hook('cluster-relation-departed')
self.reassign_agent_resources.assert_not_called()
def test_cluster_departed_ovs_leader(self):
self.eligible_leader.return_value = True
self._call_hook('cluster-relation-departed')
self.reassign_agent_resources.assert_called()

169
unit_tests/test_quantum_utils.py Normal file
View File

@ -0,0 +1,169 @@
from mock import MagicMock, patch, call
import charmhelpers.contrib.openstack.templating as templating
templating.OSConfigRenderer = MagicMock()
import quantum_utils
from collections import OrderedDict
from test_utils import (
CharmTestCase
)
TO_PATCH = [
'get_os_codename_package',
'config',
'get_os_codename_install_source',
'apt_update',
'apt_install',
'configure_installation_source',
'log',
'add_bridge',
'add_bridge_port'
]
class TestQuantumUtils(CharmTestCase):
def setUp(self):
super(TestQuantumUtils, self).setUp(quantum_utils, TO_PATCH)
def test_valid_plugin(self):
self.config.return_value = 'ovs'
self.assertTrue(quantum_utils.valid_plugin())
self.config.return_value = 'nvp'
self.assertTrue(quantum_utils.valid_plugin())
def test_invalid_plugin(self):
self.config.return_value = 'invalid'
self.assertFalse(quantum_utils.valid_plugin())
def test_get_early_packages_ovs(self):
self.config.return_value = 'ovs'
self.assertEquals(quantum_utils.get_early_packages(),
['openvswitch-datapath-dkms'])
def test_get_early_packages_nvp(self):
self.config.return_value = 'nvp'
self.assertEquals(quantum_utils.get_early_packages(),
[])
def test_get_packages_ovs(self):
self.config.return_value = 'ovs'
self.assertNotEqual(quantum_utils.get_packages(), [])
def test_configure_ovs_ovs_ext_port(self):
self.config.side_effect = self.test_config.get
self.test_config.set('plugin', 'ovs')
self.test_config.set('ext-port', 'eth0')
quantum_utils.configure_ovs()
self.add_bridge.assert_has_calls([
call('br-int'),
call('br-ex')
])
self.add_bridge_port.assert_called_with('br-ex', 'eth0')
def test_configure_ovs_nvp(self):
self.config.return_value = 'nvp'
quantum_utils.configure_ovs()
self.add_bridge.assert_called_with('br-int')
def test_do_openstack_upgrade(self):
self.config.side_effect = self.test_config.get
self.test_config.set('openstack-origin', 'cloud:precise-havana')
self.test_config.set('plugin', 'ovs')
self.config.return_value = 'cloud:precise-havana'
self.get_os_codename_install_source.return_value = 'havana'
configs = MagicMock()
quantum_utils.do_openstack_upgrade(configs)
configs.set_release.assert_called_with(openstack_release='havana')
self.log.assert_called()
self.apt_update.assert_called_with(fatal=True)
dpkg_opts = [
'--option', 'Dpkg::Options::=--force-confnew',
'--option', 'Dpkg::Options::=--force-confdef',
]
self.apt_install.assert_called_with(
packages=quantum_utils.GATEWAY_PKGS['ovs'],
options=dpkg_opts, fatal=True
)
self.configure_installation_source.assert_called_with(
'cloud:precise-havana'
)
def test_register_configs_ovs(self):
self.config.return_value = 'ovs'
self.get_os_codename_package.return_value = 'havana'
configs = quantum_utils.register_configs()
confs = [quantum_utils.DHCP_AGENT_CONF,
quantum_utils.METADATA_AGENT_CONF,
quantum_utils.NOVA_CONF,
quantum_utils.QUANTUM_CONF,
quantum_utils.L3_AGENT_CONF,
quantum_utils.OVS_PLUGIN_CONF,
quantum_utils.EXT_PORT_CONF]
for conf in confs:
configs.register.assert_any_call(
conf,
quantum_utils.CONFIG_FILES[quantum_utils.OVS][conf]
['hook_contexts']
)
def test_restart_map_ovs(self):
self.config.return_value = 'ovs'
ex_map = OrderedDict([
(quantum_utils.L3_AGENT_CONF, ['quantum-l3-agent']),
(quantum_utils.OVS_PLUGIN_CONF,
['quantum-plugin-openvswitch-agent']),
(quantum_utils.NOVA_CONF, ['nova-api-metadata']),
(quantum_utils.METADATA_AGENT_CONF, ['quantum-metadata-agent']),
(quantum_utils.DHCP_AGENT_CONF, ['quantum-dhcp-agent']),
(quantum_utils.QUANTUM_CONF, ['quantum-l3-agent',
'quantum-dhcp-agent',
'quantum-metadata-agent',
'quantum-plugin-openvswitch-agent'])
])
self.assertEquals(quantum_utils.restart_map(), ex_map)
def test_register_configs_nvp(self):
self.config.return_value = 'nvp'
self.get_os_codename_package.return_value = 'havana'
configs = quantum_utils.register_configs()
confs = [quantum_utils.DHCP_AGENT_CONF,
quantum_utils.METADATA_AGENT_CONF,
quantum_utils.NOVA_CONF,
quantum_utils.QUANTUM_CONF]
for conf in confs:
configs.register.assert_any_call(
conf,
quantum_utils.CONFIG_FILES[quantum_utils.NVP][conf]
['hook_contexts']
)
def test_restart_map_nvp(self):
self.config.return_value = 'nvp'
ex_map = OrderedDict([
(quantum_utils.DHCP_AGENT_CONF, ['quantum-dhcp-agent']),
(quantum_utils.NOVA_CONF, ['nova-api-metadata']),
(quantum_utils.QUANTUM_CONF, ['quantum-dhcp-agent',
'quantum-metadata-agent']),
(quantum_utils.METADATA_AGENT_CONF, ['quantum-metadata-agent']),
])
self.assertEquals(quantum_utils.restart_map(), ex_map)
def test_register_configs_pre_install(self):
self.config.return_value = 'ovs'
self.get_os_codename_package.return_value = None
configs = quantum_utils.register_configs()
confs = [quantum_utils.DHCP_AGENT_CONF,
quantum_utils.METADATA_AGENT_CONF,
quantum_utils.NOVA_CONF,
quantum_utils.QUANTUM_CONF,
quantum_utils.L3_AGENT_CONF,
quantum_utils.OVS_PLUGIN_CONF,
quantum_utils.EXT_PORT_CONF]
for conf in confs:
configs.register.assert_any_call(
conf,
quantum_utils.CONFIG_FILES[quantum_utils.OVS][conf]
['hook_contexts']
)

97
unit_tests/test_utils.py Normal file
View File

@ -0,0 +1,97 @@
import logging
import unittest
import os
import yaml
from mock import patch
def load_config():
'''
Walk backwords from __file__ looking for config.yaml, load and return the
'options' section'
'''
config = None
f = __file__
while config is None:
d = os.path.dirname(f)
if os.path.isfile(os.path.join(d, 'config.yaml')):
config = os.path.join(d, 'config.yaml')
break
f = d
if not config:
logging.error('Could not find config.yaml in any parent directory '
'of %s. ' % file)
raise Exception
return yaml.safe_load(open(config).read())['options']
def get_default_config():
'''
Load default charm config from config.yaml return as a dict.
If no default is set in config.yaml, its value is None.
'''
default_config = {}
config = load_config()
for k, v in config.iteritems():
if 'default' in v:
default_config[k] = v['default']
else:
default_config[k] = None
return default_config
class CharmTestCase(unittest.TestCase):
def setUp(self, obj, patches):
super(CharmTestCase, self).setUp()
self.patches = patches
self.obj = obj
self.test_config = TestConfig()
self.test_relation = TestRelation()
self.patch_all()
def patch(self, method):
_m = patch.object(self.obj, method)
mock = _m.start()
self.addCleanup(_m.stop)
return mock
def patch_all(self):
for method in self.patches:
setattr(self, method, self.patch(method))
class TestConfig(object):
def __init__(self):
self.config = get_default_config()
def get(self, attr):
try:
return self.config[attr]
except KeyError:
return None
def get_all(self):
return self.config
def set(self, attr, value):
if attr not in self.config:
raise KeyError
self.config[attr] = value
class TestRelation(object):
def __init__(self, relation_data={}):
self.relation_data = relation_data
def set(self, relation_data):
self.relation_data = relation_data
def get(self, attr=None, unit=None, rid=None):
if attr is None:
return self.relation_data
elif attr in self.relation_data:
return self.relation_data[attr]
return None