Merge remote-tracking branch 'origin/ISSUE-#5769' into ISSUE-#5769

cloudinit/analyze/__init__.py

@@ -5,7 +5,7 @@
 import argparse
 import re
 import sys
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import IO
 
 from cloudinit.analyze import dump, show
@@ -128,17 +128,21 @@ def analyze_boot(name, args):
     infh, outfh = configure_io(args)
     kernel_info = show.dist_check_timestamp()
     status_code, kernel_start, kernel_end, ci_sysd_start = kernel_info
-    kernel_start_timestamp = datetime.utcfromtimestamp(kernel_start)
-    kernel_end_timestamp = datetime.utcfromtimestamp(kernel_end)
-    ci_sysd_start_timestamp = datetime.utcfromtimestamp(ci_sysd_start)
+    kernel_start_timestamp = datetime.fromtimestamp(kernel_start, timezone.utc)
+    kernel_end_timestamp = datetime.fromtimestamp(kernel_end, timezone.utc)
+    ci_sysd_start_timestamp = datetime.fromtimestamp(
+        ci_sysd_start, timezone.utc
+    )
     try:
         last_init_local = [
             e
             for e in _get_events(infh)
             if e["name"] == "init-local"
             and "starting search" in e["description"]
         ][-1]
-        ci_start = datetime.utcfromtimestamp(last_init_local["timestamp"])
+        ci_start = datetime.fromtimestamp(
+            last_init_local["timestamp"], timezone.utc
+        )
     except IndexError:
         ci_start = "Could not find init-local log-line in cloud-init.log"
         status_code = show.FAIL_CODE

cloudinit/analyze/show.py

@@ -85,7 +85,9 @@ def event_timestamp(event):
 
 
 def event_datetime(event):
-    return datetime.datetime.utcfromtimestamp(event_timestamp(event))
+    return datetime.datetime.fromtimestamp(
+        event_timestamp(event), datetime.timezone.utc
+    )
 
 
 def delta_seconds(t1, t2):

cloudinit/config/schema.py

@@ -533,8 +533,7 @@ def get_jsonschema_validator():
         **validator_kwargs,
     )
 
