test: add get_syslog_or_console for minimal images without syslog

Minimal images do not have rsyslog installed. As a result, no
/var/log/syslog exists. Add helper function get_syslog_or_console
to allow minimal images to use pycloudlib.Instance.get_console
instead of reading /var/log/syslog.

tests/integration_tests/modules/test_keys_to_console.py

@@ -5,10 +5,15 @@
 
 import pytest
 
+from tests.integration_tests import integration_settings
 from tests.integration_tests.decorators import retry
 from tests.integration_tests.instances import IntegrationInstance
 from tests.integration_tests.integration_settings import PLATFORM
-from tests.integration_tests.util import get_console_log
+from tests.integration_tests.util import (
+    HAS_CONSOLE_LOG,
+    get_console_log,
+    get_syslog_or_console,
+)
 
 BLACKLIST_USER_DATA = """\
 #cloud-config
@@ -38,58 +43,79 @@
 
 
 @pytest.mark.user_data(BLACKLIST_USER_DATA)
+@pytest.mark.skipif(
+    integration_settings.OS_IMAGE_TYPE == "minimal" and not HAS_CONSOLE_LOG,
+    reason=f"No console_log available for minimal images on {PLATFORM}",
+)
 class TestKeysToConsoleBlacklist:
     """Test that the blacklist options work as expected."""
 
     @pytest.mark.parametrize("key_type", ["ECDSA"])
     def test_excluded_keys(self, class_client, key_type):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "({})".format(key_type) not in syslog
+        assert "({})".format(key_type) not in get_syslog_or_console(
+            class_client
+        )
 
     # retry decorator here because it can take some time to be reflected
     # in syslog
     @retry(tries=30, delay=1)
     @pytest.mark.parametrize("key_type", ["ED25519", "RSA"])
     def test_included_keys(self, class_client, key_type):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "({})".format(key_type) in syslog
+        assert "({})".format(key_type) in get_syslog_or_console(class_client)
 
 
 @pytest.mark.user_data(BLACKLIST_ALL_KEYS_USER_DATA)
+@pytest.mark.skipif(
+    integration_settings.OS_IMAGE_TYPE == "minimal" and not HAS_CONSOLE_LOG,
+    reason=f"No console_log available for minimal images on {PLATFORM}",
+)
 class TestAllKeysToConsoleBlacklist:
     """Test that when key blacklist contains all key types that
     no header/footer are output.
     """
 
     def test_header_excluded(self, class_client):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in syslog
+        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+            class_client
+        )
 
     def test_footer_excluded(self, class_client):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "END SSH HOST KEY FINGERPRINTS" not in syslog
+        assert "END SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+            class_client
+        )
 
 
 @pytest.mark.user_data(DISABLED_USER_DATA)
+@pytest.mark.skipif(
+    integration_settings.OS_IMAGE_TYPE == "minimal" and not HAS_CONSOLE_LOG,
+    reason=f"No console_log available for minimal images on {PLATFORM}",
+)
 class TestKeysToConsoleDisabled:
     """Test that output can be fully disabled."""
 
     @pytest.mark.parametrize("key_type", ["ECDSA", "ED25519", "RSA"])
     def test_keys_excluded(self, class_client, key_type):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "({})".format(key_type) not in syslog
+        assert "({})".format(key_type) not in get_syslog_or_console(
+            class_client
+        )
 
     def test_header_excluded(self, class_client):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in syslog
+        assert "BEGIN SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+            class_client
+        )
 
     def test_footer_excluded(self, class_client):
-        syslog = class_client.read_from_file("/var/log/syslog")
-        assert "END SSH HOST KEY FINGERPRINTS" not in syslog
+        assert "END SSH HOST KEY FINGERPRINTS" not in get_syslog_or_console(
+            class_client
+        )
 
 
 @pytest.mark.user_data(ENABLE_KEYS_TO_CONSOLE_USER_DATA)
 @retry(tries=30, delay=1)
+@pytest.mark.skipif(
+    integration_settings.OS_IMAGE_TYPE == "minimal" and not HAS_CONSOLE_LOG,
+    reason=f"No console_log available for minimal images on {PLATFORM}",
+)
 @pytest.mark.skipif(
     PLATFORM not in ["ec2", "lxd_container", "oci", "openstack"],
     reason=(

tests/integration_tests/modules/test_ssh_auth_key_fingerprints.py

@@ -15,6 +15,11 @@
 
 from tests.integration_tests.decorators import retry
 from tests.integration_tests.instances import IntegrationInstance
+from tests.integration_tests.integration_settings import (
+    OS_IMAGE_TYPE,
+    PLATFORM,
+)
+from tests.integration_tests.util import HAS_CONSOLE_LOG, get_syslog_or_console
 
 USER_DATA_SSH_AUTHKEY_DISABLE = """\
 #cloud-config
