diff --git a/unit_tests/utilities/test_zaza_utilities_juju.py b/unit_tests/utilities/test_zaza_utilities_juju.py index 05ea9cf5b..4569f981f 100644 --- a/unit_tests/utilities/test_zaza_utilities_juju.py +++ b/unit_tests/utilities/test_zaza_utilities_juju.py @@ -370,16 +370,38 @@ def test_get_machine_series(self): new_callable=mock.MagicMock(), name='_get_machine_status' ) - self._get_machine_status.return_value = 'xenial' + self._get_machine_status.return_value = {'series': '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_machine_series_juju3x(self): + self.patch( + 'zaza.utilities.juju.get_machine_status', + new_callable=mock.MagicMock(), + name='_get_machine_status' + ) + self.patch( + 'zaza.utilities.juju.launchpad.get_ubuntu_series_by_version', + new_callable=mock.MagicMock(), + name='_get_ubuntu_series_by_version' + ) + self._get_machine_status.return_value = {'base': 'ubuntu@22.04'} + self._get_ubuntu_series_by_version.return_value = { + '22.04': {'name': 'jammy'}} + expected = 'jammy' + actual = juju_utils.get_machine_series('6') + self._get_machine_status.assert_called_with( + machine='6', + model_name=None + ) + self._get_ubuntu_series_by_version.assert_called_with('22.04') + self.assertEqual(expected, actual) + def test_get_subordinate_units(self): juju_status = mock.MagicMock() juju_status.applications = { diff --git a/zaza/utilities/juju.py b/zaza/utilities/juju.py index d0ef1c54d..a9d117544 100644 --- a/zaza/utilities/juju.py +++ b/zaza/utilities/juju.py @@ -24,8 +24,9 @@ model, controller, ) -from zaza.utilities import generic as generic_utils from zaza.utilities import exceptions as zaza_exceptions +from zaza.utilities import generic as generic_utils +from zaza.utilities import launchpad KUBERNETES_PROVIDER_NAME = 'kubernetes' @@ -279,11 +280,26 @@ def get_machine_series(machine, model_name=None): :returns: Juju series :rtype: string """ - return get_machine_status( + status = get_machine_status( machine=machine, - key='series', model_name=model_name ) + try: + if 'series' in status: + return status.get('series') + except KeyError: + # libjuju will raise make the above check return KeyError when not + # present... + pass + + distro, version = status.get('base', '@').split('@') + if not distro: + raise ValueError("Unable to determine distro from status: '{}'" + .format(status)) + if distro != 'ubuntu': + raise NotImplementedError("Series resolution not implemented for " + "distro: '{}'".format(distro)) + return launchpad.get_ubuntu_series_by_version()[version]['name'] def get_machine_uuids_for_application(application, model_name=None): diff --git a/zaza/utilities/launchpad.py b/zaza/utilities/launchpad.py new file mode 100644 index 000000000..4ba7ee429 --- /dev/null +++ b/zaza/utilities/launchpad.py @@ -0,0 +1,52 @@ +# Copyright 2023 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. +"""Module for interacting with Launchpad API.""" + +import json +import requests +import typing + + +def get_ubuntu_series( +) -> typing.Dict[str, typing.List[typing.Dict[str, any]]]: + """Contact Launchpad API and retrieve a list of all Ubuntu releases. + + Launchpad documentation for the returned data structure can be found here: + https://launchpad.net/+apidoc/devel.html#distribution + https://launchpad.net/+apidoc/devel.html#distro_series + """ + r = requests.get('https://api.launchpad.net/devel/ubuntu/series') + return json.loads(r.text) + + +def get_ubuntu_series_by_version() -> typing.Dict[str, typing.Dict[str, any]]: + """Get a Dict of distro series information indexed by version number. + + Please refer to the `get_ubuntu_series()` function docstring for docs. + """ + return { + entry['version']: entry + for entry in get_ubuntu_series().get('entries', {}) + } + + +def get_ubuntu_series_by_name() -> typing.Dict[str, typing.Dict[str, any]]: + """Get a Dict of distro series information indexed by version name. + + Please refer to the `get_ubuntu_series()` function docstring for docs. + """ + return { + entry['name']: entry + for entry in get_ubuntu_series().get('entries', {}) + }