utilities: Convert juju base into Ubuntu series

Juju 3.x replaced the `series` status key with a `base` key that
consists of Distribution type and version number.

To avoid maintenance burden we add a Launchpad module that
implements functions to look up available Ubuntu series data.

Update the `get_machine_series` helper function to determine
Ubuntu series from `base` when no `series` key is available.

Signed-off-by: Frode Nordahl <[email protected]>

unit_tests/utilities/test_zaza_utilities_juju.py

@@ -370,16 +370,76 @@ 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_exceptions(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.side_effect = KeyError
+        try:
+            juju_utils.get_machine_series('6')
+        except KeyError:
+            self.fail('Did not expect `get_machine_series` '
+                      'to raise a KeyError')
+        self._get_machine_status.reset_mock()
+
+        self._get_machine_status.return_value = {}
+        self.assertRaises(
+            ValueError,
+            juju_utils.get_machine_series,
+            '6')
+
+        base = mock.MagicMock()
+        base.name = 'someOtherDistro'
+        base.channel = '22.04/stable'
+        self._get_machine_status.return_value = {'base': base}
+        self.assertRaises(
+            NotImplementedError,
+            juju_utils.get_machine_series,
+            '6')
+
+    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'
+        )
+        base = mock.MagicMock()
+        base.name = 'ubuntu'
+        base.channel = '22.04/stable'
+        self._get_machine_status.return_value = {'base': base}
+        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_once_with()
+        self.assertEqual(expected, actual)
+
     def test_get_subordinate_units(self):
         juju_status = mock.MagicMock()
         juju_status.applications = {

unit_tests/utilities/test_zaza_utilities_launchpad.py

@@ -0,0 +1,55 @@
+# 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.
+
+import json
+import unittest
+
+import unit_tests.utils as ut_utils
+import zaza.utilities.launchpad as launchpad
+
+
+class TestUtilitiesLaunchpad(ut_utils.BaseTestCase):
+
+    def test_get_ubuntu_series(self):
+        self.patch_object(launchpad.requests, 'get')
+        expect = {'entries': {}}
+        r = unittest.mock.MagicMock()
+        r.text = json.dumps(expect)
+        self.get.return_value = r
+        self.assertEquals(
+            launchpad.get_ubuntu_series(),
+            expect,
+        )
+        self.get.assert_called_once_with(
+            'https://api.launchpad.net/devel/ubuntu/series')
+
+    def test_get_ubuntu_series_by_version(self):
+        self.patch_object(launchpad, 'get_ubuntu_series')
+
+        self.get_ubuntu_series.return_value = {
+            'entries': [{'version': 'fakeVersion'}]}
+
+        self.assertEquals(
+            launchpad.get_ubuntu_series_by_version(),
+            {'fakeVersion': {'version': 'fakeVersion'}})
+
+    def test_get_ubuntu_series_by_name(self):
+        self.patch_object(launchpad, 'get_ubuntu_series')
+
+        self.get_ubuntu_series.return_value = {
+            'entries': [{'name': 'fakeName'}]}
+
+        self.assertEquals(
+            launchpad.get_ubuntu_series_by_name(),
+            {'fakeName': {'name': 'fakeName'}})

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,28 @@ 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
+
+    base = status.get('base')
+    if not base:
+        raise ValueError("Unable to determine distro from status: '{}'"
+                         .format(status))
+    if base.name != 'ubuntu':
+        raise NotImplementedError("Series resolution not implemented for "
+                                  "distro: '{}'".format(base.name))
+
+    version, risk = base.channel.split('/')
+    return launchpad.get_ubuntu_series_by_version()[version]['name']
 
 
 def get_machine_uuids_for_application(application, model_name=None):

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', {})
+    }