@@ -45,9 +50,12 @@ def test_ssh_authkey_fingerprints_disable(self, client):
     # in syslog
     @retry(tries=30, delay=1)
     @pytest.mark.user_data(USER_DATA_SSH_AUTHKEY_ENABLE)
+    @pytest.mark.skipif(
+        OS_IMAGE_TYPE == "minimal" and not HAS_CONSOLE_LOG,
+        reason=f"No console_log available for minimal images on {PLATFORM}",
+    )
     def test_ssh_authkey_fingerprints_enable(self, client):
-        syslog_output = client.read_from_file("/var/log/syslog")
-
+        syslog_output = get_syslog_or_console(client)
         assert re.search(r"256 SHA256:.*(ECDSA)", syslog_output) is not None
         assert re.search(r"256 SHA256:.*(ED25519)", syslog_output) is not None
         assert re.search(r"2048 SHA256:.*(RSA)", syslog_output) is None

tests/integration_tests/modules/test_write_files.py

@@ -69,18 +69,18 @@ class TestWriteFiles:
     @pytest.mark.parametrize(
         "cmd,expected_out",
         (
-            ("file /root/file_b64", ASCII_TEXT),
+            ("md5sum </root/file_b64", "84baab0d01c1374924dcedfb5972697c"),
             ("md5sum </root/file_binary", "3801184b97bb8c6e63fa0e1eae2920d7"),
             (
                 "sha256sum </root/file_binary",
                 "2c791c4037ea5bd7e928d6a87380f8ba"
                 "7a803cd83d5e4f269e28f5090f0f2c9a",
             ),
             (
-                "file /root/file_gzip",
-                "POSIX shell script, ASCII text executable",
+                "md5sum </root/file_gzip",
+                "ec96d4a61ed762f0ff3725e1140661de",
             ),
-            ("file /root/file_text", ASCII_TEXT),
+            ("md5sum </root/file_text", "a2b6d22fa3d7aa551e22bb0c8acd9121"),
         ),
     )
     def test_write_files(self, cmd, expected_out, class_client):

tests/integration_tests/util.py

@@ -14,7 +14,11 @@
 import pytest
 
 from cloudinit.subp import subp
-from tests.integration_tests.integration_settings import PLATFORM
+from tests.integration_tests.decorators import retry
+from tests.integration_tests.integration_settings import (
+    OS_IMAGE_TYPE,
+    PLATFORM,
+)
 from tests.integration_tests.releases import CURRENT_RELEASE, NOBLE
 
 LOG = logging.getLogger("integration_testing.util")
@@ -29,6 +33,14 @@
 ASSETS_DIR = Path("tests/integration_tests/assets")
 KEY_PATH = ASSETS_DIR / "keys"
 
+HAS_CONSOLE_LOG = PLATFORM in (
+    "ec2",
+    "lxd_container",
+    "oci",
+    "openstack",
+    "qemu",
+)
+
 
 def verify_ordered_items_in_text(to_verify: list, text: str):
     """Assert all items in list appear in order in text.
@@ -546,6 +558,15 @@ def get_console_log(client: "IntegrationInstance"):
     return console_log
 
 
+@retry(tries=5, delay=1)  # Retry on get_console_log failures
+def get_syslog_or_console(client: "IntegrationInstance") -> str:
+    """minimal OS_IMAGE_TYPE does not contain rsyslog"""
+    if OS_IMAGE_TYPE == "minimal" and HAS_CONSOLE_LOG:
+        return get_console_log(client)
+    else:
+        return client.read_from_file("/var/log/syslog")
+
+
 @lru_cache()
 def lxd_has_nocloud(client: "IntegrationInstance") -> bool:
     # Bionic or Focal may be detected as NoCloud rather than LXD