我们从Python开源项目中,提取了以下26个代码示例,用于说明如何使用charmhelpers.core.hookenv.relation_id()。
def test_relation_clear(self, local_unit, relation_get, relation_set): local_unit.return_value = 'local-unit' relation_get.return_value = { 'private-address': '10.5.0.1', 'foo': 'bar', 'public-address': '146.192.45.6' } hookenv.relation_clear('relation:1') relation_get.assert_called_with(rid='relation:1', unit='local-unit') relation_set.assert_called_with( relation_id='relation:1', **{'private-address': '10.5.0.1', 'foo': None, 'public-address': '146.192.45.6'})
def test_gets_execution_environment_no_relation( self, os_, relations_get, relations, relation_id, local_unit, relation_type, config): config.return_value = 'some-config' relation_type.return_value = 'some-type' local_unit.return_value = 'some-unit' relation_id.return_value = None relations.return_value = 'all-relations' relations_get.return_value = 'some-relations' os_.environ = 'some-environment' result = hookenv.execution_environment() self.assertEqual(result, { 'conf': 'some-config', 'unit': 'some-unit', 'rels': 'all-relations', 'env': 'some-environment', })
def test_gets_the_relation_id(self, os_, relation_ids, remote_service_name): os_.environ = { 'JUJU_RELATION_ID': 'foo', } self.assertEqual(hookenv.relation_id(), 'foo') relation_ids.return_value = ['r:1', 'r:2'] remote_service_name.side_effect = ['other', 'service'] self.assertEqual(hookenv.relation_id('rel', 'service/0'), 'r:2') relation_ids.assert_called_once_with('rel') self.assertEqual(remote_service_name.call_args_list, [ call('r:1'), call('r:2'), ]) remote_service_name.side_effect = ['other', 'service'] self.assertEqual(hookenv.relation_id('rel', 'service'), 'r:2')
def peer_store_and_set(relation_id=None, peer_relation_name='cluster', peer_store_fatal=False, relation_settings=None, delimiter='_', **kwargs): """Store passed-in arguments both in argument relation and in peer storage. It functions like doing relation_set() and peer_store() at the same time, with the same data. @param relation_id: the id of the relation to store the data on. Defaults to the current relation. @param peer_store_fatal: Set to True, the function will raise an exception should the peer sotrage not be avialable.""" relation_settings = relation_settings if relation_settings else {} relation_set(relation_id=relation_id, relation_settings=relation_settings, **kwargs) if is_relation_made(peer_relation_name): for key, value in six.iteritems(dict(list(kwargs.items()) + list(relation_settings.items()))): key_prefix = relation_id or current_relation_id() peer_store(key_prefix + delimiter + key, value, relation_name=peer_relation_name) else: if peer_store_fatal: raise ValueError('Unable to detect ' 'peer relation {}'.format(peer_relation_name))
def test_gets_related_units(self, relation_id, check_output): relid = 123 units = ['foo', 'bar'] relation_id.return_value = relid check_output.return_value = json.dumps(units).encode('UTF-8') result = hookenv.related_units() self.assertEqual(result, units) check_output.assert_called_with(['relation-list', '--format=json', '-r', relid])
def test_gets_related_units_empty_array(self, relation_id, check_output): relid = str(123) units = [] relation_id.return_value = relid check_output.return_value = json.dumps(None).encode('UTF-8') result = hookenv.related_units() self.assertEqual(result, units) check_output.assert_called_with(['relation-list', '--format=json', '-r', relid])
def test_related_units_no_relation(self, relation_id, check_output): units = ['foo', 'bar'] relation_id.return_value = None check_output.return_value = json.dumps(units).encode('UTF-8') result = hookenv.related_units() self.assertEqual(result, units) check_output.assert_called_with(['relation-list', '--format=json'])
def test_gets_related_units_for_id(self, relation_id, check_output): relid = 123 units = ['foo', 'bar'] check_output.return_value = json.dumps(units).encode('UTF-8') result = hookenv.related_units(relid) self.assertEqual(result, units) check_output.assert_called_with(['relation-list', '--format=json', '-r', relid]) self.assertFalse(relation_id.called)
def test_relation_id_none_if_no_env(self, os_): os_.environ = {} self.assertEqual(hookenv.relation_id(), None)
def test_sets_relation_with_relation_id(self, check_call_, check_output, local_unit): hookenv.relation_set(relation_id="foo", bar="baz") check_call_.assert_called_with(['relation-set', '-r', 'foo', 'bar=baz'])