openstack/charm-ceph-radosgw
Code Issues Proposed changes

[gnuoy,r=hopem]

Browse Source 
Adds unit tests.
This commit is contained in:
Edward Hope-Morley 2015-03-30 18:04:31 +01:00
commit 1e0e459e8a
8 changed files with 732 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.bzrignore
View File

@ -1,3 +1,5 @@
.project
.pydevproject
bin
.coveragerc
.coverage

7
.coveragerc Normal file
View File

@ -0,0 +1,7 @@
[report]
# Regexes for lines to exclude from consideration
exclude_lines =
if __name__ == .__main__.:
include=
hooks/ceph.py
hooks/hooks.py

5
Makefile
View File

@ -2,9 +2,12 @@
PYTHON := /usr/bin/env python
lint:
@flake8 --exclude hooks/charmhelpers hooks tests
@flake8 --exclude hooks/charmhelpers hooks tests unit_tests
@charm proof
unit_test:
@$(PYTHON) /usr/bin/nosetests --nologcapture --with-coverage unit_tests
test:
@echo Starting Amulet tests...
# coreycb note: The -v should only be temporary until Amulet sends

4
hooks/utils.py
View File

@ -83,8 +83,10 @@ def enable_pocket(pocket):
sources.write(line)
def get_host_ip(hostname=unit_get('private-address')):
def get_host_ip(hostname=None):
try:
if not hostname:
hostname = unit_get('private-address')
# Test to see if already an IPv4 address
socket.inet_aton(hostname)
return hostname

3
unit_tests/__init__.py Normal file
View File

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

197
unit_tests/test_ceph.py Normal file
View File

