T2117: Cloud-init updated to 22.1

Merged with 22.1 tag from the upstream Cloud-init repository.

Our modules were slightly modified for compatibility with the new
version.

1 files changed, 342 insertions, 0 deletions

diff --git a/tests/integration_tests/modules/test_combined.py b/tests/integration_tests/modules/test_combined.py
new file mode 100644
index 00000000..7a9a6e27
--- /dev/null
+++ b/tests/integration_tests/modules/test_combined.py
@@ -0,0 +1,342 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+"""A set of somewhat unrelated tests that can be combined into a single
+instance launch. Generally tests should only be added here if a failure
+of the test would be unlikely to affect the running of another test using
+the same instance launch. Most independent module coherence tests can go
+here.
+"""
+import json
+import re
+
+import pytest
+
+from tests.integration_tests.clouds import ImageSpecification
+from tests.integration_tests.instances import IntegrationInstance
+from tests.integration_tests.util import (
+    retry,
+    verify_clean_log,
+    verify_ordered_items_in_text,
+)
+
+USER_DATA = """\
+#cloud-config
+apt:
+  primary:
+    - arches: [default]
+      uri: http://us.archive.ubuntu.com/ubuntu/
+byobu_by_default: enable
+final_message: |
+  This is my final message!
+  $version
+  $timestamp
+  $datasource
+  $uptime
+locale: en_GB.UTF-8
+locale_configfile: /etc/default/locale
+ntp:
+  servers: ['ntp.ubuntu.com']
+package_update: true
+random_seed:
+  data: 'MYUb34023nD:LFDK10913jk;dfnk:Df'
+  encoding: raw
+  file: /root/seed
+rsyslog:
+  configs:
+    - "*.* @@127.0.0.1"
+    - filename: 0-basic-config.conf
+      content: |
+        module(load="imtcp")
+        input(type="imtcp" port="514")
+        $template RemoteLogs,"/var/tmp/rsyslog.log"
+        *.* ?RemoteLogs
+        & ~
+  remotes:
+    me: "127.0.0.1"
+runcmd:
+  - echo 'hello world' > /var/tmp/runcmd_output
+
+  - #
+  - logger "My test log"
+snap:
+  squashfuse_in_container: true
+  commands:
+    - snap install hello-world
+ssh_import_id:
+  - gh:powersj
+  - lp:smoser
+timezone: US/Aleutian
+"""
+
+
+@pytest.mark.ci
+@pytest.mark.user_data(USER_DATA)
+class TestCombined:
+    def test_final_message(self, class_client: IntegrationInstance):
+        """Test that final_message module works as expected.
+
+        Also tests LP 1511485: final_message is silent.
+        """
+        client = class_client
+        log = client.read_from_file("/var/log/cloud-init.log")
+        expected = (
+            "This is my final message!\n"
+            r"\d+\.\d+.*\n"
+            r"\w{3}, \d{2} \w{3} \d{4} \d{2}:\d{2}:\d{2} \+\d{4}\n"  # Datetime
+            "DataSource.*\n"
+            r"\d+\.\d+"
+        )
+
+        assert re.search(expected, log)
+
+    def test_ntp_with_apt(self, class_client: IntegrationInstance):
+        """LP #1628337.
+
+        cloud-init tries to install NTP before even
+        configuring the archives.
+        """
+        client = class_client
+        log = client.read_from_file("/var/log/cloud-init.log")
+        assert "W: Failed to fetch" not in log
+        assert "W: Some index files failed to download" not in log
+        assert "E: Unable to locate package ntp" not in log
+
+    def test_byobu(self, class_client: IntegrationInstance):
+        """Test byobu configured as enabled by default."""
+        client = class_client
+        assert client.execute('test -e "/etc/byobu/autolaunch"').ok
+
+    def test_configured_locale(self, class_client: IntegrationInstance):
+        """Test locale can be configured correctly."""
+        client = class_client
+        default_locale = client.read_from_file("/etc/default/locale")
+        assert "LANG=en_GB.UTF-8" in default_locale
+
+        locale_a = client.execute("locale -a")
+        verify_ordered_items_in_text(["en_GB.utf8", "en_US.utf8"], locale_a)
+
+        locale_gen = client.execute(
+            "cat /etc/locale.gen | grep -v '^#' | uniq"
+        )
+        verify_ordered_items_in_text(
+            ["en_GB.UTF-8", "en_US.UTF-8"], locale_gen
+        )
+
+    def test_random_seed_data(self, class_client: IntegrationInstance):
+        """Integration test for the random seed module.
+
+        This test specifies a command to be executed by the ``seed_random``
+        module, by providing a different data to be used as seed data. We will
+        then check if that seed data was actually used.
+        """
+        client = class_client
+
+        # Only read the first 31 characters, because the rest could be
+        # binary data
+        result = client.execute("head -c 31 < /root/seed")
+        assert result.startswith("MYUb34023nD:LFDK10913jk;dfnk:Df")
+
+    def test_rsyslog(self, class_client: IntegrationInstance):
+        """Test rsyslog is configured correctly."""
+        client = class_client
+        assert "My test log" in client.read_from_file("/var/tmp/rsyslog.log")
+
+    def test_runcmd(self, class_client: IntegrationInstance):
+        """Test runcmd works as expected"""
+        client = class_client
+        assert "hello world" == client.read_from_file("/var/tmp/runcmd_output")
+
+    @retry(tries=30, delay=1)
+    def test_ssh_import_id(self, class_client: IntegrationInstance):
+        """Integration test for the ssh_import_id module.
+
+        This test specifies ssh keys to be imported by the ``ssh_import_id``
+        module and then checks that if the ssh keys were successfully imported.
+
+        TODO:
+        * This test assumes that SSH keys will be imported into the
+        /home/ubuntu; this will need modification to run on other OSes.
+        """
+        client = class_client
+        ssh_output = client.read_from_file("/home/ubuntu/.ssh/authorized_keys")
+
+        assert "# ssh-import-id gh:powersj" in ssh_output
+        assert "# ssh-import-id lp:smoser" in ssh_output
+
+    def test_snap(self, class_client: IntegrationInstance):
+        """Integration test for the snap module.
+
+        This test specifies a command to be executed by the ``snap`` module
+        and then checks that if that command was executed during boot.
+        """
+        client = class_client
+        snap_output = client.execute("snap list")
+        assert "core " in snap_output
+        assert "hello-world " in snap_output
+
+    def test_timezone(self, class_client: IntegrationInstance):
+        """Integration test for the timezone module.
+
+        This test specifies a timezone to be used by the ``timezone`` module
+        and then checks that if that timezone was respected during boot.
+        """
+        client = class_client
+        timezone_output = client.execute(
+            'date "+%Z" --date="Thu, 03 Nov 2016 00:47:00 -0400"'
+        )
+        assert timezone_output.strip() == "HDT"
+
+    def test_no_problems(self, class_client: IntegrationInstance):
+        """Test no errors, warnings, or tracebacks"""
+        client = class_client
+        status_file = client.read_from_file("/run/cloud-init/status.json")
+        status_json = json.loads(status_file)["v1"]
+        for stage in ("init", "init-local", "modules-config", "modules-final"):
+            assert status_json[stage]["errors"] == []
+        result_file = client.read_from_file("/run/cloud-init/result.json")
+        result_json = json.loads(result_file)["v1"]
+        assert result_json["errors"] == []
+
+        log = client.read_from_file("/var/log/cloud-init.log")
+        verify_clean_log(log)
+
+    def test_correct_datasource_detected(
+        self, class_client: IntegrationInstance
+    ):
+        """Test datasource is detected at the proper boot stage."""
+        client = class_client
+        status_file = client.read_from_file("/run/cloud-init/status.json")
+        parsed_datasource = json.loads(status_file)["v1"]["datasource"]
+
+        if client.settings.PLATFORM in ["lxd_container", "lxd_vm"]:
+            assert parsed_datasource.startswith("DataSourceNoCloud")
+        else:
+            platform_datasources = {
+                "azure": "DataSourceAzure [seed=/dev/sr0]",
+                "ec2": "DataSourceEc2Local",
+                "gce": "DataSourceGCELocal",
+                "oci": "DataSourceOracle",
+                "openstack": "DataSourceOpenStackLocal [net,ver=2]",
+            }
+            assert (
+                platform_datasources[client.settings.PLATFORM]
+                == parsed_datasource
+            )
+
+    def test_cloud_id_file_symlink(self, class_client: IntegrationInstance):
+        cloud_id = class_client.execute("cloud-id").stdout
+        expected_link_output = (
+            "'/run/cloud-init/cloud-id' -> "
+            f"'/run/cloud-init/cloud-id-{cloud_id}'"
+        )
+        assert expected_link_output == str(
+            class_client.execute("stat -c %N /run/cloud-init/cloud-id")
+        )
+
+    def _check_common_metadata(self, data):
+        assert data["base64_encoded_keys"] == []
+        assert data["merged_cfg"] == "redacted for non-root user"
+
+        image_spec = ImageSpecification.from_os_image()
+        assert data["sys_info"]["dist"][0] == image_spec.os
+
+        v1_data = data["v1"]
+        assert re.match(r"\d\.\d+\.\d+-\d+", v1_data["kernel_release"])
+        assert v1_data["variant"] == image_spec.os
+        assert v1_data["distro"] == image_spec.os
+        assert v1_data["distro_release"] == image_spec.release
+        assert v1_data["machine"] == "x86_64"
+        assert re.match(r"3.\d\.\d", v1_data["python_version"])
+
+    @pytest.mark.lxd_container
+    def test_instance_json_lxd(self, class_client: IntegrationInstance):
+        client = class_client
+        instance_json_file = client.read_from_file(
+            "/run/cloud-init/instance-data.json"
+        )
+
+        data = json.loads(instance_json_file)
+        self._check_common_metadata(data)
+        v1_data = data["v1"]
+        assert v1_data["cloud_name"] == "unknown"
+        assert v1_data["platform"] == "lxd"
+        assert v1_data["cloud_id"] == "lxd"
+        assert f"{v1_data['cloud_id']}" == client.read_from_file(
+            "/run/cloud-init/cloud-id-lxd"
+        )
+        assert (
+            v1_data["subplatform"]
+            == "seed-dir (/var/lib/cloud/seed/nocloud-net)"
+        )
+        assert v1_data["availability_zone"] is None
+        assert v1_data["instance_id"] == client.instance.name
+        assert v1_data["local_hostname"] == client.instance.name
+        assert v1_data["region"] is None
+
+    @pytest.mark.lxd_vm
+    def test_instance_json_lxd_vm(self, class_client: IntegrationInstance):
+        client = class_client
+        instance_json_file = client.read_from_file(
+            "/run/cloud-init/instance-data.json"
+        )
+
+        data = json.loads(instance_json_file)
+        self._check_common_metadata(data)
+        v1_data = data["v1"]
+        assert v1_data["cloud_name"] == "unknown"
+        assert v1_data["platform"] == "lxd"
+        assert v1_data["cloud_id"] == "lxd"
+        assert f"{v1_data['cloud_id']}" == client.read_from_file(
+            "/run/cloud-init/cloud-id-lxd"
+        )
+        assert any(
+            [
+                "/var/lib/cloud/seed/nocloud-net" in v1_data["subplatform"],
+                "/dev/sr0" in v1_data["subplatform"],
+            ]
+        )
+        assert v1_data["availability_zone"] is None
+        assert v1_data["instance_id"] == client.instance.name
+        assert v1_data["local_hostname"] == client.instance.name
+        assert v1_data["region"] is None
+
+    @pytest.mark.ec2
+    def test_instance_json_ec2(self, class_client: IntegrationInstance):
+        client = class_client
+        instance_json_file = client.read_from_file(
+            "/run/cloud-init/instance-data.json"
+        )
+        data = json.loads(instance_json_file)
+        v1_data = data["v1"]
+        assert v1_data["cloud_name"] == "aws"
+        assert v1_data["platform"] == "ec2"
+        # Different regions will show up as ec2-(gov|china)
+        assert v1_data["cloud_id"].startswith("ec2")
+        assert f"{v1_data['cloud_id']}" == client.read_from_file(
+            "/run/cloud-init/cloud-id-ec2"
+        )
+        assert v1_data["subplatform"].startswith("metadata")
+        assert (
+            v1_data["availability_zone"] == client.instance.availability_zone
+        )
+        assert v1_data["instance_id"] == client.instance.name
+        assert v1_data["local_hostname"].startswith("ip-")
+        assert v1_data["region"] == client.cloud.cloud_instance.region
+
+    @pytest.mark.gce
+    def test_instance_json_gce(self, class_client: IntegrationInstance):
+        client = class_client
+        instance_json_file = client.read_from_file(
+            "/run/cloud-init/instance-data.json"
+        )
+        data = json.loads(instance_json_file)
+        self._check_common_metadata(data)
+        v1_data = data["v1"]
+        assert v1_data["cloud_name"] == "gce"
+        assert v1_data["platform"] == "gce"
+        assert f"{v1_data['cloud_id']}" == client.read_from_file(
+            "/run/cloud-init/cloud-id-gce"
+        )
+        assert v1_data["subplatform"].startswith("metadata")
+        assert v1_data["availability_zone"] == client.instance.zone
+        assert v1_data["instance_id"] == client.instance.instance_id
+        assert v1_data["local_hostname"] == client.instance.name