Update purge suffix so that resources are preserved for at least as long as the configured timeout. (#2265)

## Changes

This PR updates the purge suffix used by integration tests so that
resources are marked for preservation _at least_ as long as the
configured timeout. The situation is that:

- Many resources can be marked with a tag (or a special suffix in the
name) to ensure they are preserved until a specific time.
- The timestamp refers to a UTC-based hour, such as: `2024072915`. This
means that resources can be deleted at any moment after 3 p.m. (UTC).
- Prior to this change, resources created as late as 2:44:59 p.m. would
be tagged for deletion with `2024072915` meaning deletion as of 3 p.m.
This is 0:15 later instead of the intended 1:15.

The default timeout for tests was 1:15 in an attempt to deal with
resources being deleted when the test started near the hour boundary.
This PR also updates the default timeout to 1:00 because the extra 0:15
isn't needed to ensure this.

Author: asnare

src/databricks/labs/ucx/installer/workflows.py

@@ -7,7 +7,7 @@
 import webbrowser
 from collections.abc import Iterator
 from dataclasses import replace
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from io import StringIO
 from pathlib import Path
 from typing import Any
@@ -53,8 +53,8 @@
 
 logger = logging.getLogger(__name__)
 
-TEST_JOBS_PURGE_TIMEOUT = timedelta(hours=1, minutes=15)
-TEST_NIGHTLY_CI_JOBS_PURGE_TIMEOUT = timedelta(hours=3)  # Buffer for debugging nightly integration test runs
+TEST_RESOURCE_PURGE_TIMEOUT = timedelta(hours=1)
+TEST_NIGHTLY_CI_RESOURCES_PURGE_TIMEOUT = timedelta(hours=3)  # Buffer for debugging nightly integration test runs
 EXTRA_TASK_PARAMS = {
     "job_id": "{{job_id}}",
     "run_id": "{{run_id}}",
@@ -462,9 +462,15 @@ def _is_nightly():
         ci_env = os.getenv("TEST_NIGHTLY")
         return ci_env is not None and ci_env.lower() == "true"
 
-    def _get_test_purge_time(self) -> str:
-        timeout = TEST_NIGHTLY_CI_JOBS_PURGE_TIMEOUT if self._is_nightly() else TEST_JOBS_PURGE_TIMEOUT
-        return (datetime.utcnow() + timeout).strftime("%Y%m%d%H")
+    @classmethod
+    def _get_test_purge_time(cls) -> str:
+        # Duplicate of mixins.fixtures.get_test_purge_time(); we don't want to import pytest as a transitive dependency.
+        timeout = TEST_NIGHTLY_CI_RESOURCES_PURGE_TIMEOUT if cls._is_nightly() else TEST_RESOURCE_PURGE_TIMEOUT
+        now = datetime.now(timezone.utc)
+        purge_deadline = now + timeout
+        # Round UP to the next hour boundary: that is when resources will be deleted.
+        purge_hour = purge_deadline + (datetime.min.replace(tzinfo=timezone.utc) - purge_deadline) % timedelta(hours=1)
+        return purge_hour.strftime("%Y%m%d%H")
 
     def _create_readme(self) -> None:
         debug_notebook_link = self._installation.workspace_markdown_link('debug notebook', 'DEBUG.py')

src/databricks/labs/ucx/mixins/fixtures.py

@@ -8,7 +8,7 @@
 import subprocess
 import sys
 from collections.abc import Callable, Generator, MutableMapping
-from datetime import timedelta, datetime
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import BinaryIO
 
@@ -58,7 +58,9 @@
 # pylint: disable=redefined-outer-name,too-many-try-statements,import-outside-toplevel,unnecessary-lambda,too-complex,invalid-name
 
 logger = logging.getLogger(__name__)
-TEST_JOBS_PURGE_TIMEOUT = timedelta(hours=1, minutes=15)  # 15 minutes grace for jobs starting at the end of the hour
+
+"""Preserve resources created during tests for at least this long."""
+TEST_RESOURCE_PURGE_TIMEOUT = timedelta(hours=1)
 
 
 def factory(name, create, remove):
@@ -1461,8 +1463,15 @@ def delete(dashboard: SDKDashboard) -> None:
     yield from factory("dashboard", create, delete)
 
 
-def get_test_purge_time() -> str:
-    return (datetime.utcnow() + TEST_JOBS_PURGE_TIMEOUT).strftime("%Y%m%d%H")
+def get_test_purge_time(timeout: timedelta = TEST_RESOURCE_PURGE_TIMEOUT) -> str:
+    """Purge time for test objects, representing the (UTC-based) hour from which objects may be purged."""
+    # Note: this code is duplicated in the workflow installer (WorkflowsDeployment) so that it can avoid the
+    # transitive pytest deployment from this module.
+    now = datetime.now(timezone.utc)
+    purge_deadline = now + timeout
+    # Round UP to the next hour boundary: that is when resources will be deleted.
+    purge_hour = purge_deadline + (datetime.min.replace(tzinfo=timezone.utc) - purge_deadline) % timedelta(hours=1)
+    return purge_hour.strftime("%Y%m%d%H")
 
 
 def get_purge_suffix() -> str: