我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用charmhelpers.core.hookenv.relation_set()。
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_relation_set_file(self, check_call, check_output, remove): """If relation-set accepts a --file parameter, it's used. Juju 1.23.2 introduced a --file parameter, which means you can pass the data through a file. Not using --file would make relation_set break if the relation data is too big. """ # check_output(["relation-set", "--help"]) is used to determine # whether we can pass --file to it. check_output.return_value = "--file" hookenv.relation_set(foo="bar") check_output.assert_called_with( ["relation-set", "--help"], universal_newlines=True) # relation-set is called with relation-set --file <temp_file> # with data as YAML and the temp_file is then removed. self.assertEqual(1, len(check_call.call_args[0])) command = check_call.call_args[0][0] self.assertEqual(3, len(command)) self.assertEqual("relation-set", command[0]) self.assertEqual("--file", command[1]) temp_file = command[2] with open(temp_file, "r") as f: self.assertEqual("{foo: bar}", f.read().strip()) remove.assert_called_with(temp_file)
def test_relation_set_file_non_str(self, check_call, check_output, remove): """If relation-set accepts a --file parameter, it's used. Any value that is not a string is converted to a string before encoding the settings to YAML. """ # check_output(["relation-set", "--help"]) is used to determine # whether we can pass --file to it. check_output.return_value = "--file" hookenv.relation_set(foo={"bar": 1}) check_output.assert_called_with( ["relation-set", "--help"], universal_newlines=True) # relation-set is called with relation-set --file <temp_file> # with data as YAML and the temp_file is then removed. self.assertEqual(1, len(check_call.call_args[0])) command = check_call.call_args[0][0] self.assertEqual(3, len(command)) self.assertEqual("relation-set", command[0]) self.assertEqual("--file", command[1]) temp_file = command[2] with open(temp_file, "r") as f: self.assertEqual("{foo: '{''bar'': 1}'}", f.read().strip()) remove.assert_called_with(temp_file)
def _save_state(self): self.msg('Publishing state'.format(self._name())) if hookenv.is_leader(): # sort_keys to ensure stability. raw = json.dumps(self.grants, sort_keys=True) hookenv.leader_set({self.key: raw}) local_unit = hookenv.local_unit() if self.relid is None: # No peers relation yet. Fallback to local state. self.msg('No peer relation. Saving local state') self._save_local_state(self.requests[local_unit]) else: # sort_keys to ensure stability. raw = json.dumps(self.requests[local_unit], sort_keys=True) hookenv.relation_set(self.relid, relation_settings={self.key: raw})
def provide_data(self): """ Set the relation data for each provider in the ``provided_data`` list. A provider must have a `name` attribute, which indicates which relation to set data on, and a `provide_data()` method, which returns a dict of data to set. The `provide_data()` method can optionally accept two parameters: * ``remote_service`` The name of the remote service that the data will be provided to. The `provide_data()` method will be called once for each connected service (not unit). This allows the method to tailor its data to the given service. * ``service_ready`` Whether or not the service definition had all of its requirements met, and thus the ``data_ready`` callbacks run. Note that the ``provided_data`` methods are now called **after** the ``data_ready`` callbacks are run. This gives the ``data_ready`` callbacks a chance to generate any data necessary for the providing to the remote services. """ for service_name, service in self.services.items(): service_ready = self.is_ready(service_name) for provider in service.get('provided_data', []): for relid in hookenv.relation_ids(provider.name): units = hookenv.related_units(relid) if not units: continue remote_service = units[0].split('/')[0] argspec = getargspec(provider.provide_data) if len(argspec.args) > 1: data = provider.provide_data(remote_service, service_ready) else: data = provider.provide_data() if data: hookenv.relation_set(relid, data)
