[3.6] fix: make distro-info tests unambiguous for plucky/questing / fix: test_power_type_set_with_parameters flaky test for redfish test case

src/maasserver/forms/tests/test_commissioning.py

@@ -36,25 +36,26 @@ def test_commissioningform_error_msg_lists_min_hwe_kernel_choices(self):
 
     def test_commissioningform_contains_real_and_ui_choice(self):
         release = factory.pick_ubuntu_release()
-        name = "ubuntu/%s" % release
+        series = release.series
+        name = "ubuntu/%s" % series
         arch = factory.make_name("arch")
-        kernel = "hwe-" + release[0]
+        kernel = "hwe-" + release.version.split(" ")[0]
         # Disable boot sources signals otherwise the test fails due to unrun
         # post-commit tasks at the end of the test.
         self.useFixture(SignalsDisabled("bootsources"))
         factory.make_RegionController()
-        factory.make_BootSourceCache(os=name, subarch=kernel, release=release)
+        factory.make_BootSourceCache(os=name, subarch=kernel, release=series)
         factory.make_usable_boot_resource(
             name=name,
             architecture=f"{arch}/{kernel}",
             rtype=BOOT_RESOURCE_TYPE.SYNCED,
         )
-        Config.objects.set_config("commissioning_distro_series", release)
+        Config.objects.set_config("commissioning_distro_series", series)
         form = CommissioningForm()
         self.assertCountEqual(
             [
                 ("", "--- No minimum kernel ---"),
-                (kernel, f"{release} ({kernel})"),
+                (kernel, f"{series} ({kernel})"),
             ],
             form.fields["default_min_hwe_kernel"].choices,
         )

src/maasserver/testing/factory.py

@@ -12,7 +12,7 @@
 import time
 from typing import Dict, Iterable, List
 
-from distro_info import UbuntuDistroInfo
+from distro_info import DistroRelease, UbuntuDistroInfo
 from django.conf import settings
 from django.contrib.auth.models import User
 from django.db import transaction
@@ -247,21 +247,26 @@ def pick_commissioning_release(self, osystem):
         releases = osystem.get_supported_commissioning_releases()
         return random.choice(releases)
 
-    def pick_ubuntu_release(self, but_not=None):
+    def pick_ubuntu_release(
+        self, but_not: list[str] | None = None
+    ) -> DistroRelease:
         """Pick a random supported Ubuntu release.
 
         :param but_not: Exclude these releases from the result
         :type but_not: Sequence
         """
         ubuntu_releases = UbuntuDistroInfo()
