Change location of ephemeral directory

src/molecule/scenario.py

@@ -26,12 +26,14 @@
 import os
 import shutil
 
+from functools import cached_property
 from pathlib import Path
 from time import sleep
 from typing import TYPE_CHECKING
 
 from molecule import scenarios, util
 from molecule.constants import RC_TIMEOUT
+from molecule.text import checksum
 
 
 if TYPE_CHECKING:
@@ -54,6 +56,14 @@ def __init__(self, config: Config) -> None:
         self.config = config
         self._setup()
 
+    def __repr__(self) -> str:
+        """Return a representation of the instance.
+
+        Returns:
+            A string.
+        """
+        return f"<Scenario {self.name} from {self.config.project_directory}>"
+
     def _remove_scenario_state_directory(self) -> None:
         """Remove scenario cached disk stored state."""
         directory = str(Path(self.ephemeral_directory).parent)
@@ -111,7 +121,7 @@ def directory(self) -> str:
             path = Path(self.config.molecule_file).parent
         return str(path)
 
-    @property
+    @cached_property
     def ephemeral_directory(self) -> str:
         """Acquire the ephemeral directory.
 
@@ -121,22 +131,17 @@ def ephemeral_directory(self) -> str:
         Raises:
             SystemExit: If lock cannot be acquired before timeout.
         """
-        path: str | Path | None = os.getenv("MOLECULE_EPHEMERAL_DIRECTORY", None)
-        if not path:
+        path: Path
+        if "MOLECULE_EPHEMERAL_DIRECTORY" not in os.environ:
             project_directory = Path(self.config.project_directory).name
 
             if self.config.is_parallel:
                 project_directory = f"{project_directory}-{self.config._run_uuid}"  # noqa: SLF001
 
-            project_scenario_directory = Path(
-                self.config.cache_directory,
-                project_directory,
-                self.name,
-            )
-            path = ephemeral_directory(project_scenario_directory)
-
-        if isinstance(path, str):
-            path = Path(path)
+            project_scenario_directory = f"molecule.{checksum(project_directory, 4)}.{self.name}"
+            path = self.config.runtime.cache_dir / "tmp" / project_scenario_directory
+        else:
+            path = Path(os.getenv("MOLECULE_EPHEMERAL_DIRECTORY", ""))
 
         if os.environ.get("MOLECULE_PARALLEL", False) and not self._lock:
             lock_file = path / ".lock"
@@ -157,7 +162,7 @@ def ephemeral_directory(self) -> str:
                     LOG.warning("Timedout trying to acquire lock on %s", path)
                     raise SystemExit(RC_TIMEOUT)
 
-        return str(path)
+        return path.absolute().as_posix()
 
     @property
     def inventory_directory(self) -> str:
@@ -306,36 +311,3 @@ def _setup(self) -> None:
         inventory = Path(self.inventory_directory)
         if not inventory.is_dir():
             inventory.mkdir(exist_ok=True, parents=True)
-
-
-def ephemeral_directory(path: Path | None = None) -> Path:
-    """Return temporary directory to be used by molecule.
-
-    Molecule users should not make any assumptions about its location,
-    permissions or its content as this may change in future release.
-
-    Args:
-        path: Ephemeral directory name.
-
-    Returns:
-        The full ephemeral directory path.
-
-    Raises:
-        RuntimeError: If ephemeral directory location cannot be determined
-    """
-    d: str | Path | None = os.getenv("MOLECULE_EPHEMERAL_DIRECTORY")
-    if not d:
-        d = os.getenv("XDG_CACHE_HOME", Path("~/.cache").expanduser())
-    if not d:
-        msg = "Unable to determine ephemeral directory to use."
-        raise RuntimeError(msg)
-
-    if isinstance(d, str):
-        d = Path(d)
-    d = d.resolve() / (path if path else "molecule")
-
-    if not d.is_dir():
-        os.umask(0o077)
-        d.mkdir(mode=0o700, parents=True, exist_ok=True)
-
-    return d

src/molecule/text.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import re
 
 from typing import TYPE_CHECKING
 
 
 if TYPE_CHECKING:
+    from pathlib import Path
     from typing import AnyStr
 
 
@@ -108,3 +111,21 @@ def _to_unicode(data: AnyStr) -> str:
     if isinstance(data, bytes):
         return data.decode("utf-8")
     return data
+
+
+def checksum(data: str | Path, length: int = 5) -> str:
+    """Returns a checksum for the given data.
+
+    Args:
+        data: The data to checksum.
+        length: The length of the checksum.
+    Returns:
+        A checksum string.
+    """
+    data = str(data)
+    # Hash the input string using SHA-256
+    hash_object = hashlib.sha256(data.encode("utf-8"))
+    # Convert the hash to a base64-encoded string
+    base64_hash = base64.urlsafe_b64encode(hash_object.digest()).decode("utf-8")
+    # Truncate the result to the desired length
+    return base64_hash[:length]

tests/unit/test_scenario.py

@@ -27,7 +27,8 @@
 
 import pytest
 
-from molecule import config, scenario, util
+from molecule import config, util
+from molecule.scenario import Scenario
 
 
 if TYPE_CHECKING:
@@ -41,11 +42,11 @@
 def _instance(
     patched_config_validate: Mock,
     config_instance: config.Config,
-) -> scenario.Scenario:
-    return scenario.Scenario(config_instance)
+) -> Scenario:
+    return Scenario(config_instance)
 
 
-def test_prune(_instance: scenario.Scenario) -> None:  # noqa: PT019, D103
+def test_prune(_instance: Scenario) -> None:  # noqa: PT019, D103
     e_dir = Path(_instance.ephemeral_directory)
     # prune data also includes files in the scenario inventory dir,
     # which is "<e_dir>/inventory" by default.
@@ -86,31 +87,31 @@ def test_prune(_instance: scenario.Scenario) -> None:  # noqa: PT019, D103
         assert not (e_dir / pruned_dir).is_dir()
 
 
-def test_config_member(_instance: scenario.Scenario) -> None:  # noqa: PT019, D103
+def test_config_member(_instance: Scenario) -> None:  # noqa: PT019, D103
     assert isinstance(_instance.config, config.Config)
 
 
 def test_scenario_init_calls_setup(  # noqa: D103
     patched_scenario_setup: Mock,
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     patched_scenario_setup.assert_called_once_with()
 
 
 def test_scenario_name_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.name == "default"
 
 
 def test_ephemeral_directory_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert os.access(_instance.ephemeral_directory, os.W_OK)
 
 
 def test_scenario_inventory_directory_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     ephemeral_directory = Path(_instance.config.scenario.ephemeral_directory)
     e_dir = ephemeral_directory / "inventory"
@@ -119,7 +120,7 @@ def test_scenario_inventory_directory_property(  # noqa: D103
 
 
 def test_check_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     sequence = [
         "dependency",
@@ -137,59 +138,59 @@ def test_check_sequence_property(  # noqa: D103
 
 
 def test_converge_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     sequence = ["dependency", "create", "prepare", "converge"]
 
     assert sequence == _instance.converge_sequence
 
 
 def test_create_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     sequence = ["dependency", "create", "prepare"]
 
     assert sequence == _instance.create_sequence
 
 
 def test_dependency_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.dependency_sequence == ["dependency"]
 
 
 def test_destroy_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.destroy_sequence == ["dependency", "cleanup", "destroy"]
 
 
 def test_idempotence_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.idempotence_sequence == ["idempotence"]
 
 
 def test_prepare_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.prepare_sequence == ["prepare"]
 
 
 def test_side_effect_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.side_effect_sequence == ["side_effect"]
 
 
 def test_syntax_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.syntax_sequence == ["syntax"]
 
 
 def test_test_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     sequence = [
         "dependency",
@@ -210,21 +211,21 @@ def test_test_sequence_property(  # noqa: D103
 
 
 def test_verify_sequence_property(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     assert _instance.verify_sequence == ["verify"]
 
 
 def test_sequence_property_with_invalid_subcommand(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     _instance.config.command_args = {"subcommand": "invalid"}
 
     assert _instance.sequence == []
 
 
 def test_setup_creates_ephemeral_and_inventory_directories(  # noqa: D103
-    _instance: scenario.Scenario,  # noqa: PT019
+    _instance: Scenario,  # noqa: PT019
 ) -> None:
     ephemeral_dir = _instance.config.scenario.ephemeral_directory
     inventory_dir = _instance.config.scenario.inventory_directory
@@ -233,12 +234,8 @@ def test_setup_creates_ephemeral_and_inventory_directories(  # noqa: D103
 
     assert Path(ephemeral_dir).is_dir()
     assert Path(inventory_dir).is_dir()
-
-
-def test_ephemeral_directory() -> None:  # noqa: D103
     # assure we can write to ephemeral directory
-    path = Path("foo/bar")
-    assert os.access(scenario.ephemeral_directory(path), os.W_OK)
+    assert os.access(ephemeral_dir, os.W_OK)
 
 
 def test_ephemeral_directory_overridden_via_env_var(
@@ -253,22 +250,9 @@ def test_ephemeral_directory_overridden_via_env_var(
     """
     monkeypatch.chdir(tmp_path)
     monkeypatch.setenv("MOLECULE_EPHEMERAL_DIRECTORY", "foo/bar")
+    scenario = Scenario(config.Config(""))
 
     path = Path("foo/bar")
-    assert os.access(scenario.ephemeral_directory(path), os.W_OK)
-
-
-def test_ephemeral_directory_overridden_via_env_var_uses_absolute_path(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    """Confirm MOLECULE_EPHEMERAL_DIRECTORY uses absolute path.
-
-    Args:
-        monkeypatch: Pytest monkeypatch fixture.
-        tmp_path: Pytest tmp_path fixture.
-    """
-    monkeypatch.chdir(tmp_path)
-    monkeypatch.setenv("MOLECULE_EPHEMERAL_DIRECTORY", "foo/bar")
-
-    assert Path(scenario.ephemeral_directory()).is_absolute()
+    assert os.access(path, os.W_OK)
+    # Confirm MOLECULE_EPHEMERAL_DIRECTORY uses absolute path.
+    assert Path(scenario.ephemeral_directory).is_absolute()