@ -0,0 +1,197 @@
import ceph
from test_utils import CharmTestCase
TO_PATCH = [
'get_unit_hostname',
'os',
'subprocess',
'time',
]
class CephRadosGWCephTests(CharmTestCase):
def setUp(self):
super(CephRadosGWCephTests, self).setUp(ceph, TO_PATCH)
def test_is_quorum_leader(self):
self.os.path.exists.return_value = True
self.get_unit_hostname.return_value = 'myhost'
self.subprocess.check_output.return_value = '{"state": "leader"}'
self.assertEqual(ceph.is_quorum(), True)
def test_is_quorum_notleader(self):
self.os.path.exists.return_value = True
self.get_unit_hostname.return_value = 'myhost'
self.subprocess.check_output.return_value = '{"state": "notleader"}'
self.assertEqual(ceph.is_quorum(), False)
def test_is_quorum_valerror(self):
self.os.path.exists.return_value = True
self.get_unit_hostname.return_value = 'myhost'
self.subprocess.check_output.return_value = "'state': 'bob'}"
self.assertEqual(ceph.is_quorum(), False)
def test_is_quorum_no_asok(self):
self.os.path.exists.return_value = False
self.assertEqual(ceph.is_quorum(), False)
def test_is_leader(self):
self.get_unit_hostname.return_value = 'myhost'
self.os.path.exists.return_value = True
self.subprocess.check_output.return_value = '{"state": "leader"}'
self.assertEqual(ceph.is_leader(), True)
def test_is_leader_notleader(self):
self.get_unit_hostname.return_value = 'myhost'
self.os.path.exists.return_value = True
self.subprocess.check_output.return_value = '{"state": "notleader"}'
self.assertEqual(ceph.is_leader(), False)
def test_is_leader_valerror(self):
self.get_unit_hostname.return_value = 'myhost'
self.os.path.exists.return_value = True
self.subprocess.check_output.return_value = "'state': 'bob'}"
self.assertEqual(ceph.is_leader(), False)
def test_is_leader_noasok(self):
self.get_unit_hostname.return_value = 'myhost'
self.os.path.exists.return_value = False
self.assertEqual(ceph.is_leader(), False)
def test_wait_for_quorum_yes(self):
results = [True, False]
def quorum():
return results.pop()
_is_quorum = self.patch('is_quorum')
_is_quorum.side_effect = quorum
ceph.wait_for_quorum()
self.time.sleep.assert_called_with(3)
def test_wait_for_quorum_no(self):
_is_quorum = self.patch('is_quorum')
_is_quorum.return_value = True
ceph.wait_for_quorum()
self.assertFalse(self.time.sleep.called)
def test_wait_for_bootstrap(self):
results = [True, False]
def bootstrapped():
return results.pop()
_is_bootstrapped = self.patch('is_bootstrapped')
_is_bootstrapped.side_effect = bootstrapped
ceph.wait_for_bootstrap()
self.time.sleep.assert_called_with(3)
def test_add_bootstrap_hint(self):
self.get_unit_hostname.return_value = 'myhost'
cmd = [
"ceph",
"--admin-daemon",
'/var/run/ceph/ceph-mon.myhost.asok',
"add_bootstrap_peer_hint",
'mypeer'
]
self.os.path.exists.return_value = True
ceph.add_bootstrap_hint('mypeer')
self.subprocess.call.assert_called_with(cmd)
def test_add_bootstrap_hint_noasok(self):
self.get_unit_hostname.return_value = 'myhost'
self.os.path.exists.return_value = False
ceph.add_bootstrap_hint('mypeer')
self.assertFalse(self.subprocess.call.called)
def test_is_osd_disk(self):
# XXX Insert real sgdisk output
self.subprocess.check_output.return_value = \
'Partition GUID code: 4FBD7E29-9D25-41B8-AFD0-062C0CEFF05D'
self.assertEqual(ceph.is_osd_disk('/dev/fmd0'), True)
def test_is_osd_disk_no(self):
# XXX Insert real sgdisk output
self.subprocess.check_output.return_value = \
'Partition GUID code: 5FBD7E29-9D25-41B8-AFD0-062C0CEFF05D'
self.assertEqual(ceph.is_osd_disk('/dev/fmd0'), False)
def test_rescan_osd_devices(self):
cmd = [
'udevadm', 'trigger',
'--subsystem-match=block', '--action=add'
]
ceph.rescan_osd_devices()
self.subprocess.call.assert_called_with(cmd)
def test_zap_disk(self):
cmd = [
'sgdisk', '--zap-all', '/dev/fmd0',
]
ceph.zap_disk('/dev/fmd0')
self.subprocess.check_call.assert_called_with(cmd)
def test_import_osd_bootstrap_key(self):
self.os.path.exists.return_value = False
cmd = [
'ceph-authtool',
'/var/lib/ceph/bootstrap-osd/ceph.keyring',
'--create-keyring',
'--name=client.bootstrap-osd',
'--add-key=mykey',
]
ceph.import_osd_bootstrap_key('mykey')
self.subprocess.check_call.assert_called_with(cmd)
def test_is_bootstrapped(self):
self.os.path.exists.return_value = True
self.assertEqual(ceph.is_bootstrapped(), True)
self.os.path.exists.return_value = False
self.assertEqual(ceph.is_bootstrapped(), False)
def test_import_radosgw_key(self):
self.os.path.exists.return_value = False
ceph.import_radosgw_key('mykey')
cmd = [
'ceph-authtool',
'/etc/ceph/keyring.rados.gateway',
'--create-keyring',
'--name=client.radosgw.gateway',
'--add-key=mykey'
]
self.subprocess.check_call.assert_called_with(cmd)
def test_get_named_key_create(self):
self.get_unit_hostname.return_value = "myhost"
self.subprocess.check_output.return_value = """
[client.dummy]
key = AQAPiu1RCMb4CxAAmP7rrufwZPRqy8bpQa2OeQ==
"""
self.assertEqual(ceph.get_named_key('dummy'),
'AQAPiu1RCMb4CxAAmP7rrufwZPRqy8bpQa2OeQ==')
cmd = [
'ceph',
'--name', 'mon.',
'--keyring',
'/var/lib/ceph/mon/ceph-myhost/keyring',
'auth', 'get-or-create', 'client.dummy',
'mon', 'allow r', 'osd', 'allow rwx'
]
self.subprocess.check_output.assert_called_with(cmd)
def test_get_named_key_get(self):
self.get_unit_hostname.return_value = "myhost"
key = "AQAPiu1RCMb4CxAAmP7rrufwZPRqy8bpQa2OeQ=="
self.subprocess.check_output.return_value = key
self.assertEqual(ceph.get_named_key('dummy'), key)
cmd = [
'ceph',
'--name', 'mon.',
'--keyring',
'/var/lib/ceph/mon/ceph-myhost/keyring',
'auth', 'get-or-create', 'client.dummy',
'mon', 'allow r', 'osd', 'allow rwx'
]
self.subprocess.check_output.assert_called_with(cmd)

