Merge lp:~gnuoy/charm-helpers/cepg-broker into lp:charm-helpers

[Liam Young]

[Chris Glass (tribaal)]

Hi, thanks for your proposal.

The code looks fine in general, but it would really help future readers if you could expand the docstrings a little bit to explain what the problem they are trying to solve is. I understand it now, after some effort, while the bug and context is still fresh, but I (and others) certainly won't in a few months if coming back to it is needed.

An extra bit at the beginning of the file or outside of a particular function about how the newly introduced functions are supposed to be called by whom and in what order would also help tremendously.

I know it seems verbose right now, but it will really save a lot of time if in the future somebody has to look at this code again.

Suggestions inline.

review: Needs Information

[Edward Hope-Morley (hopem)]

In general I think this is good but please see comments in https://code.launchpad.net/~gnuoy/charms/trusty/ceph/1453940/+merge/268614

Can you please make log lvels specific. Right now they are all default i.e. INFO but mpost should be DEBUG.

[Edward Hope-Morley (hopem)]

Still reviewing but in general I prefer this approach. 1 inline comment.

[lp:~gnuoy/charm-helpers/cepg-broker]

443. By Liam Young on 2015-09-07

Change how ops injection works in light of review comments

444. By Liam Young on 2015-09-10

Some tidyup

[Chris Glass (tribaal)]

Looks good now after the cleanups and additions, thanks a lot. +1

review: Approve

[Edward Hope-Morley (hopem)]

LGTM +1

review: Approve

=== modified file 'charmhelpers/contrib/openstack/context.py'
--- charmhelpers/contrib/openstack/context.py	2015-09-02 14:09:43 +0000
+++ charmhelpers/contrib/openstack/context.py	2015-09-10 09:26:02 +0000
@@ -485,13 +485,15 @@
 
         log('Generating template context for ceph', level=DEBUG)
         mon_hosts = []
-        auth = None
-        key = None
-        use_syslog = str(config('use-syslog')).lower()
+        ctxt = {
+            'use_syslog': str(config('use-syslog')).lower()
+        }
         for rid in relation_ids('ceph'):
             for unit in related_units(rid):