-    # Add deprecation handling
-    def is_valid(self, instance, _schema=None, **__):
+    def is_valid_pre_4_0_0(self, instance, _schema=None, **__):
         """Override version of `is_valid`.
 
         It does ignore instances of `SchemaDeprecationError`.
@@ -547,9 +546,27 @@ def is_valid(self, instance, _schema=None, **__):
         )
         return next(errors, None) is None
 
+    def is_valid(self, instance, _schema=None, **__):
+        """Override version of `is_valid`.
+
+        It does ignore instances of `SchemaDeprecationError`.
+        """
+        errors = filter(
+            lambda e: not isinstance(  # pylint: disable=W1116
+                e, SchemaDeprecationError
+            ),
+            self.evolve(schema=_schema).iter_errors(instance),
+        )
+        return next(errors, None) is None
+
+    # Add deprecation handling
+    is_valid_fn = is_valid_pre_4_0_0
+    if hasattr(cloudinitValidator, "evolve"):
+        is_valid_fn = is_valid
+
     # this _could_ be an error, but it's not in this case
     # https://github.com/python/mypy/issues/2427#issuecomment-1419206807
-    cloudinitValidator.is_valid = is_valid  # type: ignore [method-assign]
+    cloudinitValidator.is_valid = is_valid_fn  # type: ignore [method-assign]
 
     return (cloudinitValidator, FormatChecker)
 

cloudinit/reporting/handlers.py

@@ -10,7 +10,7 @@
 import threading
 import time
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from threading import Event
 from typing import Union
 
@@ -375,7 +375,9 @@ def _encode_event(self, event):
             "name": event.name,
             "type": event.event_type,
             "ts": (
-                datetime.utcfromtimestamp(event.timestamp).isoformat() + "Z"
+                datetime.fromtimestamp(
+                    event.timestamp, timezone.utc
+                ).isoformat()
             ),
         }
         if hasattr(event, self.RESULT_KEY):

cloudinit/sources/DataSourceGCE.py

@@ -213,14 +213,14 @@ def _has_expired(public_key):
         return False
 
     expire_str = json_obj["expireOn"]
-    format_str = "%Y-%m-%dT%H:%M:%S+0000"
+    format_str = "%Y-%m-%dT%H:%M:%S%z"
     try:
         expire_time = datetime.datetime.strptime(expire_str, format_str)
     except ValueError:
         return False
 
     # Expire the key if and only if we have exceeded the expiration timestamp.
-    return datetime.datetime.utcnow() > expire_time
+    return datetime.datetime.now(datetime.timezone.utc) > expire_time
 
 
 def _parse_public_keys(public_keys_data, default_user=None):

cloudinit/sources/azure/errors.py

@@ -6,7 +6,7 @@
 import csv
 import logging
 import traceback
-from datetime import datetime
+from datetime import datetime, timezone
 from io import StringIO
 from typing import Any, Dict, List, Optional, Tuple
 from xml.etree import ElementTree as ET  # nosec B405
@@ -52,7 +52,7 @@ def __init__(
         else:
             self.supporting_data = {}
 
-        self.timestamp = datetime.utcnow()
+        self.timestamp = datetime.now(timezone.utc)
 
         try:
             self.vm_id = identity.query_vm_id()

cloudinit/sources/azure/kvp.py

@@ -3,7 +3,7 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Optional
 
 from cloudinit import version
@@ -49,7 +49,7 @@ def report_success_to_host() -> bool:
         [
             "result=success",
             f"agent=Cloud-Init/{version.version_string()}",
-            f"timestamp={datetime.utcnow().isoformat()}",
+            f"timestamp={datetime.now(timezone.utc).isoformat()}",
             f"vm_id={vm_id}",
         ]
     )

cloudinit/sources/helpers/azure.py

@@ -10,7 +10,7 @@
 import textwrap
 import zlib
 from contextlib import contextmanager
-from datetime import datetime
+from datetime import datetime, timezone
 from time import sleep, time
 from typing import Callable, List, Optional, TypeVar, Union
 from xml.etree import ElementTree as ET  # nosec B405
@@ -122,9 +122,11 @@ def get_boot_telemetry():
         "boot-telemetry",
         "kernel_start=%s user_start=%s cloudinit_activation=%s"
         % (
-            datetime.utcfromtimestamp(kernel_start).isoformat() + "Z",
-            datetime.utcfromtimestamp(user_start).isoformat() + "Z",
-            datetime.utcfromtimestamp(cloudinit_activation).isoformat() + "Z",
+            datetime.fromtimestamp(kernel_start, timezone.utc).isoformat(),
+            datetime.fromtimestamp(user_start, timezone.utc).isoformat(),
+            datetime.fromtimestamp(
+                cloudinit_activation, timezone.utc
+            ).isoformat(),
         ),
         events.DEFAULT_EVENT_ORIGIN,
     )
@@ -1011,7 +1013,7 @@ def parse_text(cls, ovf_env_xml: str) -> "OvfEnvXml":
             raise errors.ReportableErrorOvfParsingException(exception=e) from e
 
         # If there's no provisioning section, it's not Azure ovf-env.xml.
-        if not root.find("./wa:ProvisioningSection", cls.NAMESPACES):
+        if root.find("./wa:ProvisioningSection", cls.NAMESPACES) is None:
             raise NonAzureDataSource(
                 "Ignoring non-Azure ovf-env.xml: ProvisioningSection not found"
             )

cloudinit/util.py

@@ -1884,7 +1884,11 @@ def ensure_dir(path, mode=None, user=None, group=None):
         # Get non existed parent dir first before they are created.
         non_existed_parent_dir = get_non_exist_parent_dir(path)
         # Make the dir and adjust the mode
-        with SeLinuxGuard(os.path.dirname(path), recursive=True):
+        dir_name = os.path.dirname(path)
+        selinux_recursive = True
+        if dir_name == "/":
+            selinux_recursive = False
+        with SeLinuxGuard(dir_name, recursive=selinux_recursive):
             os.makedirs(path)
         chmod(path, mode)
         # Change the ownership

tests/integration_tests/clouds.py

@@ -330,7 +330,9 @@ def _perform_launch(
         except KeyError:
             profile_list = self._get_or_set_profile_list(release)
 
-        prefix = datetime.datetime.utcnow().strftime("cloudinit-%m%d-%H%M%S")
+        prefix = datetime.datetime.now(datetime.timezone.utc).strftime(
+            "cloudinit-%m%d-%H%M%S"
+        )
         default_name = prefix + "".join(
             random.choices(string.ascii_lowercase + string.digits, k=8)
         )

tests/integration_tests/test_paths.py

@@ -1,5 +1,5 @@
 import os
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Iterator
 
 import pytest
@@ -66,7 +66,11 @@ def collect_logs(self, custom_client: IntegrationInstance):
         found_logs = custom_client.execute(
             "tar -tf cloud-init.tar.gz"
         ).stdout.splitlines()
-        dirname = datetime.utcnow().date().strftime("cloud-init-logs-%Y-%m-%d")
+        dirname = (
+            datetime.now(timezone.utc)
+            .date()
+            .strftime("cloud-init-logs-%Y-%m-%d")
+        )
         expected_logs = [
             f"{dirname}/",
             f"{dirname}/dmesg.txt",
@@ -98,7 +102,11 @@ def collect_logs(self, custom_client: IntegrationInstance):
         found_logs = custom_client.execute(
             "tar -tf cloud-init.tar.gz"
         ).stdout.splitlines()
-        dirname = datetime.utcnow().date().strftime("cloud-init-logs-%Y-%m-%d")
+        dirname = (
+            datetime.now(timezone.utc)
+            .date()
+            .strftime("cloud-init-logs-%Y-%m-%d")
+        )
         assert f"{dirname}/new-cloud-dir/data/result.json" in found_logs
 
     # LXD inserts some agent setup code into VMs on Bionic under

tests/unittests/cmd/devel/test_logs.py

@@ -3,6 +3,7 @@
 import glob
 import os
 import pathlib
+import sys
 import tarfile
 from datetime import datetime, timezone
 
@@ -140,8 +141,12 @@ def test_collect_logs_end_to_end(self, mocker, tmp_path):
         )
         extract_to = tmp_path / "extracted"
         extract_to.mkdir()
+
+        tar_kwargs = {}
+        if sys.version_info > (3, 11):
+            tar_kwargs = {"filter": "fully_trusted"}
         with tarfile.open(tmp_path / "cloud-init.tar.gz") as tar:
-            tar.extractall(extract_to)
+            tar.extractall(extract_to, **tar_kwargs)  # type: ignore[arg-type]
         extracted_dir = extract_to / f"cloud-init-logs-{today}"
 
         for name in to_collect:

tests/unittests/config/test_cc_ubuntu_pro.py

@@ -1,4 +1,5 @@
 # This file is part of cloud-init. See LICENSE file for license information.
+import importlib.metadata
 import json
 import logging
 import re
@@ -447,13 +448,10 @@ class TestUbuntuProSchema:
                         " Use **ubuntu_pro** instead"
                     ),
                 ),
-                # If __version__ no longer exists on jsonschema, that means
-                # we're using a high enough version of jsonschema to not need
-                # to skip this test.
                 (
                     JSONSCHEMA_SKIP_REASON
                     if lifecycle.Version.from_str(
-                        getattr(jsonschema, "__version__", "999")
+                        importlib.metadata.version("jsonschema")
                     )
                     < lifecycle.Version(4)
                     else ""

tests/unittests/distros/package_management/test_apt.py

@@ -119,41 +119,43 @@ def test_search_stem(self, m_subp, m_which, mocker):
         )
 
 
-@mock.patch.object(
-    apt,
-    "APT_LOCK_FILES",
-    [f"{TMP_DIR}/{FILE}" for FILE in apt.APT_LOCK_FILES],
-)
-class TestUpdatePackageSources:
-    def __init__(self):
-        MockPaths = get_mock_paths(TMP_DIR)
-        self.MockPaths = MockPaths({}, FakeDataSource())
+@pytest.fixture(scope="function")
+def apt_paths(tmpdir):
+    MockPaths = get_mock_paths(str(tmpdir))
+    with mock.patch.object(
+        apt,
+        "APT_LOCK_FILES",
+        [f"{tmpdir}/{FILE}" for FILE in apt.APT_LOCK_FILES],
+    ):
+        yield MockPaths({}, FakeDataSource())
 
+
+class TestUpdatePackageSources:
     @mock.patch.object(apt.subp, "which", return_value=True)
     @mock.patch.object(apt.subp, "subp")
-    def test_force_update_calls_twice(self, m_subp, m_which):
+    def test_force_update_calls_twice(self, m_subp, m_which, apt_paths):
         """Ensure that force=true calls apt update again"""
-        instance = apt.Apt(helpers.Runners(self.MockPaths))
+        instance = apt.Apt(helpers.Runners(apt_paths))
         instance.update_package_sources()
         instance.update_package_sources(force=True)
         assert 2 == len(m_subp.call_args_list)
         TMP_DIR.cleanup()
 
     @mock.patch.object(apt.subp, "which", return_value=True)
     @mock.patch.object(apt.subp, "subp")
-    def test_force_update_twice_calls_twice(self, m_subp, m_which):
+    def test_force_update_twice_calls_twice(self, m_subp, m_which, apt_paths):
         """Ensure that force=true calls apt update again when called twice"""
-        instance = apt.Apt(helpers.Runners(self.MockPaths))
+        instance = apt.Apt(helpers.Runners(apt_paths))
         instance.update_package_sources(force=True)
         instance.update_package_sources(force=True)
         assert 2 == len(m_subp.call_args_list)
         TMP_DIR.cleanup()
 
     @mock.patch.object(apt.subp, "which", return_value=True)
     @mock.patch.object(apt.subp, "subp")
-    def test_no_force_update_calls_once(self, m_subp, m_which):
+    def test_no_force_update_calls_once(self, m_subp, m_which, apt_paths):
         """Ensure that apt-get update calls are deduped unless expected"""
-        instance = apt.Apt(helpers.Runners(self.MockPaths))
+        instance = apt.Apt(helpers.Runners(apt_paths))
         instance.update_package_sources()
         instance.update_package_sources()
         assert 1 == len(m_subp.call_args_list)

tests/unittests/sources/azure/test_errors.py

@@ -20,9 +20,9 @@ def agent_string():
 
 @pytest.fixture()
 def fake_utcnow():
-    timestamp = datetime.datetime.utcnow()
+    timestamp = datetime.datetime.now(datetime.timezone.utc)
     with mock.patch.object(errors, "datetime", autospec=True) as m:
-        m.utcnow.return_value = timestamp
+        m.now.return_value = timestamp
         yield timestamp
 
 

tests/unittests/sources/azure/test_kvp.py

@@ -1,6 +1,6 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
-from datetime import datetime
+from datetime import datetime, timezone
 from unittest import mock
 
 import pytest
@@ -11,9 +11,9 @@
 
 @pytest.fixture()
 def fake_utcnow():
-    timestamp = datetime.utcnow()
+    timestamp = datetime.now(timezone.utc)
     with mock.patch.object(kvp, "datetime", autospec=True) as m:
-        m.utcnow.return_value = timestamp
+        m.now.return_value = timestamp
         yield timestamp
 
 

tests/unittests/sources/test_azure.py

@@ -315,7 +315,7 @@ def mock_subp_subp():
 
 @pytest.fixture
 def mock_timestamp():
-    timestamp = datetime.datetime.utcnow()
+    timestamp = datetime.datetime.now(datetime.timezone.utc)
     with mock.patch.object(errors, "datetime", autospec=True) as m:
         m.utcnow.return_value = timestamp
         yield timestamp