Fix EE volume mount issue

docs/changelog-fragments.d/1061.breaking.md

@@ -0,0 +1,5 @@
+The environment variable for user specified execution environment mounts is
+now a semi-colon delimited string. e.g.
+`ANSIBLE_NAVIGATOR_EXECUTION_ENVIRONMENT_VOLUME_MOUNTS='\p1:\p1;\p2:p2'`
+
+-- by {user}`cidrblock`

docs/conf.py

@@ -110,11 +110,8 @@ def parse_with_fetch(*args, **kwargs) -> str:
     ),
     ("py:class", "ansible_navigator.configuration_subsystem.defs_presentable.TCli"),
     ("py:class", "ansible_navigator.configuration_subsystem.defs_presentable.TEnt"),
-    ("py:class", "ansible_navigator.steps.T"),
     ("py:class", "ansible_navigator.tm_tokenize.fchainmap.TKey"),
     ("py:class", "ansible_navigator.tm_tokenize.fchainmap.TValue"),
-    ("py:class", "ansible_navigator.tm_tokenize.utils.T"),
-    ("py:class", "ansible_navigator.ui_framework.content_defs.T"),
     ("py:class", "ansible_runner.runner.Runner"),
     ("py:class", "argparse._SubParsersAction"),
     ("py:class", "Captures"),
@@ -125,6 +122,7 @@ def parse_with_fetch(*args, **kwargs) -> str:
     ("py:class", "ContentView"),
     ("py:class", "CursesLine"),
     ("py:class", "CursesLines"),
+    ("py:class", "dataclasses.InitVar"),
     ("py:class", "Entry"),
     ("py:class", "FieldButton"),
     ("py:class", "FieldChecks"),
@@ -149,10 +147,13 @@ def parse_with_fetch(*args, **kwargs) -> str:
     ("py:class", "Window"),
     ("py:class", "yaml.cyaml.CSafeDumper"),
     ("py:class", "yaml.nodes.ScalarNode"),
-    ("py:obj", "ansible_navigator.steps.T"),
     ("py:obj", "ansible_navigator.tm_tokenize.fchainmap.TKey"),
     ("py:obj", "ansible_navigator.tm_tokenize.fchainmap.TValue"),
-    ("py:obj", "ansible_navigator.ui_framework.content_defs.T"),
+]
+
+nitpick_ignore_regex = [
+    # Any single letter TypeVar, class or object
+    ("py:(class|obj)", r"^.*\.[A-Z]$"),
 ]
 
 # Add any Sphinx extension module names here, as strings. They can be

src/ansible_navigator/configuration_subsystem/configurator.py

@@ -196,7 +196,10 @@ def _apply_environment_variables(self) -> None:
             if set_env_var is not None:
                 if self._config.internals.initializing or entry.change_after_initial:
                     if entry.cli_parameters is not None and entry.cli_parameters.nargs == "+":
-                        entry.value.current = set_env_var.split(",")
+                        entry.value.current = [
+                            value.strip()
+                            for value in set_env_var.split(entry.environment_variable_split_char)
+                        ]
                     else:
                         entry.value.current = set_env_var
                     entry.value.source = C.ENVIRONMENT_VARIABLE

src/ansible_navigator/configuration_subsystem/definitions.py

@@ -127,6 +127,8 @@ class SettingsEntry:
     #: Override the default, generated environment variable
     environment_variable_override: Optional[str] = None
     #: Over the default settings file path, dot delimited representation in tree
+    environment_variable_split_char: str = ","
+    #: The character used to split an environment variable value into a list
     settings_file_path_override: Optional[str] = None
     #: Subcommand this should this be used for
     subcommands: Union[List[str], Constants] = Constants.ALL
@@ -164,10 +166,16 @@ def name_dashed(self) -> str:
 
     def settings_file_path(self, prefix: str) -> str:
         """Generate an effective settings file path for this entry"""
+        if prefix:
+            prefix_str = f"{prefix}."
+        else:
+            prefix_str = prefix
+
         if self.settings_file_path_override is not None:
-            sfp = f"{prefix}.{self.settings_file_path_override}"
+            sfp = f"{prefix_str}{self.settings_file_path_override}"
         else:
-            sfp = f"{prefix}.{self.name}"
+            sfp = f"{prefix_str}{self.name}"
+
         return sfp.replace("_", "-")
 
 

src/ansible_navigator/configuration_subsystem/navigator_configuration.py

@@ -305,6 +305,7 @@ class Internals:
             name="execution_environment_volume_mounts",
             cli_parameters=CliParameters(action="append", nargs="+", short="--eev"),
             delay_post_process=True,
+            environment_variable_split_char=";",
             settings_file_path_override="execution-environment.volume-mounts",
             short_description=(
                 "Specify volume to be bind mounted within an execution environment"

src/ansible_navigator/configuration_subsystem/navigator_post_processor.py

@@ -6,13 +6,17 @@
 import shlex
 import shutil
 
+from dataclasses import InitVar
 from dataclasses import dataclass
 from dataclasses import field
 from enum import Enum
 from functools import partialmethod
+from itertools import chain
+from itertools import repeat
 from pathlib import Path
 from typing import List
 from typing import Tuple
+from typing import TypeVar
 
 from ..utils.functions import ExitMessage
 from ..utils.functions import ExitPrefix
@@ -50,6 +54,9 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
+PostProcessorReturn = Tuple[List[LogMessage], List[ExitMessage]]
+
+
 class VolumeMountOption(Enum):
     """Options that can be tagged on to the end of volume mounts.
 
@@ -66,29 +73,79 @@ class VolumeMountOption(Enum):
     z = "z"  # pylint: disable=invalid-name
 
 
-@dataclass
-class VolumeMount:
-    """Describes EE volume mounts."""
+V = TypeVar("V", bound="VolumeMount")  # pylint: disable=invalid-name
 
-    #: The name of the config option requiring this volume mount.
-    calling_option: str
 
-    #: The source path of the volume mount.
-    src: str
+class VolumeMountError(Exception):
+    """Custom exception raised when building VolumeMounts."""
 
-    #: The destination path in the container for the volume mount.
-    dest: str
 
-    #: Options for the bind mount.
-    options: List[VolumeMountOption] = field(default_factory=list)
+@dataclass(frozen=True)
+class VolumeMount:
+    """Describes EE volume mounts."""
 
-    def exists(self) -> bool:
-        """Determine if the volume mount source exists."""
-        return Path(self.src).exists()
+    fs_source: str
+    """The source file system path of the volume mount"""
+    fs_destination: str
+    """The destination file system path in the container for the volume mount"""
+    settings_entry: str
+    """The name of the settings entry requiring this volume mount"""
+    source: C = field(compare=False)
+    """The settings source for this volume mount"""
+    options_string: InitVar[str]
+    """Comma delimited options"""
+    options: Tuple[VolumeMountOption, ...] = ()
+    """Options for the bind mount"""
+
+    def __post_init__(self, options_string):
+        """Post process the ``VolumeMount`` and perform sanity checks.
+
+        :raises VolumeMountError: When a viable VolumeMount cannot be created
+        """
+        # pylint: disable=too-many-branches
+        errors = []
+        # Validate the source
+        if isinstance(self.fs_source, str):
+            if self.fs_source == "":
+                errors.append("Source not provided.")
+            elif not Path(self.fs_source).exists():
+                errors.append(f"Source: '{self.fs_source}' does not exist.")
+        else:
+            errors.append(f"Source: '{self.fs_source}' is not a string.")
+
+        # Validate the destination
+        if isinstance(self.fs_destination, str):
+            if self.fs_destination == "":
+                errors.append("Destination not provided.")
+        else:
+            errors.append(f"Destination: '{self.fs_destination}' is not a string.")
+
+        # Validate and populate _options
+        if isinstance(options_string, str):
+            if not options_string == "":
+                options = []
+                option_values = [o.value for o in VolumeMountOption]
+                for option in options_string.split(","):
+                    if option not in option_values:
+                        errors.append(
+                            f"Unrecognized label: '{option}',"
+                            f" available labels include"
+                            f" {oxfordcomma(option_values, condition='and/or')}.",
+                        )
+                    else:
+                        options.append(VolumeMountOption(option))
+                unique = sorted(set(options), key=options.index)
+                # frozen, cannot use simple assignment to initialize fields, and must use:
+                object.__setattr__(self, "options", tuple(unique))
+        else:
+            errors.append(f"Labels: '{options_string}' is not a string.")
+
+        if errors:
+            raise VolumeMountError(" ".join(errors))
 
     def to_string(self) -> str:
         """Render the volume mount in a way that (docker|podman) understands."""
-        out = f"{self.src}:{self.dest}"
+        out = f"{self.fs_source}:{self.fs_destination}"
         if self.options:
             joined_opts = ",".join(o.value for o in self.options)
             out += f":{joined_opts}"
@@ -112,9 +169,6 @@ class ModeChangeRequest:
     """The desired mode"""
 
 
-PostProcessorReturn = Tuple[List[LogMessage], List[ExitMessage]]
-
-
 class NavigatorPostProcessor:
     # pylint:disable=too-many-public-methods
     """application post processor"""
@@ -355,86 +409,105 @@ def execution_environment_volume_mounts(
         entry: SettingsEntry,
         config: ApplicationConfiguration,
     ) -> PostProcessorReturn:
+        """Post process set_environment_variable.
+
+        :param entry: The current settings entry
+        :param config: The full application configuration
+        :return: An instance of the standard post process return object
+        """
         # pylint: disable=unused-argument
         # pylint: disable=too-many-branches
-        """Post process set_environment_variable"""
+        # pylint: disable=too-many-locals
+
         messages: List[LogMessage] = []
         exit_messages: List[ExitMessage] = []
-        if entry.value.source in [
-            C.ENVIRONMENT_VARIABLE,
-            C.USER_CLI,
-        ]:
-            volume_mounts = flatten_list(entry.value.current)
-            for mount_path in volume_mounts:
-                parts = mount_path.split(":")
-                if len(parts) > 3:
+        entry_name = entry.settings_file_path(prefix="")
+        entry_source = entry.value.source
+        volume_mounts: List[VolumeMount] = []
+
+        if entry_source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI):
+            hint = (
+                "Try again with format <source-path>:<destination-path>:<labels>'."
+                " Note: label is optional."
+            )
+            mount_strings = flatten_list(entry.value.current)
+            for mount_str in mount_strings:
+                src, dest, labels, *left_overs = chain(mount_str.split(":"), repeat("", 3))
+                if any(left_overs):
                     exit_msg = (
-                        "The following execution-environment-volume-mounts"
-                        f" entry could not be parsed: {mount_path}"
+                        f"The following {entry_name} entry could not be parsed:"
+                        f" {mount_str} ({entry_source.value})"
                     )
                     exit_messages.append(ExitMessage(message=exit_msg))
-                    if entry.cli_parameters:
-                        exit_msg = (
-                            "Try again with format "
-                            + f"'{entry.cli_parameters.short}"
-                            + " <source-path>:<destination-path>:<label(Z or z)>'."
-                            + " Note: label is optional."
-                        )
-                        exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT))
-                    return messages, exit_messages
-
-            entry.value.current = volume_mounts
+                    exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT))
 
-        elif entry.value.current is not C.NOT_SET:
-            parsed_volume_mounts = []
-            volume_mounts = to_list(entry.value.current)
-            for mount_obj in volume_mounts:
-                if not isinstance(mount_obj, dict):
+                try:
+                    volume_mounts.append(
+                        VolumeMount(
+                            fs_source=src,
+                            fs_destination=dest,
+                            options_string=labels,
+                            settings_entry=entry_name,
+                            source=entry_source,
+                        ),
+                    )
+                except VolumeMountError as exc:
                     exit_msg = (
-                        "The following execution-environment.volume-mounts"
-                        f" entry could not be parsed: {mount_obj}"
+                        f"The following {entry_name} entry could not be parsed:"
+                        f" {mount_str} ({entry.value.source.value}). Errors were found: {str(exc)}"
                     )
                     exit_messages.append(ExitMessage(message=exit_msg))
-                    if entry.cli_parameters:
+                    exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT))
+
+        elif entry.value.source is C.USER_CFG:
+            hint = (
+                "The value of execution-environment.volume-mounts should be a list"
+                " of dictionaries and valid keys are 'src', 'dest' and 'label'."
+            )
+            if not isinstance(entry.value.current, list):
+                exit_msg = f"{entry_name} entries could not be parsed. ({entry_source.value})"
+                exit_messages.append(ExitMessage(message=exit_msg))
+                exit_msg = hint
+                exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT))
+            else:
+                for volume_mount in entry.value.current:
+                    try:
+                        volume_mounts.append(
+                            VolumeMount(
+                                fs_source=volume_mount.get("src"),
+                                fs_destination=volume_mount.get("dest"),
+                                options_string=volume_mount.get("label", ""),
+                                settings_entry=entry_name,
+                                source=entry_source,
+                            ),
+                        )
+                    except (AttributeError, VolumeMountError) as exc:
                         exit_msg = (
-                            "The value of execution-environment.volume-mounts"
-                            + "should be list of dictionaries"
-                            + " and valid keys are 'src', 'dest' and 'label'."
+                            f"The following {entry_name} entry could not be parsed:  {volume_mount}"
+                            f" ({entry_source.value}). Errors were found: {str(exc)}"
                         )
-                        exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT))
-                    return messages, exit_messages
+                        exit_messages.append(ExitMessage(message=exit_msg))
+                        exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT))
 
-                try:
-                    mount_path = f"{mount_obj['src']}:{mount_obj['dest']}"
-                    if mount_obj.get("label"):
-                        mount_path += f":{mount_obj['label']}"
-                    parsed_volume_mounts.append(mount_path)
-                except KeyError as exc:
-                    exit_msg = (
-                        f"Failed to parse following execution-environment.volume-mounts"
-                        f" entry: '{mount_obj}'. Value of '{str(exc)}' key not provided."
-                    )
-                    exit_messages.append(ExitMessage(message=exit_msg))
-                    exit_hint_msg = (
-                        " Valid keys are 'src', 'dest' and 'label'. Note: label key is optional."
-                    )
-                    exit_messages.append(ExitMessage(message=exit_hint_msg, prefix=ExitPrefix.HINT))
+        # Get out fast if we had any errors
+        if exit_messages:
+            return messages, exit_messages
 
-                    return messages, exit_messages
+        # New mounts were provided
+        if volume_mounts:
+            entry.value.current = [v.to_string() for v in volume_mounts]
 
-            entry.value.current = parsed_volume_mounts
+        # Extra mounts were requested, these get added to either
+        # new_mounts, C.PREVIOUS_CLI or C.NOT_SET
+        if self.extra_volume_mounts:
+            if not isinstance(entry.value.current, list):
+                entry.value.current = set()
+            entry.value.current.extend(v.to_string() for v in self.extra_volume_mounts)
 
-        if self.extra_volume_mounts and entry.value.current is C.NOT_SET:
-            entry.value.current = []
+        # Finally, ensure the list has no duplicates
+        if isinstance(entry.value.current, list):
+            entry.value.current = sorted(set(entry.value.current), key=entry.value.current.index)
 
-        for mount in self.extra_volume_mounts:
-            if not mount.exists():
-                exit_msg = (
-                    f"The volume mount source path '{mount.src}', needed by "
-                    f"{mount.calling_option}, does not exist."
-                )
-                exit_messages.append(ExitMessage(message=exit_msg))
-            entry.value.current.append(mount.to_string())
         return messages, exit_messages
 
     @staticmethod