Ubuntu/devel

cloudinit/sources/__init__.py

@@ -1177,9 +1177,9 @@ def parse_cmdline() -> str:
     Passing by command line overrides runtime datasource detection
     """
     cmdline = util.get_cmdline()
-    ds_parse_0 = re.search(r"ds=([a-zA-Z]+)(\s|$|;)", cmdline)
-    ds_parse_1 = re.search(r"ci\.ds=([a-zA-Z]+)(\s|$|;)", cmdline)
-    ds_parse_2 = re.search(r"ci\.datasource=([a-zA-Z]+)(\s|$|;)", cmdline)
+    ds_parse_0 = re.search(r"ds=([^\s;]+)", cmdline)
+    ds_parse_1 = re.search(r"ci\.ds=([^\s;]+)", cmdline)
+    ds_parse_2 = re.search(r"ci\.datasource=([^\s;]+)", cmdline)
     ds = ds_parse_0 or ds_parse_1 or ds_parse_2
     deprecated = ds_parse_1 or ds_parse_2
     if deprecated:

debian/changelog

@@ -1,3 +1,9 @@
+cloud-init (23.2-0ubuntu2) mantic; urgency=medium
+
+  * yep 
+
+ -- Chad Smith <[email protected]>  Tue, 27 Jun 2023 06:23:45 -0600
+
 cloud-init (23.2-0ubuntu1) mantic; urgency=medium
 
   * d/control: Remove pep8 dependency. It is no longer used.

tests/data/kernel_cmdline_match/meta-data

@@ -0,0 +1 @@
+instance=id: dfa16c21-4b25-4767-b94c-f98f5d73736d

tests/data/kernel_cmdline_match/user-data

@@ -0,0 +1,3 @@
+#cloud-config
+runcmd:
+- touch /cmdline-userdata-success

tests/data/kernel_cmdline_match/vendor-data

@@ -0,0 +1,2 @@
+#cloud-config
+{}

tests/integration_tests/cmd/test_status.py

@@ -1,33 +1,11 @@
 """Tests for `cloud-init status`"""
-from time import sleep
-
 import pytest
 
 from tests.integration_tests.clouds import IntegrationCloud
 from tests.integration_tests.instances import IntegrationInstance
 from tests.integration_tests.integration_settings import PLATFORM
 from tests.integration_tests.releases import CURRENT_RELEASE, IS_UBUNTU, JAMMY
-
-
-# We're implementing our own here in case cloud-init status --wait
-# isn't working correctly (LP: #1966085)
-def _wait_for_cloud_init(client: IntegrationInstance):
-    last_exception = None
-    for _ in range(30):
-        try:
-            result = client.execute("cloud-init status")
-            if (
-                result
-                and result.ok
-                and ("running" not in result or "not run" not in result)
-            ):
-                return result
-        except Exception as e:
-            last_exception = e
-        sleep(1)
-    raise Exception(
-        "cloud-init status did not return successfully."
-    ) from last_exception
+from tests.integration_tests.util import wait_for_cloud_init
 
 
 def _remove_nocloud_dir_and_reboot(client: IntegrationInstance):
@@ -66,6 +44,7 @@ def test_wait_when_no_datasource(session_cloud: IntegrationCloud, setup_image):
         # Jammy and above will use LXD datasource by default
         if CURRENT_RELEASE < JAMMY:
             _remove_nocloud_dir_and_reboot(client)
-        status_out = _wait_for_cloud_init(client).stdout.strip()
+        status_out = wait_for_cloud_init(client).stdout.strip()
         assert "status: disabled" in status_out
         assert client.execute("cloud-init status --wait").ok
+        assert client.execute("test -f /cmdline-userdata-success").ok

tests/integration_tests/test_kernel_commandline_match.py

@@ -1,7 +1,9 @@
 import pytest
 
+from tests.integration_tests.clouds import IntegrationCloud
 from tests.integration_tests.instances import IntegrationInstance
 from tests.integration_tests.integration_settings import PLATFORM
+from tests.integration_tests.util import wait_for_cloud_init
 
 
 def override_kernel_cmdline(ds_str: str, c: IntegrationInstance) -> str:
@@ -75,18 +77,44 @@ def test_lxd_datasource_kernel_override(
     ) in override_kernel_cmdline(ds_str, client)
 
 
+GH_REPO_PATH = "https://raw.githubusercontent.com/canonical/cloud-init/main/"
+
+
 @pytest.mark.skipif(PLATFORM != "lxd_vm", reason="Modifies grub config")
 @pytest.mark.lxd_use_exec
-@pytest.mark.parametrize("ds_str", ("ci.ds=nocloud-net",))
+@pytest.mark.parametrize(
+    "ds_str",
+    (f"ds=nocloud-net;s={GH_REPO_PATH}tests/data/kernel_cmdline_match/",),
+)
 def test_lxd_datasource_kernel_override_nocloud_net(
-    ds_str, client: IntegrationInstance
+    ds_str, session_cloud: IntegrationCloud, setup_image
 ):
     """NoCloud requirements vary slightly from other datasources with parsing
     nocloud-net due to historical reasons. Do to this variation, this is
     implemented in NoCloud's ds_detect and therefore has a different log
     message.
     """
-
-    assert (
-        "Machine is running on DataSourceNoCloud [seed=None][dsmode=net]."
-    ) in override_kernel_cmdline(ds_str, client)
+    _ds_name, _, seed_url = ds_str.partition(";")
+    _key, _, url_val = seed_url.partition("=")
+    with session_cloud.launch(
+        launch_kwargs={
+            # On Jammy and above, we detect the LXD datasource using a
+            # socket available to the container. This prevents the socket
+            # from being exposed in the container, so LXD will not be detected.
+            # This allows us to wait for detection in 'init' stage with
+            # DataSourceNoCloudNet.
+            "config_dict": {"security.devlxd": False},
+            "wait": False,  # to prevent cloud-init status --wait
+        }
+    ) as client:
+        # We know this will be an LXD instance due to our pytest mark
+        client.instance.execute_via_ssh = False  # pyright: ignore
+        assert wait_for_cloud_init(client).ok
+        assert (
+            "Machine is running on DataSourceNoCloud [seed=None][dsmode=net]."
+        ) in override_kernel_cmdline(ds_str, client)
+        assert (
+            "nocloud"
+            == client.execute("cloud-init query platform").stdout.strip()
+        )
+        assert url_val in client.execute("cloud-init query subplatform").stdout

tests/integration_tests/util.py

@@ -155,6 +155,27 @@ def get_test_rsa_keypair(key_name: str = "test1") -> key_pair:
     return key_pair(public_key, private_key)
 
 
+# We're implementing our own here in case cloud-init status --wait
+# isn't working correctly (LP: #1966085)
+def wait_for_cloud_init(client: IntegrationInstance):
+    last_exception = None
+    for _ in range(30):
+        try:
+            result = client.execute("cloud-init status")
+            if (
+                result
+                and result.ok
+                and ("running" not in result or "not run" not in result)
+            ):
+                return result
+        except Exception as e:
+            last_exception = e
+        time.sleep(1)
+    raise Exception(
+        "cloud-init status did not return successfully."
+    ) from last_exception
+
+
 def get_console_log(client: IntegrationInstance):
     try:
         console_log = client.instance.console_log()

tests/unittests/sources/test___init__.py

@@ -4,42 +4,44 @@
 from cloudinit.sources import DataSourceOpenStack as ds
 from tests.unittests.helpers import mock
 
+openstack_ds_name = ds.DataSourceOpenStack.dsname.lower()
+
 
 @pytest.mark.parametrize(
-    "m_cmdline",
+    "m_cmdline, expected_ds",
     (
         # test ci.ds=
-        "aosiejfoij ci.ds=OpenStack ",
-        "ci.ds=OpenStack",
-        "aosiejfoij ci.ds=OpenStack blah",
-        "aosiejfoij ci.ds=OpenStack faljskebflk",
-        "ci.ds=OpenStack;",
-        "ci.ds=openstack;",
+        ("aosiejfoij ci.ds=OpenStack ", openstack_ds_name),
+        ("ci.ds=OpenStack", openstack_ds_name),
+        ("aosiejfoij ci.ds=OpenStack blah", openstack_ds_name),
+        ("aosiejfoij ci.ds=OpenStack faljskebflk", openstack_ds_name),
+        ("ci.ds=OpenStack;", openstack_ds_name),
+        ("ci.ds=openstack;", openstack_ds_name),
         # test ci.datasource=
-        "aosiejfoij ci.datasource=OpenStack ",
-        "ci.datasource=OpenStack",
-        "aosiejfoij ci.datasource=OpenStack blah",
-        "aosiejfoij ci.datasource=OpenStack faljskebflk",
-        "ci.datasource=OpenStack;",
-        "ci.datasource=openstack;",
+        ("aosiejfoij ci.datasource=OpenStack ", openstack_ds_name),
+        ("ci.datasource=OpenStack", openstack_ds_name),
+        ("aosiejfoij ci.datasource=OpenStack blah", openstack_ds_name),
+        ("aosiejfoij ci.datasource=OpenStack faljskebflk", openstack_ds_name),
+        ("ci.datasource=OpenStack;", openstack_ds_name),
+        ("ci.datasource=openstack;", openstack_ds_name),
         # weird whitespace
-        "ci.datasource=OpenStack\n",
-        "ci.datasource=OpenStack\t",
-        "ci.datasource=OpenStack\r",
-        "ci.datasource=OpenStack\v",
-        "ci.ds=OpenStack\n",
-        "ci.ds=OpenStack\t",
-        "ci.ds=OpenStack\r",
-        "ci.ds=OpenStack\v",
+        ("ci.datasource=OpenStack\n", openstack_ds_name),
+        ("ci.datasource=OpenStack\t", openstack_ds_name),
+        ("ci.datasource=OpenStack\r", openstack_ds_name),
+        ("ci.datasource=OpenStack\v", openstack_ds_name),
+        ("ci.ds=OpenStack\n", openstack_ds_name),
+        ("ci.ds=OpenStack\t", openstack_ds_name),
+        ("ci.ds=OpenStack\r", openstack_ds_name),
+        ("ci.ds=OpenStack\v", openstack_ds_name),
+        ("ci.ds=nocloud-net\v", "nocloud-net"),
     ),
 )
-def test_ds_detect_kernel_commandline(m_cmdline):
+def test_ds_detect_kernel_commandline(m_cmdline, expected_ds):
     """check commandline match"""
     with mock.patch(
         "cloudinit.util.get_cmdline",
         return_value=m_cmdline,
     ):
         assert (
-            ds.DataSourceOpenStack.dsname.lower()
-            == sources.parse_cmdline().lower()
+            expected_ds == sources.parse_cmdline().lower()
         ), f"could not parse [{m_cmdline}]"