-        supported_releases = ubuntu_releases.all[
+        supported_releases = ubuntu_releases._releases[
             ubuntu_releases.all.index("precise") :
         ]
         if but_not is None:
             but_not = []
-        return random.choice(
-            [choice for choice in supported_releases if choice not in but_not]
-        )
+        choices = [
+            choice
+            for choice in supported_releases
+            if choice.series not in but_not
+        ]
+        return random.choice(choices)
 
     def make_script_content(
         self,

src/maasserver/tests/test_workers.py

@@ -7,7 +7,7 @@
 import os
 import random
 import sys
-from unittest.mock import call
+from unittest.mock import call, Mock
 
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks
@@ -156,26 +156,45 @@ def mock_unregisterWorker(*args, **kwargs):
 
     @wait_for_reactor
     @inlineCallbacks
-    def test_stopService_doesnt(self):
+    def test_stopService(self):
         set_max_workers_count(1)
-        service = WorkersService(reactor, worker_cmd="cat")
+
+        mock_worker = Mock(spec=WorkerProcess)
+        mock_worker.pid = 123
+        mock_worker.worker_id = "0"
+        mock_worker.runningImport = False
+        mock_worker.signal = Mock()
+
+        mock_reactor = Mock(reactor)
+
+        service = WorkersService(mock_reactor, worker_cmd="cat")
+        service.missing_worker_ids = ["0"]
+
+        def fake_spawnWorker(worker_id, runningImport=False):
+            service.registerWorker(mock_worker)
+
+        self.patch(service, "_spawnWorker").side_effect = fake_spawnWorker
+
+        service.startService()
+        # test that the worker was registered
+        self.assertEqual(len(service.workers), 1)
 
         dv = DeferredValue()
-        original_unregisterWorker = service.unregisterWorker
 
-        def mock_unregisterWorker(*args, **kwargs):
-            original_unregisterWorker(*args, **kwargs)
+        def fake_unregisterWorker(worker, status):
+            del service.workers[worker.pid]
+            service.missing_worker_ids.append(mock_worker.worker_id)
             dv.set(None)
 
         self.patch(service, "unregisterWorker").side_effect = (
-            mock_unregisterWorker
+            fake_unregisterWorker
         )
+        mock_worker.processEnded = Mock(side_effect=service.unregisterWorker)
 
-        try:
-            service.startService()
-            assert len(service.workers) == 1
-        finally:
-            service.stopService()
+        service.stopService()
+        service.workers[mock_worker.pid].processEnded(mock_worker, status=0)
 
         yield dv.get(timeout=2)
-        assert len(service.workers) == 0
+
+        # test that the worker was unregistered
+        self.assertEqual(len(service.workers), 0)

src/maasserver/utils/tests/test_osystems.py

@@ -438,12 +438,13 @@ def test_list_commissioning_choices_returns_current_selection(self):
     def test_make_hwe_kernel_ui_text_finds_release_from_bootsourcecache(self):
         self.useFixture(SignalsDisabled("bootsources"))
         release = factory.pick_ubuntu_release()
-        kernel = "hwe-" + release[0]
+        series = release.series
+        kernel = "hwe-" + release.version.split(" ")[0]
         factory.make_BootSourceCache(
-            os="ubuntu/%s" % release, subarch=kernel, release=release
+            os="ubuntu/%s" % series, subarch=kernel, release=series
         )
         self.assertEqual(
-            f"{release} ({kernel})", make_hwe_kernel_ui_text(kernel)
+            f"{series} ({kernel})", make_hwe_kernel_ui_text(kernel)
         )
 
     def test_make_hwe_kernel_ui_finds_release_from_ubuntudistroinfo(self):
@@ -549,6 +550,11 @@ def __init__(self, *args, **kwargs):
                 {
                     "string": "hwe-%s" % release["series"][0],
                     "expected": version_tuple + tuple([OLD_STYLE_HWE_WEIGHT]),
+                    # 'hwe-<first-letter>' is ambiguous for series >= 25
+                    # (e.g. 'p' matches precise and plucky,
+                    # 'q' matches quantal and questing),
+                    # so skip the check for those releases.
+                    "skip_if": version_tuple[0] >= 25,
                 },
             ),
             (
@@ -785,6 +791,8 @@ def __init__(self, *args, **kwargs):
         )
 
     def test_get_release_version_from_string(self):
+        if getattr(self, "skip_if", False):
+            self.skipTest("not applicable to this scenario")
         self.assertEqual(
             self.expected, get_release_version_from_string(self.string)
         )
@@ -1061,10 +1069,10 @@ def test_finds_by_series(self):
             release, get_release_from_distro_info(release["series"])
         )
 
-    def test_finds_by_series_first_letter(self):
+    def test_finds_by_series_prefix(self):
         release = self.pick_release()
         self.assertEqual(
-            release, get_release_from_distro_info(release["series"][0])
+            release, get_release_from_distro_info(release["series"][0:3])
         )
 
     def test_finds_by_version(self):

src/metadataserver/tests/test_api.py

@@ -4325,7 +4325,9 @@ def test_power_type_set_but_no_parameters(self):
     def test_power_type_set_with_parameters(self):
         # When power_type is valid, and power_parameters is valid JSON, both
         # fields are set on the node, and the node is saved.
-        power_type = factory.pick_power_type(but_not=[self.node.power_type])
+        power_type = factory.pick_power_type(
+            but_not=[self.node.power_type, "redfish"]
+        )
         power_parameters = {"foo": [1, 2, 3]}
         self.request.POST = {
             "power_type": power_type,
@@ -4337,6 +4339,25 @@ def test_power_type_set_with_parameters(self):
         self.assertTrue(self.node.bmc.created_by_commissioning)
         self.save.assert_called_once_with()
 
+    def test_power_type_set_with_parameters_redfish(self):
+        power_type = "redfish"
+        power_parameters = {
+            "node_id": 1,
+            "power_address": "10.0.0.1",
+            "power_pass": "test",
+            "power_user": "test",
+        }
+        self.request.POST = {
+            "power_type": power_type,
+            "power_parameters": json.dumps(power_parameters),
+        }
+        with post_commit_hooks:
+            store_node_power_parameters(self.node, self.request)
+        self.assertEqual(power_type, self.node.power_type)
+        self.assertEqual(power_parameters, self.node.get_power_parameters())
+        self.assertTrue(self.node.bmc.created_by_commissioning)
+        self.save.assert_called_once_with()
+
     def test_power_type_set_with_invalid_parameters(self):
         # When power_type is valid, but power_parameters is invalid JSON, the
         # node is not saved, and an exception is raised.