394
unit_tests/test_hooks.py Normal file
View File

@ -0,0 +1,394 @@
from mock import (
call,
patch,
MagicMock
)
from test_utils import (
CharmTestCase,
patch_open
)
dnsmock = MagicMock()
modules = {
'dns': dnsmock,
'dns.resolver': dnsmock,
}
module_patcher = patch.dict('sys.modules', modules)
module_patcher.start()
with patch('charmhelpers.fetch.apt_install'):
with patch('utils.register_configs'):
import hooks as ceph_hooks
TO_PATCH = [
'add_source',
'apt_update',
'apt_install',
'apt_purge',
'config',
'cmp_pkgrevno',
'execd_preinstall',
'enable_pocket',
'get_host_ip',
'get_iface_for_address',
'get_netmask_for_address',
'get_unit_hostname',
'glob',
'is_apache_24',
'log',
'lsb_release',
'open_port',
'os',
'related_units',
'relation_ids',
'relation_set',
'relation_get',
'render_template',
'resolve_address',
'shutil',
'subprocess',
'sys',
'unit_get',
]
class CephRadosGWTests(CharmTestCase):
def setUp(self):
super(CephRadosGWTests, self).setUp(ceph_hooks, TO_PATCH)
self.config.side_effect = self.test_config.get
self.test_config.set('source', 'distro')
self.test_config.set('key', 'secretkey')
self.test_config.set('use-syslog', False)
def test_install_www_scripts(self):
self.glob.glob.return_value = ['files/www/bob']
ceph_hooks.install_www_scripts()
self.shutil.copy.assert_called_with('files/www/bob', '/var/www/')
def test_install_ceph_optimised_packages(self):
self.lsb_release.return_value = {'DISTRIB_CODENAME': 'vivid'}
fastcgi_source = (
'http://gitbuilder.ceph.com/'
'libapache-mod-fastcgi-deb-vivid-x86_64-basic/ref/master')
apache_source = (
'http://gitbuilder.ceph.com/'
'apache2-deb-vivid-x86_64-basic/ref/master')
calls = [
call(fastcgi_source, key='6EAEAE2203C3951A'),
call(apache_source, key='6EAEAE2203C3951A'),
]
ceph_hooks.install_ceph_optimised_packages()
self.add_source.assert_has_calls(calls)
def test_install_packages(self):
self.test_config.set('use-ceph-optimised-packages', '')
ceph_hooks.install_packages()
self.add_source.assert_called_with('distro', 'secretkey')
self.apt_update.assert_called()
self.apt_install.assert_called_with(['libapache2-mod-fastcgi',
'apache2'], fatal=True)
def test_install_optimised_packages_no_embedded(self):
self.test_config.set('use-ceph-optimised-packages', True)
self.test_config.set('use-embedded-webserver', False)
_install_packages = self.patch('install_ceph_optimised_packages')
ceph_hooks.install_packages()
self.add_source.assert_called_with('distro', 'secretkey')
self.apt_update.assert_called()
_install_packages.assert_called()
self.apt_install.assert_called_with(['libapache2-mod-fastcgi',
'apache2'], fatal=True)
def test_install_optimised_packages_embedded(self):
self.test_config.set('use-ceph-optimised-packages', True)
self.test_config.set('use-embedded-webserver', True)
_install_packages = self.patch('install_ceph_optimised_packages')
ceph_hooks.install_packages()
self.add_source.assert_called_with('distro', 'secretkey')
self.apt_update.assert_called()
_install_packages.assert_called()
self.apt_install.assert_called_with(['radosgw',
'ntp',
'haproxy'], fatal=True)
self.apt_purge.assert_called_with(['libapache2-mod-fastcgi',
'apache2'])
def test_install(self):
_install_packages = self.patch('install_packages')
ceph_hooks.install()
self.execd_preinstall.assert_called()
_install_packages.assert_called()
self.enable_pocket.assert_called_with('multiverse')
self.os.makedirs.called_with('/var/lib/ceph/nss')
def test_emit_cephconf(self):
_get_keystone_conf = self.patch('get_keystone_conf')
_get_auth = self.patch('get_auth')
_get_mon_hosts = self.patch('get_mon_hosts')
_get_auth.return_value = 'cephx'
_get_keystone_conf.return_value = {'keystone_key': 'keystone_value'}
_get_mon_hosts.return_value = ['10.0.0.1:6789', '10.0.0.2:6789']
self.get_unit_hostname.return_value = 'bob'
self.os.path.exists.return_value = False
cephcontext = {
'auth_supported': 'cephx',
'mon_hosts': '10.0.0.1:6789 10.0.0.2:6789',
'hostname': 'bob',
'old_auth': False,
'use_syslog': 'false',
'keystone_key': 'keystone_value',
'embedded_webserver': False,
}
self.cmp_pkgrevno.return_value = 1
with patch_open() as (_open, _file):
ceph_hooks.emit_cephconf()
self.os.makedirs.assert_called_with('/etc/ceph')
_open.assert_called_with('/etc/ceph/ceph.conf', 'w')
self.render_template.assert_called_with('ceph.conf', cephcontext)
def test_emit_apacheconf(self):
self.is_apache_24.return_value = True
self.unit_get.return_value = '10.0.0.1'
apachecontext = {
"hostname": '10.0.0.1',
}
vhost_file = '/etc/apache2/sites-available/rgw.conf'
with patch_open() as (_open, _file):
ceph_hooks.emit_apacheconf()
_open.assert_called_with(vhost_file, 'w')
self.render_template.assert_called_with('rgw', apachecontext)
def test_apache_sites24(self):
self.is_apache_24.return_value = True
ceph_hooks.apache_sites()
calls = [
call(['a2dissite', '000-default']),
call(['a2ensite', 'rgw']),
]
self.subprocess.check_call.assert_has_calls(calls)
def test_apache_sites22(self):
self.is_apache_24.return_value = False
ceph_hooks.apache_sites()
calls = [
call(['a2dissite', 'default']),
call(['a2ensite', 'rgw']),
]
self.subprocess.check_call.assert_has_calls(calls)
def test_apache_modules(self):
ceph_hooks.apache_modules()
calls = [
call(['a2enmod', 'fastcgi']),
call(['a2enmod', 'rewrite']),
]
self.subprocess.check_call.assert_has_calls(calls)
def test_apache_reload(self):
ceph_hooks.apache_reload()
calls = [
call(['service', 'apache2', 'reload']),
]
self.subprocess.call.assert_has_calls(calls)
def test_config_changed(self):
_install_packages = self.patch('install_packages')
_emit_cephconf = self.patch('emit_cephconf')
_emit_apacheconf = self.patch('emit_apacheconf')
_install_www_scripts = self.patch('install_www_scripts')
_apache_sites = self.patch('apache_sites')
_apache_modules = self.patch('apache_modules')
_apache_reload = self.patch('apache_reload')
ceph_hooks.config_changed()
_install_packages.assert_called()
_emit_cephconf.assert_called()
_emit_apacheconf.assert_called()
_install_www_scripts.assert_called()
_apache_sites.assert_called()
_apache_modules.assert_called()
_apache_reload.assert_called()
def test_get_mon_hosts(self):
self.relation_ids.return_value = ['monrelid']
self.related_units.return_value = ['monunit']
self.relation_get.return_value = '10.0.0.1'
self.get_host_ip.return_value = '10.0.0.1'
self.assertEquals(ceph_hooks.get_mon_hosts(), ['10.0.0.1:6789'])
def test_get_conf(self):
self.relation_ids.return_value = ['monrelid']
self.related_units.return_value = ['monunit']
self.relation_get.return_value = 'bob'
self.assertEquals(ceph_hooks.get_conf('key'), 'bob')
def test_get_conf_nomatch(self):
self.relation_ids.return_value = ['monrelid']
self.related_units.return_value = ['monunit']
self.relation_get.return_value = ''
self.assertEquals(ceph_hooks.get_conf('key'), None)
def test_get_auth(self):
self.relation_ids.return_value = ['monrelid']
self.related_units.return_value = ['monunit']
self.relation_get.return_value = 'bob'
self.assertEquals(ceph_hooks.get_auth(), 'bob')
def test_get_keystone_conf(self):
self.test_config.set('operator-roles', 'admin')
self.test_config.set('cache-size', '42')
self.test_config.set('revocation-check-interval', '21')
self.relation_ids.return_value = ['idrelid']
self.related_units.return_value = ['idunit']
def _relation_get(key, unit, relid):
ks_dict = {
'auth_protocol': 'https',
'auth_host': '10.0.0.2',
'auth_port': '8090',
'admin_token': 'sectocken',
}
return ks_dict[key]
self.relation_get.side_effect = _relation_get
self.assertEquals(ceph_hooks.get_keystone_conf(), {
'auth_type': 'keystone',
'auth_protocol': 'https',
'admin_token': 'sectocken',
'user_roles': 'admin',
'auth_host': '10.0.0.2',
'cache_size': '42',
'auth_port': '8090',
'revocation_check_interval': '21'})
def test_get_keystone_conf_missinginfo(self):
self.test_config.set('operator-roles', 'admin')
self.test_config.set('cache-size', '42')
self.test_config.set('revocation-check-interval', '21')
self.relation_ids.return_value = ['idrelid']
self.related_units.return_value = ['idunit']
def _relation_get(key, unit, relid):
ks_dict = {
'auth_protocol': 'https',
'auth_host': '10.0.0.2',
'auth_port': '8090',
}
return ks_dict[key] if key in ks_dict else None
self.relation_get.side_effect = _relation_get
self.assertEquals(ceph_hooks.get_keystone_conf(), None)
def test_mon_relation(self):
_emit_cephconf = self.patch('emit_cephconf')
_ceph = self.patch('ceph')
_restart = self.patch('restart')
self.relation_get.return_value = 'seckey'
ceph_hooks.mon_relation()
_restart.assert_called()
_ceph.import_radosgw_key.assert_called_with('seckey')
_emit_cephconf.assert_called()
def test_mon_relation_nokey(self):
_emit_cephconf = self.patch('emit_cephconf')
_ceph = self.patch('ceph')
_restart = self.patch('restart')
self.relation_get.return_value = None
ceph_hooks.mon_relation()
self.assertFalse(_ceph.import_radosgw_key.called)
self.assertFalse(_restart.called)
_emit_cephconf.assert_called()
def test_gateway_relation(self):
self.unit_get.return_value = 'myserver'
ceph_hooks.gateway_relation()
self.relation_set.assert_called_with(hostname='myserver', port=80)
def test_start(self):
ceph_hooks.start()
cmd = ['service', 'radosgw', 'start']
self.subprocess.call.assert_called_with(cmd)
def test_stop(self):
ceph_hooks.stop()
cmd = ['service', 'radosgw', 'stop']
self.subprocess.call.assert_called_with(cmd)
def test_restart(self):
ceph_hooks.restart()
cmd = ['service', 'radosgw', 'restart']
self.subprocess.call.assert_called_with(cmd)
def test_identity_joined_early_version(self):
self.cmp_pkgrevno.return_value = -1
ceph_hooks.identity_joined()
self.sys.exit.assert_called_with(1)
def test_identity_joined(self):
self.cmp_pkgrevno.return_value = 1
self.resolve_address.return_value = 'myserv'
self.test_config.set('region', 'region1')
self.test_config.set('operator-roles', 'admin')
self.unit_get.return_value = 'myserv'
ceph_hooks.identity_joined(relid='rid')
self.relation_set.assert_called_with(
service='swift',
region='region1',
public_url='http://myserv:80/swift/v1',
internal_url='http://myserv:80/swift/v1',
requested_roles='admin',
relation_id='rid',
admin_url='http://myserv:80/swift')
def test_identity_changed(self):
_emit_cephconf = self.patch('emit_cephconf')
_restart = self.patch('restart')
ceph_hooks.identity_changed()
_emit_cephconf.assert_called()
_restart.assert_called()
def test_canonical_url_ipv6(self):
ipv6_addr = '2001:db8:85a3:8d3:1319:8a2e:370:7348'
self.resolve_address.return_value = ipv6_addr
self.assertEquals(ceph_hooks.canonical_url({}),
'http://[%s]' % ipv6_addr)
@patch.object(ceph_hooks, 'CONFIGS')
def test_cluster_changed(self, configs):
_id_joined = self.patch('identity_joined')
self.relation_ids.return_value = ['rid']
ceph_hooks.cluster_changed()
configs.write_all.assert_called()
_id_joined.assert_called_with(relid='rid')
def test_ha_relation_joined_no_vip(self):
self.test_config.set('vip', '')
ceph_hooks.ha_relation_joined()
self.sys.exit.assert_called_with(1)
def test_ha_relation_joined_vip(self):
self.test_config.set('ha-bindiface', 'eth8')
self.test_config.set('ha-mcastport', '5000')
self.test_config.set('vip', '10.0.0.10')
self.get_iface_for_address.return_value = 'eth7'
self.get_netmask_for_address.return_value = '255.255.0.0'
ceph_hooks.ha_relation_joined()
eth_params = ('params ip="10.0.0.10" cidr_netmask="255.255.0.0" '
'nic="eth7"')
resources = {'res_cephrg_haproxy': 'lsb:haproxy',
'res_cephrg_eth7_vip': 'ocf:heartbeat:IPaddr2'}
resource_params = {'res_cephrg_haproxy': 'op monitor interval="5s"',
'res_cephrg_eth7_vip': eth_params}
self.relation_set.assert_called_with(
init_services={'res_cephrg_haproxy': 'haproxy'},
corosync_bindiface='eth8',
corosync_mcastport='5000',
resource_params=resource_params,
resources=resources,
clones={'cl_cephrg_haproxy': 'res_cephrg_haproxy'})
def test_ha_relation_changed(self):
_id_joined = self.patch('identity_joined')
self.relation_get.return_value = True
self.relation_ids.return_value = ['rid']
ceph_hooks.ha_relation_changed()
_id_joined.assert_called_with(relid='rid')

122
unit_tests/test_utils.py Normal file
View File

@ -0,0 +1,122 @@
import logging
import os
import unittest
import yaml
from contextlib import contextmanager
from mock import (
patch,
MagicMock,
)
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=None):
if not attr:
return self.get_all()
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
@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