-                auth = relation_get('auth', rid=rid, unit=unit)
-                key = relation_get('key', rid=rid, unit=unit)
+                if not ctxt.get('auth'):
+                    ctxt['auth'] = relation_get('auth', rid=rid, unit=unit)
+                if not ctxt.get('key'):
+                    ctxt['key'] = relation_get('key', rid=rid, unit=unit)
                 ceph_pub_addr = relation_get('ceph-public-address', rid=rid,
                                              unit=unit)
                 unit_priv_addr = relation_get('private-address', rid=rid,
@@ -500,10 +502,7 @@
                 ceph_addr = format_ipv6_addr(ceph_addr) or ceph_addr
                 mon_hosts.append(ceph_addr)
 
-        ctxt = {'mon_hosts': ' '.join(sorted(mon_hosts)),
-                'auth': auth,
-                'key': key,
-                'use_syslog': use_syslog}
+        ctxt['mon_hosts'] = ' '.join(sorted(mon_hosts))
 
         if not os.path.isdir('/etc/ceph'):
             os.mkdir('/etc/ceph')

=== modified file 'charmhelpers/contrib/storage/linux/ceph.py'
--- charmhelpers/contrib/storage/linux/ceph.py	2015-09-03 14:22:08 +0000
+++ charmhelpers/contrib/storage/linux/ceph.py	2015-09-10 09:26:02 +0000
@@ -28,6 +28,7 @@
 import shutil
 import json
 import time
+import uuid
 
 from subprocess import (
     check_call,
@@ -35,8 +36,10 @@
     CalledProcessError,
 )
 from charmhelpers.core.hookenv import (
+    local_unit,
     relation_get,
     relation_ids,
+    relation_set,
     related_units,
     log,
     DEBUG,
@@ -402,17 +405,52 @@
 
     The API is versioned and defaults to version 1.
     """
-    def __init__(self, api_version=1):
+    def __init__(self, api_version=1, request_id=None):
         self.api_version = api_version
+        if request_id:
+            self.request_id = request_id
+        else:
+            self.request_id = str(uuid.uuid1())
         self.ops = []
 
     def add_op_create_pool(self, name, replica_count=3):
         self.ops.append({'op': 'create-pool', 'name': name,
                          'replicas': replica_count})
 
+    def set_ops(self, ops):
+        """Set request ops to provided value.
+
+        Useful for injecting ops that come from a previous request
+        to allow comparisons to ensure validity.
+        """
+        self.ops = ops
+
     @property
     def request(self):
-        return json.dumps({'api-version': self.api_version, 'ops': self.ops})
+        return json.dumps({'api-version': self.api_version, 'ops': self.ops,
+                           'request-id': self.request_id})
+
+    def _ops_equal(self, other):
+        if len(self.ops) == len(other.ops):
+            for req_no in range(0, len(self.ops)):
+                for key in ['replicas', 'name', 'op']:
+                    if self.ops[req_no][key] != other.ops[req_no][key]:
+                        return False
+        else:
+            return False
+        return True
+
+    def __eq__(self, other):
+        if not isinstance(other, self.__class__):
+            return False
+        if self.api_version == other.api_version and \
+                self._ops_equal(other):
+            return True
+        else:
+            return False
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
 
 
 class CephBrokerRsp(object):
@@ -422,14 +460,198 @@
 
     The API is versioned and defaults to version 1.
     """
+
     def __init__(self, encoded_rsp):
         self.api_version = None
         self.rsp = json.loads(encoded_rsp)
 
     @property
+    def request_id(self):
+        return self.rsp.get('request-id')
+
+    @property
     def exit_code(self):
         return self.rsp.get('exit-code')
 
     @property
     def exit_msg(self):
         return self.rsp.get('stderr')
+
+
+# Ceph Broker Conversation:
+# If a charm needs an action to be taken by ceph it can create a CephBrokerRq
+# and send that request to ceph via the ceph relation. The CephBrokerRq has a
+# unique id so that the client can identity which CephBrokerRsp is associated
+# with the request. Ceph will also respond to each client unit individually
+# creating a response key per client unit eg glance/0 will get a CephBrokerRsp
+# via key broker-rsp-glance-0
+#
+# To use this the charm can just do something like:
+#
+# from charmhelpers.contrib.storage.linux.ceph import (
+#     send_request_if_needed,
+#     is_request_complete,
+#     CephBrokerRq,
+# )
+#
+# @hooks.hook('ceph-relation-changed')
+# def ceph_changed():
+#     rq = CephBrokerRq()
+#     rq.add_op_create_pool(name='poolname', replica_count=3)
+#
+#     if is_request_complete(rq):
+#         <Request complete actions>
+#     else:
+#         send_request_if_needed(get_ceph_request())
+#
+# CephBrokerRq and CephBrokerRsp are serialized into JSON. Below is an example
+# of glance having sent a request to ceph which ceph has successfully processed
+#  'ceph:8': {
+#      'ceph/0': {
+#          'auth': 'cephx',
+#          'broker-rsp-glance-0': '{"request-id": "0bc7dc54", "exit-code": 0}',
+#          'broker_rsp': '{"request-id": "0da543b8", "exit-code": 0}',
+#          'ceph-public-address': '10.5.44.103',
+#          'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==',
+#          'private-address': '10.5.44.103',
+#      },
+#      'glance/0': {
+#          'broker_req': ('{"api-version": 1, "request-id": "0bc7dc54", '
+#                         '"ops": [{"replicas": 3, "name": "glance", '
+#                         '"op": "create-pool"}]}'),
+#          'private-address': '10.5.44.109',
+#      },
+#  }
+
+def get_previous_request(rid):
+    """Return the last ceph broker request sent on a given relation
+
+    @param rid: Relation id to query for request
+    """
+    request = None
+    broker_req = relation_get(attribute='broker_req', rid=rid,
+                              unit=local_unit())
+    if broker_req:
+        request_data = json.loads(broker_req)
+        request = CephBrokerRq(api_version=request_data['api-version'],
+                               request_id=request_data['request-id'])
+        request.set_ops(request_data['ops'])
+
+    return request
+
+
+def get_request_states(request):
+    """Return a dict of requests per relation id with their corresponding
+       completion state.
+
+    This allows a charm, which has a request for ceph, to see whether there is
+    an equivalent request already being processed and if so what state that
+    request is in.
+
+    @param request: A CephBrokerRq object
+    """
+    complete = []
+    requests = {}
+    for rid in relation_ids('ceph'):
+        complete = False
+        previous_request = get_previous_request(rid)
+        if request == previous_request:
+            sent = True
+            complete = is_request_complete_for_rid(previous_request, rid)
+        else:
+            sent = False
+            complete = False
+
+        requests[rid] = {
+            'sent': sent,
+            'complete': complete,
+        }
+
+    return requests
+
+
+def is_request_sent(request):
+    """Check to see if a functionally equivalent request has already been sent
+
+    Returns True if a similair request has been sent
+
+    @param request: A CephBrokerRq object
+    """
+    states = get_request_states(request)
+    for rid in states.keys():
+        if not states[rid]['sent']:
+            return False
+
+    return True
+
+
+def is_request_complete(request):
+    """Check to see if a functionally equivalent request has already been
+    completed
+
+    Returns True if a similair request has been completed
+
+    @param request: A CephBrokerRq object
+    """
+    states = get_request_states(request)
+    for rid in states.keys():
+        if not states[rid]['complete']:
+            return False
+
+    return True
+
+
+def is_request_complete_for_rid(request, rid):
+    """Check if a given request has been completed on the given relation
+
+    @param request: A CephBrokerRq object
+    @param rid: Relation ID
+    """
+    broker_key = get_broker_rsp_key()
+    for unit in related_units(rid):
+        rdata = relation_get(rid=rid, unit=unit)
+        if rdata.get(broker_key):
+            rsp = CephBrokerRsp(rdata.get(broker_key))
+            if rsp.request_id == request.request_id:
+                if not rsp.exit_code:
+                    return True
+        else:
+            # The remote unit sent no reply targeted at this unit so either the
+            # remote ceph cluster does not support unit targeted replies or it
+            # has not processed our request yet.
+            if rdata.get('broker_rsp'):
+                request_data = json.loads(rdata['broker_rsp'])
+                if request_data.get('request-id'):
+                    log('Ignoring legacy broker_rsp without unit key as remote '
+                        'service supports unit specific replies', level=DEBUG)
+                else:
+                    log('Using legacy broker_rsp as remote service does not '
+                        'supports unit specific replies', level=DEBUG)
+                    rsp = CephBrokerRsp(rdata['broker_rsp'])
+                    if not rsp.exit_code:
+                        return True
+
+    return False
+
+
+def get_broker_rsp_key():
+    """Return broker response key for this unit
+
+    This is the key that ceph is going to use to pass request status
+    information back to this unit
+    """
+    return 'broker-rsp-' + local_unit().replace('/', '-')
+
+
+def send_request_if_needed(request):
+    """Send broker request if an equivalent request has not already been sent
+
+    @param request: A CephBrokerRq object
+    """
+    if is_request_sent(request):
+        log('Request already sent but not complete, not sending new request',
+            level=DEBUG)
+    else:
+        for rid in relation_ids('ceph'):
+            log('Sending request {}'.format(request.request_id), level=DEBUG)
+            relation_set(relation_id=rid, broker_req=request.request)

=== modified file 'tests/contrib/openstack/test_os_contexts.py'
--- tests/contrib/openstack/test_os_contexts.py	2015-09-02 14:09:43 +0000
+++ tests/contrib/openstack/test_os_contexts.py	2015-09-10 09:26:02 +0000
@@ -74,7 +74,7 @@
 
     def relation_ids(self, relation):
         rids = []
-        for rid in self.relation_data.keys():
+        for rid in sorted(self.relation_data.keys()):
             if relation + ':' in rid:
                 rids.append(rid)
         return rids
@@ -82,7 +82,7 @@
     def relation_units(self, relation_id):
         if relation_id not in self.relation_data:
             return None
-        return self.relation_data[relation_id].keys()
+        return sorted(self.relation_data[relation_id].keys())
 
 SHARED_DB_RELATION = {
     'db_host': 'dbserver.local',
@@ -1029,7 +1029,7 @@
     def test_ceph_context_with_data(self, ensure_packages, mkdir, isdir,
                                     config):
         '''Test ceph context with all relation data'''
-        config.return_value = True
+        config.side_effect = fake_config({'use-syslog': 'True'})
         isdir.return_value = False
         relation = FakeRelation(relation_data=CEPH_RELATION)
         self.relation_get.side_effect = relation.get
@@ -1051,7 +1051,7 @@
     @patch.object(context, 'ensure_packages')
     def test_ceph_context_with_missing_data(self, ensure_packages, mkdir):
         '''Test ceph context with missing relation data'''
-        relation = copy(CEPH_RELATION)
+        relation = deepcopy(CEPH_RELATION)
         for k, v in six.iteritems(relation):
             for u in six.iterkeys(v):
                 del relation[k][u]['auth']
@@ -1068,6 +1068,38 @@
     @patch('os.path.isdir')
     @patch('os.mkdir')
     @patch.object(context, 'ensure_packages')
+    def test_ceph_context_partial_missing_data(self, ensure_packages, mkdir,
+                                               isdir, config):
+        '''Test ceph context last unit missing data
+
+           Tests a fix to a previously bug which meant only the config from
+           last unit was returned so if a valid value was supplied from an
+           earlier unit it would be ignored'''
+        config.side_effect = fake_config({'use-syslog': 'True'})
+        relation = deepcopy(CEPH_RELATION)
+        for k, v in six.iteritems(relation):
+            last_unit = sorted(six.iterkeys(v))[-1]
+            unit_data = relation[k][last_unit]
+            del unit_data['auth']
+            relation[k][last_unit] = unit_data
+        relation = FakeRelation(relation_data=relation)
+        self.relation_get.side_effect = relation.get
+        self.relation_ids.side_effect = relation.relation_ids
+        self.related_units.side_effect = relation.relation_units
+        ceph = context.CephContext()
+        result = ceph()
+        expected = {
+            'mon_hosts': 'ceph_node1 ceph_node2',
+            'auth': 'foo',
+            'key': 'bar',
+            'use_syslog': 'true'
+        }
+        self.assertEquals(result, expected)
+
+    @patch.object(context, 'config')
+    @patch('os.path.isdir')
+    @patch('os.mkdir')
+    @patch.object(context, 'ensure_packages')
     def test_ceph_context_with_public_addr(
             self, ensure_packages, mkdir, isdir, config):
         '''Test ceph context in host with multiple networks with all

=== modified file 'tests/contrib/storage/test_linux_ceph.py'
--- tests/contrib/storage/test_linux_ceph.py	2015-09-03 14:22:08 +0000
+++ tests/contrib/storage/test_linux_ceph.py	2015-09-10 09:26:02 +0000
@@ -5,10 +5,11 @@
 from threading import Timer
 from testtools import TestCase
 import json
+import copy
 
 import charmhelpers.contrib.storage.linux.ceph as ceph_utils
 from subprocess import CalledProcessError
-from tests.helpers import patch_open
+from tests.helpers import patch_open, FakeRelation
 import nose.plugins.attrib
 import os
 import time
@@ -31,6 +32,46 @@
 baz
 """
 
+CEPH_CLIENT_RELATION = {
+    'ceph:8': {
+        'ceph/0': {
+            'auth': 'cephx',
+            'broker-rsp-glance-0': '{"request-id": "0bc7dc54", "exit-code": 0}',
+            'broker-rsp-glance-1': '{"request-id": "0880e22a", "exit-code": 0}',
+            'broker-rsp-glance-2': '{"request-id": "0da543b8", "exit-code": 0}',
+            'broker_rsp': '{"request-id": "0da543b8", "exit-code": 0}',
+            'ceph-public-address': '10.5.44.103',
+            'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==',
+            'private-address': '10.5.44.103',
+        },
+        'ceph/1': {
+            'auth': 'cephx',
+            'ceph-public-address': '10.5.44.104',
+            'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==',
+            'private-address': '10.5.44.104',
+        },
+        'ceph/2': {
+            'auth': 'cephx',
+            'ceph-public-address': '10.5.44.105',
+            'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==',
+            'private-address': '10.5.44.105',
+        },
+        'glance/0': {
+            'broker_req': '{"api-version": 1, "request-id": "0bc7dc54", "ops": [{"replicas": 3, "name": "glance", "op": "create-pool"}]}',
+            'private-address': '10.5.44.109',
+        },
+    }
+}
+
+CEPH_CLIENT_RELATION_LEGACY = copy.deepcopy(CEPH_CLIENT_RELATION)
+CEPH_CLIENT_RELATION_LEGACY['ceph:8']['ceph/0'] = {
+    'auth': 'cephx',
+    'broker_rsp': '{"exit-code": 0}',
+    'ceph-public-address': '10.5.44.103',
+    'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==',
+    'private-address': '10.5.44.103',
+}
+
 
 class CephUtilsTests(TestCase):
     def setUp(self):
@@ -38,6 +79,10 @@
         [self._patch(m) for m in [
             'check_call',
             'check_output',
+            'relation_get',
+            'related_units',
+            'relation_ids',
+            'relation_set',
             'log',
         ]]
 
@@ -554,19 +599,303 @@
             b'ceph version 0.67.4 (ad85b8bfafea6232d64cb7ba76a8b6e8252fa0c7)'
         self.assertEquals(ceph_utils.ceph_version(), '0.67.4')
 
-    def test_ceph_broker_rq_class(self):
+    @patch.object(ceph_utils, 'uuid')
+    def test_ceph_broker_rq_class(self, uuid):
+        uuid.uuid1.return_value = 'uuid'
         rq = ceph_utils.CephBrokerRq()
         rq.add_op_create_pool('pool1', replica_count=1)
         rq.add_op_create_pool('pool2')
-        expected = json.dumps({'api-version': 1,
-                               'ops': [{'op': 'create-pool', 'name': 'pool1',
-                                        'replicas': 1},
-                                       {'op': 'create-pool', 'name': 'pool2',
-                                        'replicas': 3}]})
-        self.assertEqual(rq.request, expected)
+        expected = {
+            'api-version': 1,
+            'request-id': 'uuid',
+            'ops': [{'op': 'create-pool', 'name': 'pool1', 'replicas': 1},
+                    {'op': 'create-pool', 'name': 'pool2', 'replicas': 3}]
+        }
+        request_dict = json.loads(rq.request)
+        for key in ['api-version', 'request-id']:
+            self.assertEqual(request_dict[key], expected[key])
+        for key in ['op', 'name', 'replicas']:
+            self.assertEqual(request_dict['ops'][0][key], expected['ops'][0][key])
+            self.assertEqual(request_dict['ops'][1][key], expected['ops'][1][key])
 
     def test_ceph_broker_rsp_class(self):
         rsp = ceph_utils.CephBrokerRsp(json.dumps({'exit-code': 0,
                                                    'stderr': "Success"}))
         self.assertEqual(rsp.exit_code, 0)
         self.assertEqual(rsp.exit_msg, "Success")
+        self.assertEqual(rsp.request_id, None)
+
+    def test_ceph_broker_rsp_class_rqid(self):
+        rsp = ceph_utils.CephBrokerRsp(json.dumps({'exit-code': 0,
+                                                   'stderr': "Success",
+                                                   'request-id': 'reqid1'}))
+        self.assertEqual(rsp.exit_code, 0)
+        self.assertEqual(rsp.exit_msg, 'Success')
+        self.assertEqual(rsp.request_id, 'reqid1')
+
+    def setup_client_relation(self, relation):
+        relation = FakeRelation(relation)
+        self.relation_get.side_effect = relation.get
+        self.relation_ids.side_effect = relation.relation_ids
+        self.related_units.side_effect = relation.related_units
+
+#    @patch.object(ceph_utils, 'uuid')
+#    @patch.object(ceph_utils, 'local_unit')
+#    def test_get_request_states(self, mlocal_unit, muuid):
+#        muuid.uuid1.return_value = '0bc7dc54'
+    @patch.object(ceph_utils, 'local_unit')
+    def test_get_request_states(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        expect = {'ceph:8': {'complete': True, 'sent': True}}
+        self.assertEqual(ceph_utils.get_request_states(rq), expect)
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_get_request_states_newrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=4)
+        expect = {'ceph:8': {'complete': False, 'sent': False}}
+        self.assertEqual(ceph_utils.get_request_states(rq), expect)
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_get_request_states_pendingrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION)
+        del rel['ceph:8']['ceph/0']['broker-rsp-glance-0']
+        self.setup_client_relation(rel)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        expect = {'ceph:8': {'complete': False, 'sent': True}}
+        self.assertEqual(ceph_utils.get_request_states(rq), expect)
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_get_request_states_failedrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION)
+        rel['ceph:8']['ceph/0']['broker-rsp-glance-0'] = '{"request-id": "0bc7dc54", "exit-code": 1}'
+        self.setup_client_relation(rel)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        expect = {'ceph:8': {'complete': False, 'sent': True}}
+        self.assertEqual(ceph_utils.get_request_states(rq), expect)
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_sent(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertTrue(ceph_utils.is_request_sent(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_sent_newrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=4)
+        self.assertFalse(ceph_utils.is_request_sent(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_sent_pending(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION)
+        del rel['ceph:8']['ceph/0']['broker-rsp-glance-0']
+        self.setup_client_relation(rel)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertTrue(ceph_utils.is_request_sent(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_sent_legacy(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION_LEGACY)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertTrue(ceph_utils.is_request_sent(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_sent_legacy_newrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION_LEGACY)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=4)
+        self.assertFalse(ceph_utils.is_request_sent(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_sent_legacy_pending(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION_LEGACY)
+        del rel['ceph:8']['ceph/0']['broker_rsp']
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertTrue(ceph_utils.is_request_sent(rq))
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = '0bc7dc54'
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertTrue(ceph_utils.is_request_complete(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_newrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=4)
+        self.assertFalse(ceph_utils.is_request_complete(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_pending(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION)
+        del rel['ceph:8']['ceph/0']['broker-rsp-glance-0']
+        self.setup_client_relation(rel)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertFalse(ceph_utils.is_request_complete(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_legacy(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION_LEGACY)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertTrue(ceph_utils.is_request_complete(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_legacy_newrq(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION_LEGACY)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=4)
+        self.assertFalse(ceph_utils.is_request_complete(rq))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_legacy_pending(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION_LEGACY)
+        del rel['ceph:8']['ceph/0']['broker_rsp']
+        self.setup_client_relation(rel)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        self.assertFalse(ceph_utils.is_request_complete(rq))
+
+    def test_equivalent_broker_requests(self):
+        rq1 = ceph_utils.CephBrokerRq()
+        rq1.add_op_create_pool(name='glance', replica_count=4)
+        rq2 = ceph_utils.CephBrokerRq()
+        rq2.add_op_create_pool(name='glance', replica_count=4)
+        self.assertTrue(rq1 == rq2)
+
+    def test_equivalent_broker_requests_diff1(self):
+        rq1 = ceph_utils.CephBrokerRq()
+        rq1.add_op_create_pool(name='glance', replica_count=3)
+        rq2 = ceph_utils.CephBrokerRq()
+        rq2.add_op_create_pool(name='glance', replica_count=4)
+        self.assertFalse(rq1 == rq2)
+
+    def test_equivalent_broker_requests_diff2(self):
+        rq1 = ceph_utils.CephBrokerRq()
+        rq1.add_op_create_pool(name='glance', replica_count=3)
+        rq2 = ceph_utils.CephBrokerRq()
+        rq2.add_op_create_pool(name='cinder', replica_count=3)
+        self.assertFalse(rq1 == rq2)
+
+    def test_equivalent_broker_requests_diff3(self):
+        rq1 = ceph_utils.CephBrokerRq()
+        rq1.add_op_create_pool(name='glance', replica_count=3)
+        rq2 = ceph_utils.CephBrokerRq(api_version=2)
+        rq2.add_op_create_pool(name='glance', replica_count=3)
+        self.assertFalse(rq1 == rq2)
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_for_rid(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = '0bc7dc54'
+        req = ceph_utils.CephBrokerRq()
+        req.add_op_create_pool(name='glance', replica_count=3)
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        self.assertTrue(ceph_utils.is_request_complete_for_rid(req, 'ceph:8'))
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_for_rid_newrq(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = 'a44c0fa6'
+        req = ceph_utils.CephBrokerRq()
+        req.add_op_create_pool(name='glance', replica_count=4)
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        self.assertFalse(ceph_utils.is_request_complete_for_rid(req, 'ceph:8'))
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_for_rid_failed(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = '0bc7dc54'
+        req = ceph_utils.CephBrokerRq()
+        req.add_op_create_pool(name='glance', replica_count=4)
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION)
+        rel['ceph:8']['ceph/0']['broker-rsp-glance-0'] = '{"request-id": "0bc7dc54", "exit-code": 1}'
+        self.setup_client_relation(rel)
+        self.assertFalse(ceph_utils.is_request_complete_for_rid(req, 'ceph:8'))
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_for_rid_pending(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = '0bc7dc54'
+        req = ceph_utils.CephBrokerRq()
+        req.add_op_create_pool(name='glance', replica_count=4)
+        mlocal_unit.return_value = 'glance/0'
+        rel = copy.deepcopy(CEPH_CLIENT_RELATION)
+        del rel['ceph:8']['ceph/0']['broker-rsp-glance-0']
+        self.setup_client_relation(rel)
+        self.assertFalse(ceph_utils.is_request_complete_for_rid(req, 'ceph:8'))
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_is_request_complete_for_rid_legacy(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = '0bc7dc54'
+        req = ceph_utils.CephBrokerRq()
+        req.add_op_create_pool(name='glance', replica_count=3)
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION_LEGACY)
+        self.assertTrue(ceph_utils.is_request_complete_for_rid(req, 'ceph:8'))
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_get_broker_rsp_key(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.assertEqual(ceph_utils.get_broker_rsp_key(), 'broker-rsp-glance-0')
+
+    @patch.object(ceph_utils, 'local_unit')
+    def test_send_request_if_needed(self, mlocal_unit):
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=3)
+        ceph_utils.send_request_if_needed(rq)
+        self.relation_set.assert_has_calls([])
+
+    @patch.object(ceph_utils, 'uuid')
+    @patch.object(ceph_utils, 'local_unit')
+    def test_send_request_if_needed_newrq(self, mlocal_unit, muuid):
+        muuid.uuid1.return_value = 'de67511e'
+        mlocal_unit.return_value = 'glance/0'
+        self.setup_client_relation(CEPH_CLIENT_RELATION)
+        rq = ceph_utils.CephBrokerRq()
+        rq.add_op_create_pool(name='glance', replica_count=4)
+        ceph_utils.send_request_if_needed(rq)
+        actual = json.loads(self.relation_set.call_args_list[0][1]['broker_req'])
+        self.assertEqual(actual['api-version'], 1)
+        self.assertEqual(actual['request-id'], 'de67511e')
+        self.assertEqual(actual['ops'][0]['replicas'], 4)
+        self.assertEqual(actual['ops'][0]['op'], 'create-pool')
+        self.assertEqual(actual['ops'][0]['name'], 'glance')

=== modified file 'tests/helpers.py'
--- tests/helpers.py	2014-11-25 15:07:02 +0000
+++ tests/helpers.py	2015-09-10 09:26:02 +0000
@@ -74,12 +74,12 @@
     def __init__(self, relation_data):
         self.relation_data = relation_data
 
-    def get(self, attr=None, unit=None, rid=None):
+    def get(self, attribute=None, unit=None, rid=None):
         if not rid or rid == 'foo:0':
-            if attr is None:
+            if attribute is None:
                 return self.relation_data
-            elif attr in self.relation_data:
-                return self.relation_data[attr]
+            elif attribute in self.relation_data:
+                return self.relation_data[attribute]
             return None
         else:
             if rid not in self.relation_data:
@@ -88,9 +88,9 @@
                 relation = self.relation_data[rid][unit]
             except KeyError:
                 return None
-            if attr in relation:
-                return relation[attr]
-            return None
+            if attribute and attribute in relation:
+                    return relation[attribute]
+            return relation
 
     def relation_ids(self, relation=None):
         return self.relation_data.keys()