Deprecate zaza.openstack.utilities.juju (#340)
This is the 3rd change in the effort to deprecate zaza.openstack.utilities.juju in favour of zaza.utilities.juju. All functions now just wrap their equivalents in zaza.utilities.juju and a decorator has been added which logs a warning if the function is used.
This commit is contained in:
Liam Young
GitHub
@@ -1,312 +0,0 @@
|
||||
# Copyright 2018 Canonical Ltd.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import mock
|
||||
import unit_tests.utils as ut_utils
|
||||
from zaza.openstack.utilities import juju as juju_utils
|
||||
|
||||
|
||||
class TestJujuUtils(ut_utils.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestJujuUtils, self).setUp()
|
||||
|
||||
# Juju Status Object and data
|
||||
self.key = "instance-id"
|
||||
self.key_data = "machine-uuid"
|
||||
self.machine = "1"
|
||||
self.machine_data = {self.key: self.key_data}
|
||||
self.unit = "app/1"
|
||||
self.unit_data = {"machine": self.machine}
|
||||
self.application = "app"
|
||||
self.application_data = {"units": {self.unit: self.unit_data}}
|
||||
self.subordinate_application = "subordinate_application"
|
||||
self.subordinate_application_data = {
|
||||
"subordinate-to": [self.application]}
|
||||
self.juju_status = mock.MagicMock()
|
||||
self.juju_status.name = "juju_status_object"
|
||||
self.juju_status.applications.get.return_value = self.application_data
|
||||
self.juju_status.machines.get.return_value = self.machine_data
|
||||
|
||||
# Model
|
||||
self.patch_object(juju_utils, "model")
|
||||
self.model_name = "model-name"
|
||||
self.model.get_juju_model.return_value = self.model_name
|
||||
self.model.get_status.return_value = self.juju_status
|
||||
self.run_output = {"Code": "0", "Stderr": "", "Stdout": "RESULT"}
|
||||
self.error_run_output = {"Code": "1", "Stderr": "ERROR", "Stdout": ""}
|
||||
self.model.run_on_unit.return_value = self.run_output
|
||||
|
||||
# Clouds
|
||||
self.cloud_name = "FakeCloudName"
|
||||
self.cloud_type = "FakeCloudType"
|
||||
self.clouds = {
|
||||
"clouds":
|
||||
{self.cloud_name:
|
||||
{"type": self.cloud_type}}}
|
||||
|
||||
# Controller
|
||||
self.patch_object(juju_utils, "controller")
|
||||
self.controller.get_cloud.return_value = self.cloud_name
|
||||
|
||||
def test_get_application_status(self):
|
||||
self.patch_object(juju_utils, "get_full_juju_status")
|
||||
self.get_full_juju_status.return_value = self.juju_status
|
||||
|
||||
# Full status juju object return
|
||||
self.assertEqual(
|
||||
juju_utils.get_application_status(), self.juju_status)
|
||||
self.get_full_juju_status.assert_called_once()
|
||||
|
||||
# Application only dictionary return
|
||||
self.assertEqual(
|
||||
juju_utils.get_application_status(application=self.application),
|
||||
self.application_data)
|
||||
|
||||
# Unit no application dictionary return
|
||||
self.assertEqual(
|
||||
juju_utils.get_application_status(unit=self.unit),
|
||||
self.unit_data)
|
||||
|
||||
def test_get_cloud_configs(self):
|
||||
self.patch_object(juju_utils.Path, "home")
|
||||
self.patch_object(juju_utils.generic_utils, "get_yaml_config")
|
||||
self.get_yaml_config.return_value = self.clouds
|
||||
|
||||
# All the cloud configs
|
||||
self.assertEqual(juju_utils.get_cloud_configs(), self.clouds)
|
||||
|
||||
# With cloud specified
|
||||
self.assertEqual(juju_utils.get_cloud_configs(self.cloud_name),
|
||||
self.clouds["clouds"][self.cloud_name])
|
||||
|
||||
def test_get_full_juju_status(self):
|
||||
self.assertEqual(juju_utils.get_full_juju_status(), self.juju_status)
|
||||
self.model.get_status.assert_called_once_with(model_name=None)
|
||||
|
||||
def test_get_machines_for_application(self):
|
||||
self.patch_object(juju_utils, "get_application_status")
|
||||
self.get_application_status.return_value = self.application_data
|
||||
|
||||
# Machine data
|
||||
self.assertEqual(
|
||||
next(juju_utils.get_machines_for_application(self.application)),
|
||||
self.machine)
|
||||
self.get_application_status.assert_called_once()
|
||||
|
||||
# Subordinate application has no units
|
||||
def _get_application_status(application, model_name=None):
|
||||
_apps = {
|
||||
self.application: self.application_data,
|
||||
self.subordinate_application:
|
||||
self.subordinate_application_data}
|
||||
return _apps[application]
|
||||
self.get_application_status.side_effect = _get_application_status
|
||||
|
||||
self.assertEqual(
|
||||
next(juju_utils.get_machines_for_application(
|
||||
self.subordinate_application)),
|
||||
self.machine)
|
||||
|
||||
def test_get_unit_name_from_host_name(self):
|
||||
unit_mock1 = mock.MagicMock()
|
||||
unit_mock1.data = {'machine-id': 12}
|
||||
unit_mock1.entity_id = 'myapp/2'
|
||||
unit_mock2 = mock.MagicMock()
|
||||
unit_mock2.data = {'machine-id': 15}
|
||||
unit_mock2.entity_id = 'myapp/5'
|
||||
self.model.get_units.return_value = [unit_mock1, unit_mock2]
|
||||
self.assertEqual(
|
||||
juju_utils.get_unit_name_from_host_name('juju-model-12', 'myapp'),
|
||||
'myapp/2')
|
||||
|
||||
def test_get_machine_status(self):
|
||||
self.patch_object(juju_utils, "get_full_juju_status")
|
||||
self.get_full_juju_status.return_value = self.juju_status
|
||||
|
||||
# All machine data
|
||||
self.assertEqual(
|
||||
juju_utils.get_machine_status(self.machine),
|
||||
self.machine_data)
|
||||
self.get_full_juju_status.assert_called_once()
|
||||
|
||||
# Request a specific key
|
||||
self.assertEqual(
|
||||
juju_utils.get_machine_status(self.machine, self.key),
|
||||
self.key_data)
|
||||
|
||||
def test_get_machine_uuids_for_application(self):
|
||||
self.patch_object(juju_utils, "get_machines_for_application")
|
||||
self.get_machines_for_application.return_value = [self.machine]
|
||||
|
||||
self.assertEqual(
|
||||
next(juju_utils.get_machine_uuids_for_application(
|
||||
self.application)),
|
||||
self.machine_data.get("instance-id"))
|
||||
self.get_machines_for_application.assert_called_once_with(
|
||||
self.application, model_name=None)
|
||||
|
||||
def test_get_provider_type(self):
|
||||
self.patch_object(juju_utils, "get_cloud_configs")
|
||||
self.get_cloud_configs.return_value = {"type": self.cloud_type}
|
||||
self.assertEqual(juju_utils.get_provider_type(),
|
||||
self.cloud_type)
|
||||
self.get_cloud_configs.assert_called_once_with(self.cloud_name)
|
||||
|
||||
def test_remote_run(self):
|
||||
_cmd = "do the thing"
|
||||
|
||||
# Success
|
||||
self.assertEqual(juju_utils.remote_run(self.unit, _cmd),
|
||||
self.run_output["Stdout"])
|
||||
self.model.run_on_unit.assert_called_once_with(
|
||||
self.unit, _cmd, timeout=None, model_name=None)
|
||||
|
||||
# Non-fatal failure
|
||||
self.model.run_on_unit.return_value = self.error_run_output
|
||||
self.assertEqual(
|
||||
juju_utils.remote_run(
|
||||
self.unit,
|
||||
_cmd,
|
||||
fatal=False,
|
||||
model_name=None),
|
||||
self.error_run_output["Stderr"])
|
||||
|
||||
# Fatal failure
|
||||
with self.assertRaises(Exception):
|
||||
juju_utils.remote_run(self.unit, _cmd, fatal=True)
|
||||
|
||||
def test_get_unit_names(self):
|
||||
self.patch('zaza.model.get_first_unit_name', new_callable=mock.Mock(),
|
||||
name='_get_first_unit_name')
|
||||
juju_utils._get_unit_names(['aunit/0', 'otherunit/0'])
|
||||
self.assertFalse(self._get_first_unit_name.called)
|
||||
|
||||
def test_get_unit_names_called_with_application_name(self):
|
||||
self.patch_object(juju_utils, 'model')
|
||||
juju_utils._get_unit_names(['aunit', 'otherunit/0'])
|
||||
self.model.get_first_unit_name.assert_called()
|
||||
|
||||
def test_get_relation_from_unit(self):
|
||||
self.patch_object(juju_utils, '_get_unit_names')
|
||||
self.patch_object(juju_utils, 'yaml')
|
||||
self.patch_object(juju_utils, 'model')
|
||||
self._get_unit_names.return_value = ['aunit/0', 'otherunit/0']
|
||||
data = {'foo': 'bar'}
|
||||
self.model.get_relation_id.return_value = 42
|
||||
self.model.run_on_unit.return_value = {'Code': 0, 'Stdout': str(data)}
|
||||
juju_utils.get_relation_from_unit('aunit/0', 'otherunit/0',
|
||||
'arelation')
|
||||
self.model.run_on_unit.assert_called_with(
|
||||
'aunit/0',
|
||||
'relation-get --format=yaml -r "42" - "otherunit/0"',
|
||||
model_name=None)
|
||||
self.yaml.safe_load.assert_called_with(str(data))
|
||||
|
||||
def test_get_relation_from_unit_fails(self):
|
||||
self.patch_object(juju_utils, '_get_unit_names')
|
||||
self.patch_object(juju_utils, 'yaml')
|
||||
self.patch_object(juju_utils, 'model')
|
||||
self._get_unit_names.return_value = ['aunit/0', 'otherunit/0']
|
||||
self.model.get_relation_id.return_value = 42
|
||||
self.model.run_on_unit.return_value = {'Code': 1, 'Stderr': 'ERROR'}
|
||||
with self.assertRaises(Exception):
|
||||
juju_utils.get_relation_from_unit('aunit/0', 'otherunit/0',
|
||||
'arelation')
|
||||
self.model.run_on_unit.assert_called_with(
|
||||
'aunit/0',
|
||||
'relation-get --format=yaml -r "42" - "otherunit/0"',
|
||||
model_name=None)
|
||||
self.assertFalse(self.yaml.safe_load.called)
|
||||
|
||||
def test_leader_get(self):
|
||||
self.patch_object(juju_utils, 'yaml')
|
||||
self.patch_object(juju_utils, 'model')
|
||||
data = {'foo': 'bar'}
|
||||
self.model.run_on_leader.return_value = {
|
||||
'Code': 0, 'Stdout': str(data)}
|
||||
juju_utils.leader_get('application')
|
||||
self.model.run_on_leader.assert_called_with(
|
||||
'application', 'leader-get --format=yaml ', model_name=None)
|
||||
self.yaml.safe_load.assert_called_with(str(data))
|
||||
|
||||
def test_leader_get_key(self):
|
||||
self.patch_object(juju_utils, 'yaml')
|
||||
self.patch_object(juju_utils, 'model')
|
||||
data = {'foo': 'bar'}
|
||||
self.model.run_on_leader.return_value = {
|
||||
'Code': 0, 'Stdout': data['foo']}
|
||||
juju_utils.leader_get('application', 'foo')
|
||||
self.model.run_on_leader.assert_called_with(
|
||||
'application', 'leader-get --format=yaml foo', model_name=None)
|
||||
self.yaml.safe_load.assert_called_with(data['foo'])
|
||||
|
||||
def test_leader_get_fails(self):
|
||||
self.patch_object(juju_utils, 'yaml')
|
||||
self.patch_object(juju_utils, 'model')
|
||||
self.model.run_on_leader.return_value = {
|
||||
'Code': 1, 'Stderr': 'ERROR'}
|
||||
with self.assertRaises(Exception):
|
||||
juju_utils.leader_get('application')
|
||||
self.model.run_on_leader.assert_called_with(
|
||||
'application', 'leader-get --format=yaml ',
|
||||
model_name=None)
|
||||
self.assertFalse(self.yaml.safe_load.called)
|
||||
|
||||
def test_get_machine_series(self):
|
||||
self.patch(
|
||||
'zaza.openstack.utilities.juju.get_machine_status',
|
||||
new_callable=mock.MagicMock(),
|
||||
name='_get_machine_status'
|
||||
)
|
||||
self._get_machine_status.return_value = 'xenial'
|
||||
expected = 'xenial'
|
||||
actual = juju_utils.get_machine_series('6')
|
||||
self._get_machine_status.assert_called_with(
|
||||
machine='6',
|
||||
key='series',
|
||||
model_name=None
|
||||
)
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
def test_get_subordinate_units(self):
|
||||
juju_status = mock.MagicMock()
|
||||
juju_status.applications = {
|
||||
'nova-compute': {
|
||||
'units': {
|
||||
'nova-compute/0': {
|
||||
'subordinates': {
|
||||
'neutron-openvswitch/2': {
|
||||
'charm': 'cs:neutron-openvswitch-22'}}}}},
|
||||
'cinder': {
|
||||
'units': {
|
||||
'cinder/1': {
|
||||
'subordinates': {
|
||||
'cinder-hacluster/0': {
|
||||
'charm': 'cs:hacluster-42'},
|
||||
'cinder-ceph/3': {
|
||||
'charm': 'cs:cinder-ceph-2'}}}}},
|
||||
}
|
||||
self.assertEqual(
|
||||
sorted(juju_utils.get_subordinate_units(
|
||||
['nova-compute/0', 'cinder/1'],
|
||||
status=juju_status)),
|
||||
sorted(['neutron-openvswitch/2', 'cinder-hacluster/0',
|
||||
'cinder-ceph/3']))
|
||||
self.assertEqual(
|
||||
juju_utils.get_subordinate_units(
|
||||
['nova-compute/0', 'cinder/1'],
|
||||
charm_name='ceph',
|
||||
status=juju_status),
|
||||
['cinder-ceph/3'])
|
||||
@@ -13,18 +13,31 @@
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
"""Module for interacting with juju."""
|
||||
import os
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
"""Deprecated, please use zaza.utilities.juju."""
|
||||
|
||||
from zaza import (
|
||||
model,
|
||||
controller,
|
||||
)
|
||||
from zaza.openstack.utilities import generic as generic_utils
|
||||
import logging
|
||||
import functools
|
||||
|
||||
import zaza.utilities.juju
|
||||
|
||||
|
||||
def deprecate():
|
||||
"""Add a deprecation warning to wrapped function."""
|
||||
def wrap(f):
|
||||
|
||||
@functools.wraps(f)
|
||||
def wrapped_f(*args, **kwargs):
|
||||
msg = (
|
||||
"{} from zaza.openstack.utilities.juju is deprecated. "
|
||||
"Please use the equivalent from zaza.utilities.juju".format(
|
||||
f.__name__))
|
||||
logging.warning(msg)
|
||||
return f(*args, **kwargs)
|
||||
return wrapped_f
|
||||
return wrap
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_application_status(application=None, unit=None, model_name=None):
|
||||
"""Return the juju status for an application.
|
||||
|
||||
@@ -37,39 +50,29 @@ def get_application_status(application=None, unit=None, model_name=None):
|
||||
:returns: Juju status output for an application
|
||||
:rtype: dict
|
||||
"""
|
||||
status = get_full_juju_status()
|
||||
|
||||
if unit and not application:
|
||||
application = unit.split("/")[0]
|
||||
|
||||
if application:
|
||||
status = status.applications.get(application)
|
||||
if unit:
|
||||
status = status.get("units").get(unit)
|
||||
return status
|
||||
return zaza.utilities.juju.get_application_status(
|
||||
application=application,
|
||||
unit=unit,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
def get_application_ip(application):
|
||||
@deprecate()
|
||||
def get_application_ip(application, model_name=None):
|
||||
"""Get the application's IP address.
|
||||
|
||||
:param application: Application name
|
||||
:type application: str
|
||||
:param model_name: Name of model to query.
|
||||
:type model_name: str
|
||||
:returns: Application's IP address
|
||||
:rtype: str
|
||||
"""
|
||||
try:
|
||||
app_config = model.get_application_config(application)
|
||||
except KeyError:
|
||||
return ''
|
||||
vip = app_config.get("vip").get("value")
|
||||
if vip:
|
||||
ip = vip
|
||||
else:
|
||||
unit = model.get_units(application)[0]
|
||||
ip = unit.public_address
|
||||
return ip
|
||||
return zaza.utilities.juju.get_application_ip(
|
||||
application,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_cloud_configs(cloud=None):
|
||||
"""Get cloud configuration from local clouds.yaml.
|
||||
|
||||
@@ -81,14 +84,11 @@ def get_cloud_configs(cloud=None):
|
||||
:returns: Dictionary of cloud configuration
|
||||
:rtype: dict
|
||||
"""
|
||||
home = str(Path.home())
|
||||
cloud_config = os.path.join(home, ".local", "share", "juju", "clouds.yaml")
|
||||
if cloud:
|
||||
return generic_utils.get_yaml_config(cloud_config)["clouds"].get(cloud)
|
||||
else:
|
||||
return generic_utils.get_yaml_config(cloud_config)
|
||||
return zaza.utilities.juju.get_cloud_configs(
|
||||
cloud=cloud)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_full_juju_status(model_name=None):
|
||||
"""Return the full juju status output.
|
||||
|
||||
@@ -97,10 +97,11 @@ def get_full_juju_status(model_name=None):
|
||||
:returns: Full juju status output
|
||||
:rtype: dict
|
||||
"""
|
||||
status = model.get_status(model_name=model_name)
|
||||
return status
|
||||
return zaza.utilities.juju.get_full_juju_status(
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_machines_for_application(application, model_name=None):
|
||||
"""Return machines for a given application.
|
||||
|
||||
@@ -111,20 +112,12 @@ def get_machines_for_application(application, model_name=None):
|
||||
:returns: machines for an application
|
||||
:rtype: Iterator[str]
|
||||
"""
|
||||
status = get_application_status(application, model_name=model_name)
|
||||
if not status:
|
||||
return
|
||||
|
||||
# libjuju juju status no longer has units for subordinate charms
|
||||
# Use the application it is subordinate-to to find machines
|
||||
if not status.get("units") and status.get("subordinate-to"):
|
||||
status = get_application_status(status.get("subordinate-to")[0],
|
||||
model_name=model_name)
|
||||
|
||||
for unit in status.get("units").keys():
|
||||
yield status.get("units").get(unit).get("machine")
|
||||
return zaza.utilities.juju.get_machines_for_application(
|
||||
application,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_unit_name_from_host_name(host_name, application, model_name=None):
|
||||
"""Return the juju unit name corresponding to a hostname.
|
||||
|
||||
@@ -135,17 +128,13 @@ def get_unit_name_from_host_name(host_name, application, model_name=None):
|
||||
:param model_name: Name of model to query.
|
||||
:type model_name: str
|
||||
"""
|
||||
# Assume that a juju managed hostname always ends in the machine number and
|
||||
# remove the domain name if it present.
|
||||
machine_number = host_name.split('-')[-1].split('.')[0]
|
||||
unit_names = [
|
||||
u.entity_id
|
||||
for u in model.get_units(application_name=application,
|
||||
model_name=model_name)
|
||||
if int(u.data['machine-id']) == int(machine_number)]
|
||||
return unit_names[0]
|
||||
return zaza.utilities.juju.get_unit_name_from_host_name(
|
||||
host_name,
|
||||
application,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_machine_status(machine, key=None, model_name=None):
|
||||
"""Return the juju status for a machine.
|
||||
|
||||
@@ -158,17 +147,13 @@ def get_machine_status(machine, key=None, model_name=None):
|
||||
:returns: Juju status output for a machine
|
||||
:rtype: dict
|
||||
"""
|
||||
status = get_full_juju_status(model_name=model_name)
|
||||
if "lxd" in machine:
|
||||
host = machine.split('/')[0]
|
||||
status = status.machines.get(host)['containers'][machine]
|
||||
else:
|
||||
status = status.machines.get(machine)
|
||||
if key:
|
||||
status = status.get(key)
|
||||
return status
|
||||
return zaza.utilities.juju.get_machine_status(
|
||||
machine,
|
||||
key=key,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_machine_series(machine, model_name=None):
|
||||
"""Return the juju series for a machine.
|
||||
|
||||
@@ -179,13 +164,12 @@ def get_machine_series(machine, model_name=None):
|
||||
:returns: Juju series
|
||||
:rtype: string
|
||||
"""
|
||||
return get_machine_status(
|
||||
machine=machine,
|
||||
key='series',
|
||||
model_name=model_name
|
||||
)
|
||||
return zaza.utilities.juju.get_machine_series(
|
||||
machine,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_machine_uuids_for_application(application, model_name=None):
|
||||
"""Return machine uuids for a given application.
|
||||
|
||||
@@ -196,30 +180,22 @@ def get_machine_uuids_for_application(application, model_name=None):
|
||||
:returns: machine uuuids for an application
|
||||
:rtype: Iterator[str]
|
||||
"""
|
||||
for machine in get_machines_for_application(application,
|
||||
model_name=model_name):
|
||||
yield get_machine_status(machine, key="instance-id",
|
||||
model_name=model_name)
|
||||
return zaza.utilities.juju.get_machine_uuids_for_application(
|
||||
application,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_provider_type():
|
||||
"""Get the type of the undercloud.
|
||||
|
||||
:returns: Name of the undercloud type
|
||||
:rtype: string
|
||||
"""
|
||||
cloud = controller.get_cloud()
|
||||
if cloud:
|
||||
# If the controller was deployed from this system with
|
||||
# the cloud configured in ~/.local/share/juju/clouds.yaml
|
||||
# Determine the cloud type directly
|
||||
return get_cloud_configs(cloud)["type"]
|
||||
else:
|
||||
# If the controller was deployed elsewhere
|
||||
# For now assume openstack
|
||||
return "openstack"
|
||||
return zaza.utilities.juju.get_provider_type()
|
||||
|
||||
|
||||
@deprecate()
|
||||
def remote_run(unit, remote_cmd, timeout=None, fatal=None, model_name=None):
|
||||
"""Run command on unit and return the output.
|
||||
|
||||
@@ -238,46 +214,15 @@ def remote_run(unit, remote_cmd, timeout=None, fatal=None, model_name=None):
|
||||
:rtype: string
|
||||
:raises: model.CommandRunFailed
|
||||
"""
|
||||
if fatal is None:
|
||||
fatal = True
|
||||
result = model.run_on_unit(
|
||||
return zaza.utilities.juju.remote_run(
|
||||
unit,
|
||||
remote_cmd,
|
||||
timeout=timeout,
|
||||
fatal=fatal,
|
||||
model_name=model_name)
|
||||
if result:
|
||||
if int(result.get("Code")) == 0:
|
||||
return result.get("Stdout")
|
||||
else:
|
||||
if fatal:
|
||||
raise model.CommandRunFailed(remote_cmd, result)
|
||||
return result.get("Stderr")
|
||||
|
||||
|
||||
def _get_unit_names(names, model_name=None):
|
||||
"""Resolve given application names to first unit name of said application.
|
||||
|
||||
Helper function that resolves application names to first unit name of
|
||||
said application. Any already resolved unit names are returned as-is.
|
||||
|
||||
:param names: List of units/applications to translate
|
||||
:type names: list(str)
|
||||
:param model_name: Name of model to query.
|
||||
:type model_name: str
|
||||
:returns: List of units
|
||||
:rtype: list(str)
|
||||
"""
|
||||
result = []
|
||||
for name in names:
|
||||
if '/' in name:
|
||||
result.append(name)
|
||||
else:
|
||||
result.append(model.get_first_unit_name(
|
||||
name,
|
||||
model_name=model_name))
|
||||
return result
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_relation_from_unit(entity, remote_entity, remote_interface_name,
|
||||
model_name=None):
|
||||
"""Get relation data passed between two units.
|
||||
@@ -302,22 +247,14 @@ def get_relation_from_unit(entity, remote_entity, remote_interface_name,
|
||||
:rtype: dict
|
||||
:raises: model.CommandRunFailed
|
||||
"""
|
||||
application = entity.split('/')[0]
|
||||
remote_application = remote_entity.split('/')[0]
|
||||
rid = model.get_relation_id(application, remote_application,
|
||||
remote_interface_name=remote_interface_name,
|
||||
model_name=model_name)
|
||||
(unit, remote_unit) = _get_unit_names(
|
||||
[entity, remote_entity],
|
||||
return zaza.utilities.juju.get_relation_from_unit(
|
||||
entity,
|
||||
remote_entity,
|
||||
remote_interface_name,
|
||||
model_name=model_name)
|
||||
cmd = 'relation-get --format=yaml -r "{}" - "{}"' .format(rid, remote_unit)
|
||||
result = model.run_on_unit(unit, cmd, model_name=model_name)
|
||||
if result and int(result.get('Code')) == 0:
|
||||
return yaml.safe_load(result.get('Stdout'))
|
||||
else:
|
||||
raise model.CommandRunFailed(cmd, result)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def leader_get(application, key='', model_name=None):
|
||||
"""Get leader settings from leader unit of named application.
|
||||
|
||||
@@ -331,14 +268,13 @@ def leader_get(application, key='', model_name=None):
|
||||
:rtype: dict
|
||||
:raises: model.CommandRunFailed
|
||||
"""
|
||||
cmd = 'leader-get --format=yaml {}'.format(key)
|
||||
result = model.run_on_leader(application, cmd, model_name=model_name)
|
||||
if result and int(result.get('Code')) == 0:
|
||||
return yaml.safe_load(result.get('Stdout'))
|
||||
else:
|
||||
raise model.CommandRunFailed(cmd, result)
|
||||
return zaza.utilities.juju.leader_get(
|
||||
application,
|
||||
key=key,
|
||||
model_name=model_name)
|
||||
|
||||
|
||||
@deprecate()
|
||||
def get_subordinate_units(unit_list, charm_name=None, status=None,
|
||||
model_name=None):
|
||||
"""Get a list of all subordinate units associated with units in unit_list.
|
||||
@@ -369,17 +305,8 @@ def get_subordinate_units(unit_list, charm_name=None, status=None,
|
||||
:returns: List of matching unit names.
|
||||
:rtype: []
|
||||
"""
|
||||
if not status:
|
||||
status = model.get_status(model_name=model_name)
|
||||
sub_units = []
|
||||
for unit_name in unit_list:
|
||||
app_name = unit_name.split('/')[0]
|
||||
subs = status.applications[app_name]['units'][unit_name].get(
|
||||
'subordinates') or {}
|
||||
if charm_name:
|
||||
for unit_name, unit_data in subs.items():
|
||||
if charm_name in unit_data['charm']:
|
||||
sub_units.append(unit_name)
|
||||
else:
|
||||
sub_units.extend([n for n in subs.keys()])
|
||||
return sub_units
|
||||
return zaza.utilities.juju.get_subordinate_units(
|
||||
unit_list,
|
||||
charm_name=charm_name,
|
||||
status=status,
|
||||
model_name=model_name)
|
||||
|
||||
Reference in New Issue
Block a user