testing: port remaining cloud tests to integration testing framework (SC-191)

tests/integration_tests/modules/test_combined.py

@@ -0,0 +1,175 @@
+# 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 pytest
+import re
+from datetime import date
+
+from tests.integration_tests.clouds import ImageSpecification
+from tests.integration_tests.instances import IntegrationInstance
+from tests.integration_tests.util import 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']
+"""
+
+
+@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')
+        today = date.today().strftime('%a, %d %b %Y')
+        expected = (
+            'This is my final message!\n'
+            r'\d+\.\d+.*\n'
+            '{}.*\n'
+            'DataSource.*\n'
+            r'\d+\.\d+'
+        ).format(today)
+
+        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_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')
+        assert 'WARN' not in log
+        assert 'Traceback' not in log
+
+    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['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['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.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'
+        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

tests/integration_tests/modules/test_command_output.py

@@ -0,0 +1,23 @@
+"""Integration test for output redirection.
+
+This test redirects the output of a command to a file and then checks the file.
+
+(This is ported from
+``tests/cloud_tests/testcases/main/command_output_simple.yaml``.)"""
+import pytest
+
+from tests.integration_tests.instances import IntegrationInstance
+
+
+USER_DATA = """\
+#cloud-config
+output: { all: "| tee -a /var/log/cloud-init-test-output" }
+final_message: "should be last line in cloud-init-test-output file"
+"""
+
+
+@pytest.mark.ci
+@pytest.mark.user_data(USER_DATA)
+def test_runcmd(client: IntegrationInstance):
+    log = client.read_from_file('/var/log/cloud-init-test-output')
+    assert 'should be last line in cloud-init-test-output file' in log

tests/integration_tests/modules/test_ntp_servers.py

@@ -1,37 +1,46 @@
-"""Integration test for the ntp module's ``servers`` functionality with ntp.
+"""Integration test for the ntp module's ntp functionality.
 
 This test specifies the use of the `ntp` NTP client, and ensures that the given
 NTP servers are configured as expected.
 
-(This is ported from ``tests/cloud_tests/testcases/modules/ntp_servers.yaml``.)
+(This is ported from ``tests/cloud_tests/testcases/modules/ntp_servers.yaml``,
+``tests/cloud_tests/testcases/modules/ntp_pools.yaml``,
+and ``tests/cloud_tests/testcases/modules/ntp_chrony.yaml``)
 """
 import re
 
 import yaml
 import pytest
 
+from tests.integration_tests.instances import IntegrationInstance
+
 USER_DATA = """\
 #cloud-config
 ntp:
   ntp_client: ntp
   servers:
       - 172.16.15.14
       - 172.16.17.18
+  pools:
+      - 0.cloud-init.mypool
+      - 1.cloud-init.mypool
+      - 172.16.15.15
 """
 
 EXPECTED_SERVERS = yaml.safe_load(USER_DATA)["ntp"]["servers"]
+EXPECTED_POOLS = yaml.safe_load(USER_DATA)["ntp"]["pools"]
 
 
 @pytest.mark.ci
 @pytest.mark.user_data(USER_DATA)
 class TestNtpServers:
 
-    def test_ntp_installed(self, class_client):
+    def test_ntp_installed(self, class_client: IntegrationInstance):
         """Test that `ntpd --version` succeeds, indicating installation."""
-        result = class_client.execute("ntpd --version")
-        assert 0 == result.return_code
+        assert class_client.execute("ntpd --version").ok
 
-    def test_dist_config_file_is_empty(self, class_client):
+    def test_dist_config_file_is_empty(self,
+                                       class_client: IntegrationInstance):
         """Test that the distributed config file is empty.
 
         (This test is skipped on all currently supported Ubuntu releases, so
@@ -42,17 +51,77 @@ def test_dist_config_file_is_empty(self, class_client):
         dist_file = class_client.read_from_file("/etc/ntp.conf.dist")
         assert 0 == len(dist_file.strip().splitlines())
 
-    def test_ntp_entries(self, class_client):
+    def test_ntp_entries(self, class_client: IntegrationInstance):
         ntp_conf = class_client.read_from_file("/etc/ntp.conf")
         for expected_server in EXPECTED_SERVERS:
             assert re.search(
                 r"^server {} iburst".format(expected_server),
                 ntp_conf,
                 re.MULTILINE
             )
+        for expected_pool in EXPECTED_POOLS:
+            assert re.search(
+                r"^pool {} iburst".format(expected_pool),
+                ntp_conf,
+                re.MULTILINE
+            )
 
-    def test_ntpq_servers(self, class_client):
+    def test_ntpq_servers(self, class_client: IntegrationInstance):
         result = class_client.execute("ntpq -p -w -n")
         assert result.ok
-        for expected_server in EXPECTED_SERVERS:
-            assert expected_server in result.stdout
+        for expected_server_or_pool in [*EXPECTED_SERVERS, *EXPECTED_POOLS]:
+            assert expected_server_or_pool in result.stdout
+
+
+CHRONY_DATA = """\
+#cloud-config
+ntp:
+  enabled: true
+  ntp_client: chrony
+"""
+
+
+@pytest.mark.ci
+@pytest.mark.user_data(CHRONY_DATA)
+def test_chrony(client: IntegrationInstance):
+    if client.execute('test -f /etc/chrony.conf').ok:
+        chrony_conf = '/etc/chrony.conf'
+    else:
+        chrony_conf = '/etc/chrony/chrony.conf'
+    contents = client.read_from_file(chrony_conf)
+    assert '.pool.ntp.org' in contents
+
+
+TIMESYNCD_DATA = """\
+#cloud-config
+ntp:
+  enabled: true
+  ntp_client: systemd-timesyncd
+"""
+
+
+@pytest.mark.ci
+@pytest.mark.user_data(TIMESYNCD_DATA)
+def test_timesyncd(client: IntegrationInstance):
+    contents = client.read_from_file(
+        '/etc/systemd/timesyncd.conf.d/cloud-init.conf'
+    )
+    assert '.pool.ntp.org' in contents
+
+
+EMPTY_NTP = """\
+#cloud-config
+ntp:
+  ntp_client: ntp
+  pools: []
+  servers: []
+"""
+
+
+@pytest.mark.user_data(EMPTY_NTP)
+def test_empty_ntp(client: IntegrationInstance):
+    assert client.execute('ntpd --version').ok
+    assert client.execute('test -f /etc/ntp.conf.dist').failed
+    assert 'pool.ntp.org iburst' in client.execute(
+        'grep -v "^#" /etc/ntp.conf'
+    )

tests/integration_tests/modules/test_snap.py

@@ -4,7 +4,7 @@
 and then checks that if that command was executed during boot.
 
 (This is ported from
-``tests/cloud_tests/testcases/modules/runcmd.yaml``.)"""
+``tests/cloud_tests/testcases/modules/snap.yaml``.)"""
 
 import pytest