Merge lp:~gnuoy/charms/trusty/ceph-radosgw/unit-tests into lp:~openstack-charmers-archive/charms/trusty/ceph-radosgw/next

[Liam Young]

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #1661 ceph-radosgw-next for gnuoy mp249167
    UNIT FAIL: unit-test failed

UNIT Results (max last 2 lines):
  FAILED (errors=1)
  make: *** [unit_test] Error 1

Full unit test output: http://paste.ubuntu.com/10156943/
Build: http://10.245.162.77:8080/job/charm_unit_test/1661/

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #1833 ceph-radosgw-next for gnuoy mp249167
    LINT OK: passed

Build: http://10.245.162.77:8080/job/charm_lint_check/1833/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #1852 ceph-radosgw-next for gnuoy mp249167
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
  ERROR subprocess encountered error code 124
  make: *** [test] Error 124

Full amulet test output: http://paste.ubuntu.com/10157281/
Build: http://10.245.162.77:8080/job/charm_amulet_test/1852/

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #1664 ceph-radosgw-next for gnuoy mp249167
    UNIT FAIL: unit-test failed

UNIT Results (max last 2 lines):
  FAILED (errors=1)
  make: *** [unit_test] Error 1

Full unit test output: http://paste.ubuntu.com/10158241/
Build: http://10.245.162.77:8080/job/charm_unit_test/1664/

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #1666 ceph-radosgw-next for gnuoy mp249167
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/1666/

[lp:~gnuoy/charms/trusty/ceph-radosgw/unit-tests]

37. By Liam Young on 2015-02-12

Hack to patch out implicit install of python-dns and subsequent import of dns.python that is triggered by importing utils

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #1947 ceph-radosgw-next for gnuoy mp249167
    LINT OK: passed

Build: http://10.245.162.77:8080/job/charm_lint_check/1947/

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #1737 ceph-radosgw-next for gnuoy mp249167
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/1737/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #1888 ceph-radosgw-next for gnuoy mp249167
    AMULET OK: passed

Build: http://10.245.162.77:8080/job/charm_amulet_test/1888/

=== modified file '.bzrignore'
--- .bzrignore	2014-09-17 14:11:53 +0000
+++ .bzrignore	2015-02-12 09:06:09 +0000
@@ -1,3 +1,5 @@
 .project
 .pydevproject
 bin
+.coveragerc
+.coverage

=== added file '.coveragerc'
--- .coveragerc	1970-01-01 00:00:00 +0000
+++ .coveragerc	2015-02-12 09:06:09 +0000
@@ -0,0 +1,7 @@
+[report]
+# Regexes for lines to exclude from consideration
+exclude_lines =
+    if __name__ == .__main__.:
+include=
+    hooks/ceph.py
+    hooks/hooks.py

=== modified file 'Makefile'
--- Makefile	2014-09-29 01:57:43 +0000
+++ Makefile	2015-02-12 09:06:09 +0000
@@ -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

=== modified file 'hooks/utils.py'
--- hooks/utils.py	2015-01-15 16:18:33 +0000
+++ hooks/utils.py	2015-02-12 09:06:09 +0000
@@ -83,8 +83,10 @@
                 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

=== added directory 'unit_tests'
=== added file 'unit_tests/__init__.py'
--- unit_tests/__init__.py	1970-01-01 00:00:00 +0000
+++ unit_tests/__init__.py	2015-02-12 09:06:09 +0000
@@ -0,0 +1,3 @@
+import sys
+
+sys.path.append('hooks/')

=== added file 'unit_tests/test_ceph.py'
--- unit_tests/test_ceph.py	1970-01-01 00:00:00 +0000
+++ unit_tests/test_ceph.py	2015-02-12 09:06:09 +0000
@@ -0,0 +1,197 @@
+from test_utils import CharmTestCase
+
+import ceph
+
+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)

=== added file 'unit_tests/test_hooks.py'
--- unit_tests/test_hooks.py	1970-01-01 00:00:00 +0000
+++ unit_tests/test_hooks.py	2015-02-12 09:06:09 +0000
@@ -0,0 +1,393 @@
+
+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'):
+    import utils
+
+_reg = utils.register_configs
+
+utils.register_configs = MagicMock()
+
+import hooks as ceph_hooks
+
+utils.register_configs = _reg
+
+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')

=== added file 'unit_tests/test_utils.py'
--- unit_tests/test_utils.py	1970-01-01 00:00:00 +0000
+++ unit_tests/test_utils.py	2015-02-12 09:06:09 +0000
@@ -0,0 +1,119 @@
+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