Add unit tests for ha-relation-joined hook

setup.cfg

@@ -0,0 +1,6 @@
+[nosetests]
+verbosity=2
+with-coverage=1
+cover-erase=1
+cover-package=hooks
+

unit_tests/test_percona_hooks.py

@@ -0,0 +1,65 @@
+import mock
+import sys
+from test_utils import CharmTestCase
+
+sys.modules['MySQLdb'] = mock.Mock()
+import percona_hooks as hooks
+
+TO_PATCH = ['log', 'config',
+            'get_db_helper',
+            'relation_ids',
+            'relation_set']
+
+
+class TestHaRelation(CharmTestCase):
+    def setUp(self):
+        CharmTestCase.setUp(self, hooks, TO_PATCH)
+
+    @mock.patch('sys.exit')
+    def test_relation_not_configured(self, exit_):
+        self.config.return_value = None
+
+        class MyError(Exception):
+            pass
+
+        def f(x):
+            raise MyError(x)
+        exit_.side_effect = f
+        self.assertRaises(MyError, hooks.ha_relation_joined)
+
+    def test_resources(self):
+        self.relation_ids.return_value = ['ha:1']
+        password = 'ubuntu'
+        helper = mock.Mock()
+        attrs = {'get_mysql_password.return_value': password}
+        helper.configure_mock(**attrs)
+        self.get_db_helper.return_value = helper
+        self.test_config.set('vip', '10.0.3.3')
+        self.test_config.set('sst-password', password)
+        def f(k):
+            return self.test_config.get(k)
+
+        self.config.side_effect = f
+        hooks.ha_relation_joined()
+
+        resources = {'res_mysql_vip': 'ocf:heartbeat:IPaddr2',
+                     'res_mysql_monitor': 'ocf:percona:mysql_monitor'}
+        resource_params = {'res_mysql_vip': ('params ip="10.0.3.3" '
+                                             'cidr_netmask="24" '
+                                             'nic="eth0"'),
+                           'res_mysql_monitor':
+                               hooks.RES_MONITOR_PARAMS % {'sstpass': 'ubuntu'}}
+        groups = {'grp_percona_cluster': 'res_mysql_vip'}
+
+        clones = {'cl_mysql_monitor': 'res_mysql_monitor meta interleave=true'}
+
+        colocations = {'vip_mysqld': 'inf: grp_percona_cluster cl_mysql_monitor'}
+
+        locations = {'loc_percona_cluster':
+                     'grp_percona_cluster rule inf: writable eq 1'}
+
+        self.relation_set.assert_called_with(
+            relation_id='ha:1', corosync_bindiface=f('ha-bindiface'),
+            corosync_mcastport=f('ha-mcastport'), resources=resources,
+            resource_params=resource_params, groups=groups,
+            clones=clones, colocations=colocations, locations=locations)

unit_tests/test_utils.py

@@ -0,0 +1,121 @@
+import logging
+import unittest
+import os
+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