From fcf7b1aabe87c32636ad3e15247e00aebc3a9abc Mon Sep 17 00:00:00 2001 From: cidrblock Date: Mon, 7 Mar 2022 15:46:26 -0800 Subject: [PATCH 01/18] Fix --- .../configuration_subsystem/definitions.py | 154 ++++++++++++++++- .../navigator_post_processor.py | 161 +++++------------- .../ansible-navigator.yml | 2 +- tests/unit/configuration_subsystem/data.py | 2 +- .../test_navigator_post_processor.py | 49 ++++-- 5 files changed, 236 insertions(+), 132 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/definitions.py b/src/ansible_navigator/configuration_subsystem/definitions.py index 90b354c9c..403c57df3 100644 --- a/src/ansible_navigator/configuration_subsystem/definitions.py +++ b/src/ansible_navigator/configuration_subsystem/definitions.py @@ -6,11 +6,16 @@ from dataclasses import dataclass from dataclasses import field from enum import Enum +from itertools import chain +from itertools import repeat +from pathlib import Path from typing import TYPE_CHECKING from typing import Any from typing import Iterable from typing import List from typing import Optional +from typing import Type +from typing import TypeVar from typing import Union from ..utils.functions import oxfordcomma @@ -164,10 +169,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("_", "-") @@ -221,3 +232,142 @@ def entry(self, name) -> SettingsEntry: def subcommand(self, name) -> SubCommand: """Retrieve a configuration subcommand by name""" return self._get_by_name(name, "subcommands") + + +class VolumeMountOption(Enum): + """Options that can be tagged on to the end of volume mounts. + + Usually these are used for things like selinux relabeling, but there are + some other valid options as well, which can and should be added here as + needed. See ``man podman-run`` and ``man docker-run`` for valid choices and + keep in mind that we support both runtimes. + """ + + # Relabel as private + Z = "Z" + + # Relabel as shared. + z = "z" # pylint: disable=invalid-name + + +V = TypeVar("V", bound="VolumeMount") # pylint: disable=invalid-name + + +@dataclass +class VolumeMount: + """Describes EE volume mounts.""" + + fs_destination: str + """The destination file system path in the container for the volume mount""" + fs_source: str + """The source file system path of the volume mount""" + settings_entry: str + """The name of the settings entry requiring this volume mount""" + source: Constants + """The settings source for this volume mount""" + options: List[VolumeMountOption] = field(default_factory=list) + """Options for the bind mount""" + + def exists(self) -> bool: + """Determine if the volume mount source exists.""" + return Path(self.fs_source).exists() + + def to_string(self) -> str: + """Render the volume mount in a way that (docker|podman) understands.""" + 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}" + return out + + @classmethod + def from_string(cls: Type[V], settings_entry: str, source: Constants, string: str) -> V: + """Create a ``VolumeMount`` from a string. + + :param settings_entry: The settings entry + :param source: The source of the string + :param string: The string from which the volume mount will be created + :raises ValueError: When source or destination are missing, or unrecognized label + :returns: A populated volume mount + """ + fs_source, fs_destination, options, *left_overs = chain(string.split(":"), repeat("", 3)) + if options: + option_list = [ + const + for const in VolumeMountOption + for option in options.split(",") + if const.value == option + ] + unrecognized_label = len(options.split(",")) != len(option_list) + + if any(left_overs): + raise ValueError("Not formatted correctly") + if unrecognized_label: + raise ValueError("Unrecognized label in volume mount string") + if not fs_source: + raise ValueError("Could not extract source from string") + if not fs_destination: + raise ValueError("Could not extract destination from string") + return cls( + fs_source=fs_source, + fs_destination=fs_destination, + options=option_list, + settings_entry=settings_entry, + source=source, + ) + + @classmethod + def from_dictionary( + cls: Type[V], + settings_entry: str, + source: Constants, + dictionary: dict, + ) -> V: + """Create a ``VolumeMount`` from a dictionary. + + :param dictionary: The dictionary from which the volume mount will be created + :param settings_entry: The settings entry + :param source: The source of the string + :raises ValueError: When source or destination are missing, or unrecognized label + :returns: A populated volume mount + """ + options = dictionary.get("label", "") + if options: + option_list = [ + const + for const in VolumeMountOption + for option in options.split(",") + if const.value == option + ] + unrecognized_label = len(options.split(",")) != len(option_list) + if unrecognized_label: + raise ValueError("Unrecognized labe in volume mount string") + try: + return cls( + fs_source=dictionary["src"], + fs_destination=dictionary["dest"], + options=option_list, + settings_entry=settings_entry, + source=source, + ) + except KeyError as exc: + raise ValueError("Source or destination missing") from exc + except TypeError as exc: + raise ValueError("Not a dictionary") from exc + + +class Mode(Enum): + """An enum to restrict mode type.""" + + STDOUT: str = "stdout" + INTERACTIVE: str = "interactive" + + +@dataclass +class ModeChangeRequest: + """Data structure to contain a mode change request by a settings entry.""" + + entry: str + """The entry making the request""" + mode: Mode + """The desired mode""" diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index ac334c0f6..a837f6889 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -6,9 +6,6 @@ import shlex import shutil -from dataclasses import dataclass -from dataclasses import field -from enum import Enum from functools import partialmethod from pathlib import Path from typing import List @@ -26,7 +23,10 @@ from .definitions import ApplicationConfiguration from .definitions import CliParameters from .definitions import Constants as C +from .definitions import Mode +from .definitions import ModeChangeRequest from .definitions import SettingsEntry +from .definitions import VolumeMount def _post_processor(func): @@ -50,68 +50,6 @@ def wrapper(*args, **kwargs): return wrapper -class VolumeMountOption(Enum): - """Options that can be tagged on to the end of volume mounts. - - Usually these are used for things like selinux relabeling, but there are - some other valid options as well, which can and should be added here as - needed. See ``man podman-run`` and ``man docker-run`` for valid choices and - keep in mind that we support both runtimes. - """ - - # Relabel as private - Z = "Z" - - # Relabel as shared. - z = "z" # pylint: disable=invalid-name - - -@dataclass -class VolumeMount: - """Describes EE volume mounts.""" - - #: The name of the config option requiring this volume mount. - calling_option: str - - #: The source path of the volume mount. - src: str - - #: The destination path in the container for the volume mount. - dest: str - - #: Options for the bind mount. - options: List[VolumeMountOption] = field(default_factory=list) - - def exists(self) -> bool: - """Determine if the volume mount source exists.""" - return Path(self.src).exists() - - def to_string(self) -> str: - """Render the volume mount in a way that (docker|podman) understands.""" - out = f"{self.src}:{self.dest}" - if self.options: - joined_opts = ",".join(o.value for o in self.options) - out += f":{joined_opts}" - return out - - -class Mode(Enum): - """An enum to restrict mode type.""" - - STDOUT: str = "stdout" - INTERACTIVE: str = "interactive" - - -@dataclass -class ModeChangeRequest: - """Data structure to contain a mode change request by a settings entry.""" - - entry: str - """The entry making the request""" - mode: Mode - """The desired mode""" - - PostProcessorReturn = Tuple[List[LogMessage], List[ExitMessage]] @@ -355,21 +293,25 @@ def execution_environment_volume_mounts( config: ApplicationConfiguration, ) -> PostProcessorReturn: # pylint: disable=unused-argument - # pylint: disable=too-many-branches """Post process set_environment_variable""" 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: + settings_entry = entry.settings_file_path(prefix="") + volume_mounts: List[VolumeMount] = [] + if entry.value.source in [C.ENVIRONMENT_VARIABLE, C.USER_CLI]: + for mount_str in flatten_list(entry.value.current): + try: + volume_mounts.append( + VolumeMount.from_string( + settings_entry=settings_entry, + source=entry.value.source, + string=mount_str, + ), + ) + except ValueError as exc: exit_msg = ( - "The following execution-environment-volume-mounts" - f" entry could not be parsed: {mount_path}" + f"The following {settings_entry} entry could not be parsed:" + f" {mount_str} ({entry.value.source.value}), {str(exc)}" ) exit_messages.append(ExitMessage(message=exit_msg)) if entry.cli_parameters: @@ -380,60 +322,43 @@ def execution_environment_volume_mounts( + " Note: label is optional." ) exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - return messages, exit_messages - - entry.value.current = volume_mounts - - 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): - exit_msg = ( - "The following execution-environment.volume-mounts" - f" entry could not be parsed: {mount_obj}" - ) - exit_messages.append(ExitMessage(message=exit_msg)) - if entry.cli_parameters: - exit_msg = ( - "The value of execution-environment.volume-mounts" - + "should be list of dictionaries" - + " and valid keys are 'src', 'dest' and 'label'." - ) - exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - return messages, exit_messages + elif entry.value.source is C.USER_CFG: + for list_entry in entry.value.current: 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: + volume_mounts.append( + VolumeMount.from_dictionary( + dictionary=list_entry, + settings_entry=settings_entry, + source=entry.value.source, + ), + ) + except ValueError as exc: exit_msg = ( - f"Failed to parse following execution-environment.volume-mounts" - f" entry: '{mount_obj}'. Value of '{str(exc)}' key not provided." + f"The following {settings_entry} entry could not be parsed:" + f" {list_entry} ({entry.value.source.value}), {str(exc)}" ) exit_messages.append(ExitMessage(message=exit_msg)) - exit_hint_msg = ( - " Valid keys are 'src', 'dest' and 'label'. Note: label key is optional." + exit_msg = ( + "The value of execution-environment.volume-mounts" + + "should be a list of dictionaries" + + " and valid keys are 'src', 'dest' and 'label'." ) - exit_messages.append(ExitMessage(message=exit_hint_msg, prefix=ExitPrefix.HINT)) - - return messages, exit_messages - - entry.value.current = parsed_volume_mounts + exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - if self.extra_volume_mounts and entry.value.current is C.NOT_SET: - entry.value.current = [] + volume_mounts.extend(self.extra_volume_mounts) - for mount in self.extra_volume_mounts: + for mount in 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." + f"The volume mount source path '{mount.fs_source}', configured with " + f"'{mount.settings_entry}' ({mount.source.value}), does not exist." ) exit_messages.append(ExitMessage(message=exit_msg)) - entry.value.current.append(mount.to_string()) + + if exit_messages: + return messages, exit_messages + entry.value.current = [mount.to_string() for mount in volume_mounts] return messages, exit_messages @staticmethod diff --git a/tests/fixtures/unit/configuration_subsystem/ansible-navigator.yml b/tests/fixtures/unit/configuration_subsystem/ansible-navigator.yml index 71094635f..0790b3815 100644 --- a/tests/fixtures/unit/configuration_subsystem/ansible-navigator.yml +++ b/tests/fixtures/unit/configuration_subsystem/ansible-navigator.yml @@ -46,7 +46,7 @@ ansible-navigator: - "--tls-verify=false" policy: never volume-mounts: - - src: "/test1" + - src: "/tmp" dest: "/test1" label: "Z" container-options: diff --git a/tests/unit/configuration_subsystem/data.py b/tests/unit/configuration_subsystem/data.py index e9688b310..a38c8bbc5 100644 --- a/tests/unit/configuration_subsystem/data.py +++ b/tests/unit/configuration_subsystem/data.py @@ -223,7 +223,7 @@ def cli_data(): ("exec_shell", "false", False), ("execution_environment", "false", False), ("execution_environment_image", "test_image:latest", "test_image:latest"), - ("execution_environment_volume_mounts", "/test1:/test1:Z", ["/test1:/test1:Z"]), + ("execution_environment_volume_mounts", "/tmp:/test1:Z", ["/tmp:/test1:Z"]), ("help_builder", "false", False), ("help_config", "false", False), ("help_doc", "false", False), diff --git a/tests/unit/configuration_subsystem/test_navigator_post_processor.py b/tests/unit/configuration_subsystem/test_navigator_post_processor.py index 639a0f573..2412656ec 100644 --- a/tests/unit/configuration_subsystem/test_navigator_post_processor.py +++ b/tests/unit/configuration_subsystem/test_navigator_post_processor.py @@ -2,24 +2,53 @@ import pytest -from ansible_navigator.configuration_subsystem.navigator_post_processor import ( - VolumeMount, -) -from ansible_navigator.configuration_subsystem.navigator_post_processor import ( - VolumeMountOption, -) +from ansible_navigator.configuration_subsystem.definitions import Constants +from ansible_navigator.configuration_subsystem.definitions import VolumeMount +from ansible_navigator.configuration_subsystem.definitions import VolumeMountOption @pytest.mark.parametrize( ("volmount", "expected"), ( - (VolumeMount("test_option", "/foo", "/bar"), "/foo:/bar"), - (VolumeMount("test_option", "/foo", "/bar", [VolumeMountOption.z]), "/foo:/bar:z"), ( - VolumeMount("test_option", "/foo", "/bar", [VolumeMountOption.z, VolumeMountOption.Z]), + VolumeMount( + settings_entry="test_option", + fs_source="/foo", + fs_destination="/bar", + source=Constants.USER_CLI, + ), + "/foo:/bar", + ), + ( + VolumeMount( + settings_entry="test_option", + fs_source="/foo", + fs_destination="/bar", + options=[VolumeMountOption.z], + source=Constants.USER_CLI, + ), + "/foo:/bar:z", + ), + ( + VolumeMount( + settings_entry="test_option", + fs_source="/foo", + fs_destination="/bar", + options=[VolumeMountOption.z, VolumeMountOption.Z], + source=Constants.USER_CLI, + ), "/foo:/bar:z,Z", ), - (VolumeMount("test_option", "/foo", "/bar", []), "/foo:/bar"), + ( + VolumeMount( + settings_entry="test_option", + fs_source="/foo", + fs_destination="/bar", + options=[], + source=Constants.USER_CLI, + ), + "/foo:/bar", + ), ), ids=( "normal mount", From 6483b2c006491887b649bf3a0a168737f2a4466c Mon Sep 17 00:00:00 2001 From: cidrblock Date: Mon, 7 Mar 2022 16:08:36 -0800 Subject: [PATCH 02/18] Move exit after CDC --- src/ansible_navigator/initialization.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/ansible_navigator/initialization.py b/src/ansible_navigator/initialization.py index 4ae2f919e..542c3ccdf 100644 --- a/src/ansible_navigator/initialization.py +++ b/src/ansible_navigator/initialization.py @@ -202,6 +202,9 @@ def parse_and_update( message = "Collection doc cache not attached to args.internals" messages.append(LogMessage(level=logging.DEBUG, message=message)) + if exit_messages: + return messages, exit_messages + for entry in args.entries: message = f"Running with {entry.name} as '{entry.value.current}'" message += f" ({type(entry.value.current).__name__}/{entry.value.source.value})" From 53caf5929d3a73ec43e29de731645b9f3952aac7 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Tue, 8 Mar 2022 09:58:10 -0800 Subject: [PATCH 03/18] Revise ad/replace logic --- .../navigator_post_processor.py | 46 ++++++++++++++----- 1 file changed, 35 insertions(+), 11 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index a837f6889..31b204383 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -6,6 +6,7 @@ import shlex import shutil +from dataclasses import asdict from functools import partialmethod from pathlib import Path from typing import List @@ -292,8 +293,15 @@ 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 - """Post process set_environment_variable""" + # pylint: disable=too-many-branches + messages: List[LogMessage] = [] exit_messages: List[ExitMessage] = [] settings_entry = entry.settings_file_path(prefix="") @@ -346,19 +354,35 @@ def execution_environment_volume_mounts( ) exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - volume_mounts.extend(self.extra_volume_mounts) - + exit_msg = ( + "The volume mount source path '{fs_source}', configured with '{settings_entry}'" + " ({source.value}), does not exist." + ) + # Check any new volume mounts first + new_mounts = [] for mount in volume_mounts: if not mount.exists(): - exit_msg = ( - f"The volume mount source path '{mount.fs_source}', configured with " - f"'{mount.settings_entry}' ({mount.source.value}), does not exist." - ) - exit_messages.append(ExitMessage(message=exit_msg)) + exit_messages.append(ExitMessage(message=exit_msg.format(**asdict(mount)))) + new_mounts.append(mount.to_string()) + + # Check extra mounts next + extra_mounts = [] + for mount in self.extra_volume_mounts: + if not mount.exists(): + exit_messages.append(ExitMessage(message=exit_msg.format(**asdict(mount)))) + extra_mounts.append(mount.to_string()) + + # New mounts were provided + if new_mounts: + entry.value.current = new_mounts + + # Extra mounts were requested, these get added to either + # new_mounts, C.PREVIOUS_CLI or C.NOT_SET + if extra_mounts: + if not isinstance(entry.value.current, list): + entry.value.current = [] + entry.value.current.extend(exit_messages) - if exit_messages: - return messages, exit_messages - entry.value.current = [mount.to_string() for mount in volume_mounts] return messages, exit_messages @staticmethod From 056ff9738b4d4578f1102732af9d5aaab7c31d9d Mon Sep 17 00:00:00 2001 From: cidrblock Date: Tue, 8 Mar 2022 10:05:36 -0800 Subject: [PATCH 04/18] TypeVar exclusion --- docs/conf.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/conf.py b/docs/conf.py index a06aa7155..fc30934d3 100644 --- a/docs/conf.py +++ b/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"), @@ -149,10 +146,13 @@ def parse_with_fetch(*args, **kwargs) -> str: ("py:class", "Window"), ("py:class", "yaml.cyaml.CDumper"), ("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 From d722019579413fb2c92698f02165482f68cc2dea Mon Sep 17 00:00:00 2001 From: cidrblock Date: Tue, 8 Mar 2022 10:28:39 -0800 Subject: [PATCH 05/18] No move --- .../configuration_subsystem/definitions.py | 144 ----------------- .../navigator_post_processor.py | 149 +++++++++++++++++- .../test_navigator_post_processor.py | 8 +- 3 files changed, 152 insertions(+), 149 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/definitions.py b/src/ansible_navigator/configuration_subsystem/definitions.py index 403c57df3..46ef5cdf4 100644 --- a/src/ansible_navigator/configuration_subsystem/definitions.py +++ b/src/ansible_navigator/configuration_subsystem/definitions.py @@ -6,16 +6,11 @@ from dataclasses import dataclass from dataclasses import field from enum import Enum -from itertools import chain -from itertools import repeat -from pathlib import Path from typing import TYPE_CHECKING from typing import Any from typing import Iterable from typing import List from typing import Optional -from typing import Type -from typing import TypeVar from typing import Union from ..utils.functions import oxfordcomma @@ -232,142 +227,3 @@ def entry(self, name) -> SettingsEntry: def subcommand(self, name) -> SubCommand: """Retrieve a configuration subcommand by name""" return self._get_by_name(name, "subcommands") - - -class VolumeMountOption(Enum): - """Options that can be tagged on to the end of volume mounts. - - Usually these are used for things like selinux relabeling, but there are - some other valid options as well, which can and should be added here as - needed. See ``man podman-run`` and ``man docker-run`` for valid choices and - keep in mind that we support both runtimes. - """ - - # Relabel as private - Z = "Z" - - # Relabel as shared. - z = "z" # pylint: disable=invalid-name - - -V = TypeVar("V", bound="VolumeMount") # pylint: disable=invalid-name - - -@dataclass -class VolumeMount: - """Describes EE volume mounts.""" - - fs_destination: str - """The destination file system path in the container for the volume mount""" - fs_source: str - """The source file system path of the volume mount""" - settings_entry: str - """The name of the settings entry requiring this volume mount""" - source: Constants - """The settings source for this volume mount""" - options: List[VolumeMountOption] = field(default_factory=list) - """Options for the bind mount""" - - def exists(self) -> bool: - """Determine if the volume mount source exists.""" - return Path(self.fs_source).exists() - - def to_string(self) -> str: - """Render the volume mount in a way that (docker|podman) understands.""" - 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}" - return out - - @classmethod - def from_string(cls: Type[V], settings_entry: str, source: Constants, string: str) -> V: - """Create a ``VolumeMount`` from a string. - - :param settings_entry: The settings entry - :param source: The source of the string - :param string: The string from which the volume mount will be created - :raises ValueError: When source or destination are missing, or unrecognized label - :returns: A populated volume mount - """ - fs_source, fs_destination, options, *left_overs = chain(string.split(":"), repeat("", 3)) - if options: - option_list = [ - const - for const in VolumeMountOption - for option in options.split(",") - if const.value == option - ] - unrecognized_label = len(options.split(",")) != len(option_list) - - if any(left_overs): - raise ValueError("Not formatted correctly") - if unrecognized_label: - raise ValueError("Unrecognized label in volume mount string") - if not fs_source: - raise ValueError("Could not extract source from string") - if not fs_destination: - raise ValueError("Could not extract destination from string") - return cls( - fs_source=fs_source, - fs_destination=fs_destination, - options=option_list, - settings_entry=settings_entry, - source=source, - ) - - @classmethod - def from_dictionary( - cls: Type[V], - settings_entry: str, - source: Constants, - dictionary: dict, - ) -> V: - """Create a ``VolumeMount`` from a dictionary. - - :param dictionary: The dictionary from which the volume mount will be created - :param settings_entry: The settings entry - :param source: The source of the string - :raises ValueError: When source or destination are missing, or unrecognized label - :returns: A populated volume mount - """ - options = dictionary.get("label", "") - if options: - option_list = [ - const - for const in VolumeMountOption - for option in options.split(",") - if const.value == option - ] - unrecognized_label = len(options.split(",")) != len(option_list) - if unrecognized_label: - raise ValueError("Unrecognized labe in volume mount string") - try: - return cls( - fs_source=dictionary["src"], - fs_destination=dictionary["dest"], - options=option_list, - settings_entry=settings_entry, - source=source, - ) - except KeyError as exc: - raise ValueError("Source or destination missing") from exc - except TypeError as exc: - raise ValueError("Not a dictionary") from exc - - -class Mode(Enum): - """An enum to restrict mode type.""" - - STDOUT: str = "stdout" - INTERACTIVE: str = "interactive" - - -@dataclass -class ModeChangeRequest: - """Data structure to contain a mode change request by a settings entry.""" - - entry: str - """The entry making the request""" - mode: Mode - """The desired mode""" diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 31b204383..74901574d 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -7,10 +7,17 @@ import shutil from dataclasses import asdict +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 Type +from typing import TypeVar from ..utils.functions import ExitMessage from ..utils.functions import ExitPrefix @@ -24,10 +31,7 @@ from .definitions import ApplicationConfiguration from .definitions import CliParameters from .definitions import Constants as C -from .definitions import Mode -from .definitions import ModeChangeRequest from .definitions import SettingsEntry -from .definitions import VolumeMount def _post_processor(func): @@ -54,6 +58,145 @@ def wrapper(*args, **kwargs): PostProcessorReturn = Tuple[List[LogMessage], List[ExitMessage]] +class VolumeMountOption(Enum): + """Options that can be tagged on to the end of volume mounts. + + Usually these are used for things like selinux relabeling, but there are + some other valid options as well, which can and should be added here as + needed. See ``man podman-run`` and ``man docker-run`` for valid choices and + keep in mind that we support both runtimes. + """ + + # Relabel as private + Z = "Z" + + # Relabel as shared. + z = "z" # pylint: disable=invalid-name + + +V = TypeVar("V", bound="VolumeMount") # pylint: disable=invalid-name + + +@dataclass +class VolumeMount: + """Describes EE volume mounts.""" + + fs_destination: str + """The destination file system path in the container for the volume mount""" + fs_source: str + """The source file system path of the volume mount""" + settings_entry: str + """The name of the settings entry requiring this volume mount""" + source: C + """The settings source for this volume mount""" + options: List[VolumeMountOption] = field(default_factory=list) + """Options for the bind mount""" + + def exists(self) -> bool: + """Determine if the volume mount source exists.""" + return Path(self.fs_source).exists() + + def to_string(self) -> str: + """Render the volume mount in a way that (docker|podman) understands.""" + 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}" + return out + + @classmethod + def from_string(cls: Type[V], settings_entry: str, source: C, string: str) -> V: + """Create a ``VolumeMount`` from a string. + + :param settings_entry: The settings entry + :param source: The source of the string + :param string: The string from which the volume mount will be created + :raises ValueError: When source or destination are missing, or unrecognized label + :returns: A populated volume mount + """ + fs_source, fs_destination, options, *left_overs = chain(string.split(":"), repeat("", 3)) + if options: + option_list = [ + const + for const in VolumeMountOption + for option in options.split(",") + if const.value == option + ] + unrecognized_label = len(options.split(",")) != len(option_list) + + if any(left_overs): + raise ValueError("Not formatted correctly") + if unrecognized_label: + raise ValueError("Unrecognized label in volume mount string") + if not fs_source: + raise ValueError("Could not extract source from string") + if not fs_destination: + raise ValueError("Could not extract destination from string") + return cls( + fs_source=fs_source, + fs_destination=fs_destination, + options=option_list, + settings_entry=settings_entry, + source=source, + ) + + @classmethod + def from_dictionary( + cls: Type[V], + settings_entry: str, + source: C, + dictionary: dict, + ) -> V: + """Create a ``VolumeMount`` from a dictionary. + + :param dictionary: The dictionary from which the volume mount will be created + :param settings_entry: The settings entry + :param source: The source of the string + :raises ValueError: When source or destination are missing, or unrecognized label + :returns: A populated volume mount + """ + options = dictionary.get("label", "") + if options: + option_list = [ + const + for const in VolumeMountOption + for option in options.split(",") + if const.value == option + ] + unrecognized_label = len(options.split(",")) != len(option_list) + if unrecognized_label: + raise ValueError("Unrecognized labe in volume mount string") + try: + return cls( + fs_source=dictionary["src"], + fs_destination=dictionary["dest"], + options=option_list, + settings_entry=settings_entry, + source=source, + ) + except KeyError as exc: + raise ValueError("Source or destination missing") from exc + except TypeError as exc: + raise ValueError("Not a dictionary") from exc + + +class Mode(Enum): + """An enum to restrict mode type.""" + + STDOUT: str = "stdout" + INTERACTIVE: str = "interactive" + + +@dataclass +class ModeChangeRequest: + """Data structure to contain a mode change request by a settings entry.""" + + entry: str + """The entry making the request""" + mode: Mode + """The desired mode""" + + class NavigatorPostProcessor: # pylint:disable=too-many-public-methods """application post processor""" diff --git a/tests/unit/configuration_subsystem/test_navigator_post_processor.py b/tests/unit/configuration_subsystem/test_navigator_post_processor.py index 2412656ec..cc8bdb398 100644 --- a/tests/unit/configuration_subsystem/test_navigator_post_processor.py +++ b/tests/unit/configuration_subsystem/test_navigator_post_processor.py @@ -3,8 +3,12 @@ import pytest from ansible_navigator.configuration_subsystem.definitions import Constants -from ansible_navigator.configuration_subsystem.definitions import VolumeMount -from ansible_navigator.configuration_subsystem.definitions import VolumeMountOption +from ansible_navigator.configuration_subsystem.navigator_post_processor import ( + VolumeMount, +) +from ansible_navigator.configuration_subsystem.navigator_post_processor import ( + VolumeMountOption, +) @pytest.mark.parametrize( From 70bb7311acd3dc9b3691ecde09d025e4b5b059c1 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Tue, 8 Mar 2022 14:25:33 -0800 Subject: [PATCH 06/18] Unit tests --- .../navigator_post_processor.py | 117 +++++++------ .../post_processors/__init__.py | 1 + ...est_execution_environment_volume_mounts.py | 157 ++++++++++++++++++ 3 files changed, 220 insertions(+), 55 deletions(-) create mode 100644 tests/unit/configuration_subsystem/post_processors/__init__.py create mode 100644 tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 74901574d..5d5f05242 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -115,19 +115,9 @@ def from_string(cls: Type[V], settings_entry: str, source: C, string: str) -> V: :returns: A populated volume mount """ fs_source, fs_destination, options, *left_overs = chain(string.split(":"), repeat("", 3)) - if options: - option_list = [ - const - for const in VolumeMountOption - for option in options.split(",") - if const.value == option - ] - unrecognized_label = len(options.split(",")) != len(option_list) if any(left_overs): raise ValueError("Not formatted correctly") - if unrecognized_label: - raise ValueError("Unrecognized label in volume mount string") if not fs_source: raise ValueError("Could not extract source from string") if not fs_destination: @@ -135,18 +125,13 @@ def from_string(cls: Type[V], settings_entry: str, source: C, string: str) -> V: return cls( fs_source=fs_source, fs_destination=fs_destination, - options=option_list, + options=cls._option_list_from_comma_string(options), settings_entry=settings_entry, source=source, ) @classmethod - def from_dictionary( - cls: Type[V], - settings_entry: str, - source: C, - dictionary: dict, - ) -> V: + def from_dictionary(cls: Type[V], settings_entry: str, source: C, dictionary: dict) -> V: """Create a ``VolumeMount`` from a dictionary. :param dictionary: The dictionary from which the volume mount will be created @@ -155,22 +140,16 @@ def from_dictionary( :raises ValueError: When source or destination are missing, or unrecognized label :returns: A populated volume mount """ + if not isinstance(dictionary, dict): + raise ValueError("Must be a list of dictionaries") + options = dictionary.get("label", "") - if options: - option_list = [ - const - for const in VolumeMountOption - for option in options.split(",") - if const.value == option - ] - unrecognized_label = len(options.split(",")) != len(option_list) - if unrecognized_label: - raise ValueError("Unrecognized labe in volume mount string") + try: return cls( fs_source=dictionary["src"], fs_destination=dictionary["dest"], - options=option_list, + options=cls._option_list_from_comma_string(options), settings_entry=settings_entry, source=source, ) @@ -179,6 +158,20 @@ def from_dictionary( except TypeError as exc: raise ValueError("Not a dictionary") from exc + @staticmethod + def _option_list_from_comma_string(options: str) -> List[VolumeMountOption]: + """Build a list of options from a comma delimited string. + + :param options: The comma delimited string + :raises ValueError: When an option is not recognized + """ + if options == "": + return [] + try: + return [getattr(VolumeMountOption, option) for option in options.split(",")] + except AttributeError as exc: + raise ValueError(f"Unrecognized label: {str(exc)}") from exc + class Mode(Enum): """An enum to restrict mode type.""" @@ -444,13 +437,18 @@ def execution_environment_volume_mounts( """ # pylint: disable=unused-argument # pylint: disable=too-many-branches + # pylint: disable=too-many-locals messages: List[LogMessage] = [] exit_messages: List[ExitMessage] = [] settings_entry = entry.settings_file_path(prefix="") volume_mounts: List[VolumeMount] = [] - if entry.value.source in [C.ENVIRONMENT_VARIABLE, C.USER_CLI]: - for mount_str in flatten_list(entry.value.current): + if entry.value.source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI): + if entry.value.source is C.USER_CLI: + mount_strings = flatten_list(entry.value.current) + elif entry.value.source is C.ENVIRONMENT_VARIABLE: + mount_strings = [mount.strip() for mount in entry.value.current.split(";")] + for mount_str in mount_strings: try: volume_mounts.append( VolumeMount.from_string( @@ -467,35 +465,40 @@ def execution_environment_volume_mounts( exit_messages.append(ExitMessage(message=exit_msg)) if entry.cli_parameters: exit_msg = ( - "Try again with format " - + f"'{entry.cli_parameters.short}" - + " ::'." - + " Note: label is optional." + f"Try again with format {entry.cli_parameters.short}" + " ::'." + " Note: label is optional." ) exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) elif entry.value.source is C.USER_CFG: - for list_entry in entry.value.current: - try: - volume_mounts.append( - VolumeMount.from_dictionary( - dictionary=list_entry, - settings_entry=settings_entry, - source=entry.value.source, - ), - ) - except ValueError as exc: - exit_msg = ( - f"The following {settings_entry} entry could not be parsed:" - f" {list_entry} ({entry.value.source.value}), {str(exc)}" - ) - exit_messages.append(ExitMessage(message=exit_msg)) - exit_msg = ( - "The value of execution-environment.volume-mounts" - + "should be a list of dictionaries" - + " and valid keys are 'src', 'dest' and 'label'." - ) - exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) + hint = ( + "The value of execution-environment.volume-mounts should be a list" + " of dictionaries and valid keys are 'src', 'dest' and 'label'." + ) + try: + for list_entry in entry.value.current: + try: + volume_mounts.append( + VolumeMount.from_dictionary( + dictionary=list_entry, + settings_entry=settings_entry, + source=entry.value.source, + ), + ) + except ValueError as exc: + exit_msg = ( + f"The following {settings_entry} entry could not be parsed:" + f" {list_entry} ({entry.value.source.value}), {str(exc)}" + ) + exit_messages.append(ExitMessage(message=exit_msg)) + exit_msg = hint + exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) + except TypeError as exc: + exit_msg = f"{settings_entry} entries could not be parsed." + exit_messages.append(ExitMessage(message=exit_msg)) + exit_msg = hint + exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) exit_msg = ( "The volume mount source path '{fs_source}', configured with '{settings_entry}'" @@ -515,6 +518,10 @@ def execution_environment_volume_mounts( exit_messages.append(ExitMessage(message=exit_msg.format(**asdict(mount)))) extra_mounts.append(mount.to_string()) + # Get out fast if we had any errors + if exit_messages: + return messages, exit_messages + # New mounts were provided if new_mounts: entry.value.current = new_mounts diff --git a/tests/unit/configuration_subsystem/post_processors/__init__.py b/tests/unit/configuration_subsystem/post_processors/__init__.py new file mode 100644 index 000000000..ffd64a605 --- /dev/null +++ b/tests/unit/configuration_subsystem/post_processors/__init__.py @@ -0,0 +1 @@ +"""Post processor related tests.""" diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py new file mode 100644 index 000000000..d83f8a9e5 --- /dev/null +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -0,0 +1,157 @@ +"""Tests for the execution environment volume mount post processor.""" +from dataclasses import dataclass +from itertools import repeat +from typing import Dict +from typing import List +from typing import Optional +from typing import Union + +import pytest + +from ansible_navigator.configuration_subsystem import Constants as C +from ansible_navigator.configuration_subsystem import NavigatorConfiguration +from ansible_navigator.configuration_subsystem.navigator_post_processor import ( + NavigatorPostProcessor, +) + + +@dataclass +class TestData: + """Data structure for EEV post processor tests.""" + + current: Union[bool, str, List, Dict] + source: C + expected: Optional[List[str]] = None + exit_message_substr: str = "" + index: int = 0 + + def __post_init__(self): + """Set the expected if errors are expected.""" + if self.expected is None and self.exit_message_substr: + self.expected = self.current + + def __str__(self): + """Provide a test id. + + :returns: The test id + """ + return f"{self.source}-{self.current}" + + +test_data = ( + TestData(current="", source=C.USER_CLI, exit_message_substr="Could not extract source"), + TestData( + current="abcdef", source=C.USER_CLI, exit_message_substr="Could not extract destination" + ), + TestData( + current=[["/tmp:/tmp"]], + expected=["/tmp:/tmp"], + source=C.USER_CLI, + ), + TestData( + current=[["/tmp:/tmp:Z"]], + expected=["/tmp:/tmp:Z"], + source=C.USER_CLI, + ), + TestData( + current=[["/tmp:/tmp:Y"]], + source=C.USER_CLI, + exit_message_substr="Unrecognized label: Y", + ), + TestData( + current=[["/tmp:/tmp:Z,z"]], + expected=["/tmp:/tmp:Z,z"], + source=C.USER_CLI, + ), + TestData( + current=[["/tmp:/tmp:Z,Y"]], + source=C.USER_CLI, + exit_message_substr="Unrecognized label: Y", + ), + TestData( + current="/tmp:/tmp", + expected=["/tmp:/tmp"], + source=C.ENVIRONMENT_VARIABLE, + ), + TestData( + current="/tmp:/tmp;/tmp:/tmp", + expected=["/tmp:/tmp", "/tmp:/tmp"], + source=C.ENVIRONMENT_VARIABLE, + ), + TestData( + current="/tmp:/tmp ; /tmp:/tmp", + expected=["/tmp:/tmp", "/tmp:/tmp"], + source=C.ENVIRONMENT_VARIABLE, + ), + TestData( + current="/tmp:/tmp:Z ; /tmp:/tmp", + expected=["/tmp:/tmp:Z", "/tmp:/tmp"], + source=C.ENVIRONMENT_VARIABLE, + ), + TestData( + current="/tmp:/tmp:Z,z ; /tmp:/tmp", + expected=["/tmp:/tmp:Z,z", "/tmp:/tmp"], + source=C.ENVIRONMENT_VARIABLE, + ), + TestData( + current="/tmp:/tmp:Z,y ; /tmp:/tmp", + source=C.ENVIRONMENT_VARIABLE, + exit_message_substr="Unrecognized label: y", + ), + TestData(current=True, source=C.USER_CFG, exit_message_substr="could not be parsed"), + TestData( + current={"src": "/tmp", "dest": "/tmp", "label": "Z"}, + source=C.USER_CFG, + exit_message_substr="Must be a list of dictionaries", + ), + TestData( + current=[{"src": "/tmp", "dest": "/tmp", "label": "Z"}], + expected=["/tmp:/tmp:Z"], + source=C.USER_CFG, + ), + TestData( + current=list(repeat({"src": "/tmp", "dest": "/tmp", "label": "Z"}, 4)), + expected=["/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z"], + source=C.USER_CFG, + ), + TestData( + current=[{"src": "/tmp", "dest": "/tmp", "label": "Z,z"}], + expected=["/tmp:/tmp:Z,z"], + source=C.USER_CFG, + ), + TestData( + current=[{"src": "/tmp", "dest": "/tmp", "label": "Z,y"}], + source=C.USER_CFG, + exit_message_substr="Unrecognized label: y", + ), + TestData( + current=[[r"C:\WINNT\System32:/tmp"]], + source=C.USER_CLI, + exit_message_substr="Unrecognized label: /tmp", + ), + TestData( + current=[[r"/WINNT/System32:/tmp"]], + source=C.USER_CLI, + exit_message_substr="does not exist", + ), +) + + +@pytest.mark.parametrize(argnames="data", argvalues=test_data, ids=str) +def test(data: TestData): + """Test the eev post processor. + + :param data: The test data + """ + settings = NavigatorConfiguration + entry = settings.entry("execution_environment_volume_mounts") + + entry.value.current = data.current + entry.value.source = data.source + + _messages, exit_messages = NavigatorPostProcessor().execution_environment_volume_mounts( + entry=entry, config=settings + ) + assert data.expected == entry.value.current + if data.exit_message_substr: + assert data.exit_message_substr in exit_messages[0].message From 047927b01a8eb4ebb86b7974e07d7ae1473dbf4d Mon Sep 17 00:00:00 2001 From: cidrblock Date: Tue, 8 Mar 2022 14:30:03 -0800 Subject: [PATCH 07/18] Order --- .../navigator_post_processor.py | 10 +++++----- .../test_navigator_post_processor.py | 5 +++-- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 5d5f05242..2fe9b8229 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -8,7 +8,6 @@ from dataclasses import asdict from dataclasses import dataclass -from dataclasses import field from enum import Enum from functools import partialmethod from itertools import chain @@ -81,16 +80,17 @@ class VolumeMountOption(Enum): class VolumeMount: """Describes EE volume mounts.""" - fs_destination: str - """The destination file system path in the container for the volume mount""" 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""" + options: List[VolumeMountOption] + """Options for the bind mount""" + settings_entry: str """The name of the settings entry requiring this volume mount""" source: C """The settings source for this volume mount""" - options: List[VolumeMountOption] = field(default_factory=list) - """Options for the bind mount""" def exists(self) -> bool: """Determine if the volume mount source exists.""" diff --git a/tests/unit/configuration_subsystem/test_navigator_post_processor.py b/tests/unit/configuration_subsystem/test_navigator_post_processor.py index cc8bdb398..1d10f87a3 100644 --- a/tests/unit/configuration_subsystem/test_navigator_post_processor.py +++ b/tests/unit/configuration_subsystem/test_navigator_post_processor.py @@ -16,9 +16,10 @@ ( ( VolumeMount( - settings_entry="test_option", - fs_source="/foo", fs_destination="/bar", + fs_source="/foo", + options=[], + settings_entry="test_option", source=Constants.USER_CLI, ), "/foo:/bar", From 7ae382a18fb2f552346a497826aefa02ea373c7b Mon Sep 17 00:00:00 2001 From: cidrblock Date: Tue, 8 Mar 2022 15:47:51 -0800 Subject: [PATCH 08/18] Update tests --- .../configuration_subsystem/configurator.py | 5 +- .../configuration_subsystem/definitions.py | 2 + .../navigator_configuration.py | 1 + .../navigator_post_processor.py | 57 +++++++++++-------- tests/unit/configuration_subsystem/data.py | 6 +- ...est_execution_environment_volume_mounts.py | 24 ++++---- 6 files changed, 55 insertions(+), 40 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/configurator.py b/src/ansible_navigator/configuration_subsystem/configurator.py index 7c616c1c6..078eb40fd 100644 --- a/src/ansible_navigator/configuration_subsystem/configurator.py +++ b/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 diff --git a/src/ansible_navigator/configuration_subsystem/definitions.py b/src/ansible_navigator/configuration_subsystem/definitions.py index 46ef5cdf4..07b5a79ad 100644 --- a/src/ansible_navigator/configuration_subsystem/definitions.py +++ b/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 diff --git a/src/ansible_navigator/configuration_subsystem/navigator_configuration.py b/src/ansible_navigator/configuration_subsystem/navigator_configuration.py index 6d27f581b..94b556e7d 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_configuration.py +++ b/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" diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 2fe9b8229..97efdbbe3 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -13,6 +13,7 @@ from itertools import chain from itertools import repeat from pathlib import Path +from typing import Dict from typing import List from typing import Tuple from typing import Type @@ -131,32 +132,27 @@ def from_string(cls: Type[V], settings_entry: str, source: C, string: str) -> V: ) @classmethod - def from_dictionary(cls: Type[V], settings_entry: str, source: C, dictionary: dict) -> V: + def from_dictionary( + cls: Type[V], + settings_entry: str, + source: C, + dictionary: Dict[str, str], + ) -> V: """Create a ``VolumeMount`` from a dictionary. :param dictionary: The dictionary from which the volume mount will be created :param settings_entry: The settings entry :param source: The source of the string - :raises ValueError: When source or destination are missing, or unrecognized label :returns: A populated volume mount """ - if not isinstance(dictionary, dict): - raise ValueError("Must be a list of dictionaries") - options = dictionary.get("label", "") - - try: - return cls( - fs_source=dictionary["src"], - fs_destination=dictionary["dest"], - options=cls._option_list_from_comma_string(options), - settings_entry=settings_entry, - source=source, - ) - except KeyError as exc: - raise ValueError("Source or destination missing") from exc - except TypeError as exc: - raise ValueError("Not a dictionary") from exc + return cls( + fs_source=dictionary["src"], + fs_destination=dictionary["dest"], + options=cls._option_list_from_comma_string(options), + settings_entry=settings_entry, + source=source, + ) @staticmethod def _option_list_from_comma_string(options: str) -> List[VolumeMountOption]: @@ -438,16 +434,14 @@ def execution_environment_volume_mounts( # pylint: disable=unused-argument # pylint: disable=too-many-branches # pylint: disable=too-many-locals + # pylint: disable=too-many-statements messages: List[LogMessage] = [] exit_messages: List[ExitMessage] = [] settings_entry = entry.settings_file_path(prefix="") volume_mounts: List[VolumeMount] = [] if entry.value.source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI): - if entry.value.source is C.USER_CLI: - mount_strings = flatten_list(entry.value.current) - elif entry.value.source is C.ENVIRONMENT_VARIABLE: - mount_strings = [mount.strip() for mount in entry.value.current.split(";")] + mount_strings = flatten_list(entry.value.current) for mount_str in mount_strings: try: volume_mounts.append( @@ -476,9 +470,22 @@ def execution_environment_volume_mounts( "The value of execution-environment.volume-mounts should be a list" " of dictionaries and valid keys are 'src', 'dest' and 'label'." ) - try: + if isinstance(entry.value.current, list): for list_entry in entry.value.current: try: + # Ensure it is a dictionary + if not isinstance(list_entry, dict): + raise ValueError + + # Ensure only required and optional keys are present + key_variations = [["dest", "src"], ["dest", "label", "src"]] + if sorted(list_entry.keys()) not in key_variations: + raise ValueError + + # Ensure all values are a string + if not all(isinstance(value, str) for value in list_entry.values()): + raise ValueError + volume_mounts.append( VolumeMount.from_dictionary( dictionary=list_entry, @@ -494,11 +501,11 @@ def execution_environment_volume_mounts( exit_messages.append(ExitMessage(message=exit_msg)) exit_msg = hint exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - except TypeError as exc: + else: exit_msg = f"{settings_entry} entries could not be parsed." exit_messages.append(ExitMessage(message=exit_msg)) exit_msg = hint - exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) + exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) exit_msg = ( "The volume mount source path '{fs_source}', configured with '{settings_entry}'" diff --git a/tests/unit/configuration_subsystem/data.py b/tests/unit/configuration_subsystem/data.py index a38c8bbc5..0958d124d 100644 --- a/tests/unit/configuration_subsystem/data.py +++ b/tests/unit/configuration_subsystem/data.py @@ -223,7 +223,11 @@ def cli_data(): ("exec_shell", "false", False), ("execution_environment", "false", False), ("execution_environment_image", "test_image:latest", "test_image:latest"), - ("execution_environment_volume_mounts", "/tmp:/test1:Z", ["/tmp:/test1:Z"]), + ( + "execution_environment_volume_mounts", + "/tmp:/test1:Z;/tmp:/test2:z", + ["/tmp:/test1:Z", "/tmp:/test2:z"], + ), ("help_builder", "false", False), ("help_config", "false", False), ("help_doc", "false", False), diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py index d83f8a9e5..1f8fd2623 100644 --- a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -41,7 +41,9 @@ def __str__(self): test_data = ( TestData(current="", source=C.USER_CLI, exit_message_substr="Could not extract source"), TestData( - current="abcdef", source=C.USER_CLI, exit_message_substr="Could not extract destination" + current="abcdef", + source=C.USER_CLI, + exit_message_substr="Could not extract destination", ), TestData( current=[["/tmp:/tmp"]], @@ -69,32 +71,27 @@ def __str__(self): exit_message_substr="Unrecognized label: Y", ), TestData( - current="/tmp:/tmp", + current=["/tmp:/tmp"], expected=["/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), TestData( - current="/tmp:/tmp;/tmp:/tmp", - expected=["/tmp:/tmp", "/tmp:/tmp"], - source=C.ENVIRONMENT_VARIABLE, - ), - TestData( - current="/tmp:/tmp ; /tmp:/tmp", + current=["/tmp:/tmp", "/tmp:/tmp"], expected=["/tmp:/tmp", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), TestData( - current="/tmp:/tmp:Z ; /tmp:/tmp", + current=["/tmp:/tmp:Z", "/tmp:/tmp"], expected=["/tmp:/tmp:Z", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), TestData( - current="/tmp:/tmp:Z,z ; /tmp:/tmp", + current=["/tmp:/tmp:Z,z", "/tmp:/tmp"], expected=["/tmp:/tmp:Z,z", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), TestData( - current="/tmp:/tmp:Z,y ; /tmp:/tmp", + current=["/tmp:/tmp:Z,y", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, exit_message_substr="Unrecognized label: y", ), @@ -102,7 +99,7 @@ def __str__(self): TestData( current={"src": "/tmp", "dest": "/tmp", "label": "Z"}, source=C.USER_CFG, - exit_message_substr="Must be a list of dictionaries", + exit_message_substr="could not be parsed", ), TestData( current=[{"src": "/tmp", "dest": "/tmp", "label": "Z"}], @@ -150,7 +147,8 @@ def test(data: TestData): entry.value.source = data.source _messages, exit_messages = NavigatorPostProcessor().execution_environment_volume_mounts( - entry=entry, config=settings + entry=entry, + config=settings, ) assert data.expected == entry.value.current if data.exit_message_substr: From d7e6eddfa7ac3adbdd2a82d085958cd4f9261c35 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 07:42:28 -0800 Subject: [PATCH 09/18] Do it all in post_init --- .../navigator_post_processor.py | 243 ++++++++---------- ...est_execution_environment_volume_mounts.py | 67 +++-- .../test_navigator_post_processor.py | 27 +- 3 files changed, 152 insertions(+), 185 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 97efdbbe3..8ffd238b3 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -6,17 +6,15 @@ import shlex import shutil -from dataclasses import asdict 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 Dict from typing import List from typing import Tuple -from typing import Type from typing import TypeVar from ..utils.functions import ExitMessage @@ -77,6 +75,10 @@ class VolumeMountOption(Enum): V = TypeVar("V", bound="VolumeMount") # pylint: disable=invalid-name +class VolumeMountError(Exception): + """Custom exception raised when building VolumeMounts.""" + + @dataclass class VolumeMount: """Describes EE volume mounts.""" @@ -85,89 +87,66 @@ class VolumeMount: """The source file system path of the volume mount""" fs_destination: str """The destination file system path in the container for the volume mount""" - options: List[VolumeMountOption] - """Options for the bind mount""" - settings_entry: str """The name of the settings entry requiring this volume mount""" source: C """The settings source for this volume mount""" + options_string: str = "" + """Comma delimited options""" + _options: List[VolumeMountOption] = field(default_factory=list) + """Options for the bind mount""" + + def __post_init__(self): + """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(self.options_string, str): + if not self.options_string == "": + options = [] + option_values = [o.value for o in VolumeMountOption] + for option in self.options_string.split(","): + if option not in option_values: + errors.append( + f"Unrecognized label: '{option}'," + " available labels include {oxfordcomma(option_values)}.", + ) + else: + options.append(VolumeMountOption(option)) + self._options = options + else: + errors.append(f"Labels: '{self.options_string}' is not a string.") - def exists(self) -> bool: - """Determine if the volume mount source exists.""" - return Path(self.fs_source).exists() + 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.fs_source}:{self.fs_destination}" - if self.options: - joined_opts = ",".join(o.value for o in self.options) + if self._options: + joined_opts = ",".join(o.value for o in self._options) out += f":{joined_opts}" return out - @classmethod - def from_string(cls: Type[V], settings_entry: str, source: C, string: str) -> V: - """Create a ``VolumeMount`` from a string. - - :param settings_entry: The settings entry - :param source: The source of the string - :param string: The string from which the volume mount will be created - :raises ValueError: When source or destination are missing, or unrecognized label - :returns: A populated volume mount - """ - fs_source, fs_destination, options, *left_overs = chain(string.split(":"), repeat("", 3)) - - if any(left_overs): - raise ValueError("Not formatted correctly") - if not fs_source: - raise ValueError("Could not extract source from string") - if not fs_destination: - raise ValueError("Could not extract destination from string") - return cls( - fs_source=fs_source, - fs_destination=fs_destination, - options=cls._option_list_from_comma_string(options), - settings_entry=settings_entry, - source=source, - ) - - @classmethod - def from_dictionary( - cls: Type[V], - settings_entry: str, - source: C, - dictionary: Dict[str, str], - ) -> V: - """Create a ``VolumeMount`` from a dictionary. - - :param dictionary: The dictionary from which the volume mount will be created - :param settings_entry: The settings entry - :param source: The source of the string - :returns: A populated volume mount - """ - options = dictionary.get("label", "") - return cls( - fs_source=dictionary["src"], - fs_destination=dictionary["dest"], - options=cls._option_list_from_comma_string(options), - settings_entry=settings_entry, - source=source, - ) - - @staticmethod - def _option_list_from_comma_string(options: str) -> List[VolumeMountOption]: - """Build a list of options from a comma delimited string. - - :param options: The comma delimited string - :raises ValueError: When an option is not recognized - """ - if options == "": - return [] - try: - return [getattr(VolumeMountOption, option) for option in options.split(",")] - except AttributeError as exc: - raise ValueError(f"Unrecognized label: {str(exc)}") from exc - class Mode(Enum): """An enum to restrict mode type.""" @@ -434,111 +413,91 @@ def execution_environment_volume_mounts( # pylint: disable=unused-argument # pylint: disable=too-many-branches # pylint: disable=too-many-locals - # pylint: disable=too-many-statements messages: List[LogMessage] = [] exit_messages: List[ExitMessage] = [] - settings_entry = entry.settings_file_path(prefix="") + entry_name = entry.settings_file_path(prefix="") + entry_source = entry.value.source volume_mounts: List[VolumeMount] = [] - if entry.value.source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI): + + if entry_source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI): + hint = ( + "Try again with format ::'." + " 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 = ( + f"The following {entry_name} entry could not be parsed:" + f" {mount_str} ({entry_source.value})" + ) + exit_messages.append(ExitMessage(message=exit_msg)) + exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT)) + try: volume_mounts.append( - VolumeMount.from_string( - settings_entry=settings_entry, - source=entry.value.source, - string=mount_str, + VolumeMount( + fs_source=src, + fs_destination=dest, + options_string=labels, + settings_entry=entry_name, + source=entry_source, ), ) - except ValueError as exc: + except VolumeMountError as exc: exit_msg = ( - f"The following {settings_entry} entry could not be parsed:" - f" {mount_str} ({entry.value.source.value}), {str(exc)}" + 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_msg = ( - f"Try again with format {entry.cli_parameters.short}" - " ::'." - " Note: label is optional." - ) - exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) + 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 isinstance(entry.value.current, list): - for list_entry in entry.value.current: + 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: - # Ensure it is a dictionary - if not isinstance(list_entry, dict): - raise ValueError - - # Ensure only required and optional keys are present - key_variations = [["dest", "src"], ["dest", "label", "src"]] - if sorted(list_entry.keys()) not in key_variations: - raise ValueError - - # Ensure all values are a string - if not all(isinstance(value, str) for value in list_entry.values()): - raise ValueError - volume_mounts.append( - VolumeMount.from_dictionary( - dictionary=list_entry, - settings_entry=settings_entry, - source=entry.value.source, + 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 ValueError as exc: + except (AttributeError, VolumeMountError) as exc: exit_msg = ( - f"The following {settings_entry} entry could not be parsed:" - f" {list_entry} ({entry.value.source.value}), {str(exc)}" + 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)) - exit_msg = hint - exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - else: - exit_msg = f"{settings_entry} entries could not be parsed." - exit_messages.append(ExitMessage(message=exit_msg)) - exit_msg = hint - exit_messages.append(ExitMessage(message=exit_msg, prefix=ExitPrefix.HINT)) - - exit_msg = ( - "The volume mount source path '{fs_source}', configured with '{settings_entry}'" - " ({source.value}), does not exist." - ) - # Check any new volume mounts first - new_mounts = [] - for mount in volume_mounts: - if not mount.exists(): - exit_messages.append(ExitMessage(message=exit_msg.format(**asdict(mount)))) - new_mounts.append(mount.to_string()) - - # Check extra mounts next - extra_mounts = [] - for mount in self.extra_volume_mounts: - if not mount.exists(): - exit_messages.append(ExitMessage(message=exit_msg.format(**asdict(mount)))) - extra_mounts.append(mount.to_string()) + exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT)) # Get out fast if we had any errors if exit_messages: return messages, exit_messages # New mounts were provided - if new_mounts: - entry.value.current = new_mounts + if volume_mounts: + entry.value.current = [v.to_string() for v in volume_mounts] # Extra mounts were requested, these get added to either # new_mounts, C.PREVIOUS_CLI or C.NOT_SET - if extra_mounts: + if self.extra_volume_mounts: if not isinstance(entry.value.current, list): entry.value.current = [] - entry.value.current.extend(exit_messages) + entry.value.current.extend([v.to_string() for v in self.extra_volume_mounts]) return messages, exit_messages diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py index 1f8fd2623..33a6dabb4 100644 --- a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -16,7 +16,7 @@ @dataclass -class TestData: +class Scenario: """Data structure for EEV post processor tests.""" current: Union[bool, str, List, Dict] @@ -39,94 +39,105 @@ def __str__(self): test_data = ( - TestData(current="", source=C.USER_CLI, exit_message_substr="Could not extract source"), - TestData( + Scenario( + current="", + source=C.USER_CLI, + exit_message_substr="Source not provided. Destination not provided", + ), + Scenario( current="abcdef", source=C.USER_CLI, - exit_message_substr="Could not extract destination", + exit_message_substr="Source: 'abcdef' does not exist. Destination not provided.", ), - TestData( + Scenario( current=[["/tmp:/tmp"]], expected=["/tmp:/tmp"], source=C.USER_CLI, ), - TestData( + Scenario( current=[["/tmp:/tmp:Z"]], expected=["/tmp:/tmp:Z"], source=C.USER_CLI, ), - TestData( + Scenario( current=[["/tmp:/tmp:Y"]], source=C.USER_CLI, - exit_message_substr="Unrecognized label: Y", + exit_message_substr="Unrecognized label: 'Y'", ), - TestData( + Scenario( current=[["/tmp:/tmp:Z,z"]], expected=["/tmp:/tmp:Z,z"], source=C.USER_CLI, ), - TestData( + Scenario( current=[["/tmp:/tmp:Z,Y"]], source=C.USER_CLI, - exit_message_substr="Unrecognized label: Y", + exit_message_substr="Unrecognized label: 'Y'", ), - TestData( + Scenario( current=["/tmp:/tmp"], expected=["/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), - TestData( + Scenario( current=["/tmp:/tmp", "/tmp:/tmp"], expected=["/tmp:/tmp", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), - TestData( + Scenario( current=["/tmp:/tmp:Z", "/tmp:/tmp"], expected=["/tmp:/tmp:Z", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), - TestData( + Scenario( current=["/tmp:/tmp:Z,z", "/tmp:/tmp"], expected=["/tmp:/tmp:Z,z", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), - TestData( + Scenario( current=["/tmp:/tmp:Z,y", "/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, - exit_message_substr="Unrecognized label: y", + exit_message_substr="Unrecognized label: 'y'", ), - TestData(current=True, source=C.USER_CFG, exit_message_substr="could not be parsed"), - TestData( + Scenario(current=True, source=C.USER_CFG, exit_message_substr="could not be parsed"), + Scenario(current=[True], source=C.USER_CFG, exit_message_substr="could not be parsed"), + Scenario(current=[[True]], source=C.USER_CFG, exit_message_substr="could not be parsed"), + Scenario( current={"src": "/tmp", "dest": "/tmp", "label": "Z"}, source=C.USER_CFG, exit_message_substr="could not be parsed", ), - TestData( + Scenario( + current=[{"my_src": "/tmp", "my_dest": "/tmp", "my_label": "Z"}], + source=C.USER_CFG, + exit_message_substr="could not be parsed", + ), + Scenario( current=[{"src": "/tmp", "dest": "/tmp", "label": "Z"}], expected=["/tmp:/tmp:Z"], source=C.USER_CFG, ), - TestData( + Scenario( current=list(repeat({"src": "/tmp", "dest": "/tmp", "label": "Z"}, 4)), expected=["/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z"], source=C.USER_CFG, ), - TestData( + Scenario( current=[{"src": "/tmp", "dest": "/tmp", "label": "Z,z"}], expected=["/tmp:/tmp:Z,z"], source=C.USER_CFG, ), - TestData( + Scenario( current=[{"src": "/tmp", "dest": "/tmp", "label": "Z,y"}], source=C.USER_CFG, - exit_message_substr="Unrecognized label: y", + exit_message_substr="Unrecognized label: 'y'", ), - TestData( + Scenario( current=[[r"C:\WINNT\System32:/tmp"]], source=C.USER_CLI, - exit_message_substr="Unrecognized label: /tmp", + exit_message_substr="Unrecognized label: '/tmp'", ), - TestData( + Scenario( current=[[r"/WINNT/System32:/tmp"]], source=C.USER_CLI, exit_message_substr="does not exist", @@ -135,7 +146,7 @@ def __str__(self): @pytest.mark.parametrize(argnames="data", argvalues=test_data, ids=str) -def test(data: TestData): +def test(data: Scenario): """Test the eev post processor. :param data: The test data diff --git a/tests/unit/configuration_subsystem/test_navigator_post_processor.py b/tests/unit/configuration_subsystem/test_navigator_post_processor.py index 1d10f87a3..09238cf04 100644 --- a/tests/unit/configuration_subsystem/test_navigator_post_processor.py +++ b/tests/unit/configuration_subsystem/test_navigator_post_processor.py @@ -6,9 +6,6 @@ from ansible_navigator.configuration_subsystem.navigator_post_processor import ( VolumeMount, ) -from ansible_navigator.configuration_subsystem.navigator_post_processor import ( - VolumeMountOption, -) @pytest.mark.parametrize( @@ -17,42 +14,42 @@ ( VolumeMount( fs_destination="/bar", - fs_source="/foo", - options=[], + fs_source="/tmp", + options_string="", settings_entry="test_option", source=Constants.USER_CLI, ), - "/foo:/bar", + "/tmp:/bar", ), ( VolumeMount( settings_entry="test_option", - fs_source="/foo", + fs_source="/tmp", fs_destination="/bar", - options=[VolumeMountOption.z], + options_string="z", source=Constants.USER_CLI, ), - "/foo:/bar:z", + "/tmp:/bar:z", ), ( VolumeMount( settings_entry="test_option", - fs_source="/foo", + fs_source="/tmp", fs_destination="/bar", - options=[VolumeMountOption.z, VolumeMountOption.Z], + options_string="z,Z", source=Constants.USER_CLI, ), - "/foo:/bar:z,Z", + "/tmp:/bar:z,Z", ), ( VolumeMount( settings_entry="test_option", - fs_source="/foo", + fs_source="/tmp", fs_destination="/bar", - options=[], + options_string="", source=Constants.USER_CLI, ), - "/foo:/bar", + "/tmp:/bar", ), ), ids=( From aec6a42b915cbadb4e8e310aca1d8d7ea5a10059 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 08:10:28 -0800 Subject: [PATCH 10/18] One more test --- .../test_execution_environment_volume_mounts.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py index 33a6dabb4..6182e3bd9 100644 --- a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -117,6 +117,15 @@ def __str__(self): expected=["/tmp:/tmp:Z"], source=C.USER_CFG, ), + Scenario( + current=[{"src": True, "dest": False, "label": 42}], + source=C.USER_CFG, + exit_message_substr=( + "Source: 'True' is not a string." + " Destination: 'False is not a string." + " Labels: '42' is not a string." + ), + ), Scenario( current=list(repeat({"src": "/tmp", "dest": "/tmp", "label": "Z"}, 4)), expected=["/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z"], From 133975ed6bb7c5d0ec326791aaeed15fbf99f8cd Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 08:13:46 -0800 Subject: [PATCH 11/18] Sort entries --- ...est_execution_environment_volume_mounts.py | 41 ++++++++++++------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py index 6182e3bd9..eef7004db 100644 --- a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -41,13 +41,13 @@ def __str__(self): test_data = ( Scenario( current="", - source=C.USER_CLI, exit_message_substr="Source not provided. Destination not provided", + source=C.USER_CLI, ), Scenario( current="abcdef", - source=C.USER_CLI, exit_message_substr="Source: 'abcdef' does not exist. Destination not provided.", + source=C.USER_CLI, ), Scenario( current=[["/tmp:/tmp"]], @@ -61,8 +61,8 @@ def __str__(self): ), Scenario( current=[["/tmp:/tmp:Y"]], - source=C.USER_CLI, exit_message_substr="Unrecognized label: 'Y'", + source=C.USER_CLI, ), Scenario( current=[["/tmp:/tmp:Z,z"]], @@ -71,8 +71,8 @@ def __str__(self): ), Scenario( current=[["/tmp:/tmp:Z,Y"]], - source=C.USER_CLI, exit_message_substr="Unrecognized label: 'Y'", + source=C.USER_CLI, ), Scenario( current=["/tmp:/tmp"], @@ -96,21 +96,33 @@ def __str__(self): ), Scenario( current=["/tmp:/tmp:Z,y", "/tmp:/tmp"], - source=C.ENVIRONMENT_VARIABLE, exit_message_substr="Unrecognized label: 'y'", + source=C.ENVIRONMENT_VARIABLE, ), - Scenario(current=True, source=C.USER_CFG, exit_message_substr="could not be parsed"), - Scenario(current=[True], source=C.USER_CFG, exit_message_substr="could not be parsed"), - Scenario(current=[[True]], source=C.USER_CFG, exit_message_substr="could not be parsed"), Scenario( - current={"src": "/tmp", "dest": "/tmp", "label": "Z"}, + current=True, + exit_message_substr="could not be parsed", source=C.USER_CFG, + ), + Scenario( + current=[True], exit_message_substr="could not be parsed", + source=C.USER_CFG, ), Scenario( - current=[{"my_src": "/tmp", "my_dest": "/tmp", "my_label": "Z"}], + current=[[True]], + exit_message_substr="could not be parsed", source=C.USER_CFG, + ), + Scenario( + current={"src": "/tmp", "dest": "/tmp", "label": "Z"}, exit_message_substr="could not be parsed", + source=C.USER_CFG, + ), + Scenario( + current=[{"my_src": "/tmp", "my_dest": "/tmp", "my_label": "Z"}], + exit_message_substr="could not be parsed", + source=C.USER_CFG, ), Scenario( current=[{"src": "/tmp", "dest": "/tmp", "label": "Z"}], @@ -119,12 +131,13 @@ def __str__(self): ), Scenario( current=[{"src": True, "dest": False, "label": 42}], - source=C.USER_CFG, exit_message_substr=( "Source: 'True' is not a string." " Destination: 'False is not a string." " Labels: '42' is not a string." ), + source=C.USER_CFG, + ), Scenario( current=list(repeat({"src": "/tmp", "dest": "/tmp", "label": "Z"}, 4)), @@ -138,18 +151,18 @@ def __str__(self): ), Scenario( current=[{"src": "/tmp", "dest": "/tmp", "label": "Z,y"}], - source=C.USER_CFG, exit_message_substr="Unrecognized label: 'y'", + source=C.USER_CFG, ), Scenario( current=[[r"C:\WINNT\System32:/tmp"]], - source=C.USER_CLI, exit_message_substr="Unrecognized label: '/tmp'", + source=C.USER_CLI, ), Scenario( current=[[r"/WINNT/System32:/tmp"]], - source=C.USER_CLI, exit_message_substr="does not exist", + source=C.USER_CLI, ), ) From 899ad4af5454592cd90ef86c31a9a0f872a76f34 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 08:59:33 -0800 Subject: [PATCH 12/18] Format --- .../post_processors/test_execution_environment_volume_mounts.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py index eef7004db..62343d92a 100644 --- a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -137,7 +137,6 @@ def __str__(self): " Labels: '42' is not a string." ), source=C.USER_CFG, - ), Scenario( current=list(repeat({"src": "/tmp", "dest": "/tmp", "label": "Z"}, 4)), From 7bb17f1be3eac92dc8fd80c08de424b348dcf5ad Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 09:25:12 -0800 Subject: [PATCH 13/18] Add changelog --- docs/changelog-fragments.d/1061.breaking.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 docs/changelog-fragments.d/1061.breaking.md diff --git a/docs/changelog-fragments.d/1061.breaking.md b/docs/changelog-fragments.d/1061.breaking.md new file mode 100644 index 000000000..cd2f9964c --- /dev/null +++ b/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` From 31e3bcac3c91d3f8aee3da6a401c06cb43221ee7 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 14:15:07 -0800 Subject: [PATCH 14/18] More --- docs/changelog-fragments.d/1061.breaking.md | 4 +- .../navigator_post_processor.py | 43 +++++++++++-------- ...est_execution_environment_volume_mounts.py | 14 ++++-- 3 files changed, 36 insertions(+), 25 deletions(-) diff --git a/docs/changelog-fragments.d/1061.breaking.md b/docs/changelog-fragments.d/1061.breaking.md index cd2f9964c..c930c723d 100644 --- a/docs/changelog-fragments.d/1061.breaking.md +++ b/docs/changelog-fragments.d/1061.breaking.md @@ -1,5 +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'`` +now a semi-colon delimited string. e.g. +`ANSIBLE_NAVIGATOR_EXECUTION_ENVIRONMENT_VOLUME_MOUNTS='\p1:\p1;\p2:p2'` -- by {user}`cidrblock` diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 7f6ae6e86..734c9af36 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -6,6 +6,7 @@ import shlex import shutil +from dataclasses import InitVar from dataclasses import dataclass from dataclasses import field from enum import Enum @@ -14,6 +15,7 @@ from itertools import repeat from pathlib import Path from typing import List +from typing import Set from typing import Tuple from typing import TypeVar @@ -79,7 +81,7 @@ class VolumeMountError(Exception): """Custom exception raised when building VolumeMounts.""" -@dataclass +@dataclass(frozen=True) class VolumeMount: """Describes EE volume mounts.""" @@ -89,14 +91,14 @@ class VolumeMount: """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 + source: C = field(compare=False) """The settings source for this volume mount""" - options_string: str = "" + options_string: InitVar[str] """Comma delimited options""" - _options: List[VolumeMountOption] = field(default_factory=list) + options: Tuple[VolumeMountOption, ...] = () """Options for the bind mount""" - def __post_init__(self): + def __post_init__(self, options_string): """Post process the ``VolumeMount`` and perform sanity checks. :raises VolumeMountError: When a viable VolumeMount cannot be created @@ -117,24 +119,27 @@ def __post_init__(self): if self.fs_destination == "": errors.append("Destination not provided.") else: - errors.append(f"Destination: '{self.fs_destination} is not a string.") + errors.append(f"Destination: '{self.fs_destination}' is not a string.") # Validate and populate _options - if isinstance(self.options_string, str): - if not self.options_string == "": + if isinstance(options_string, str): + if not options_string == "": options = [] option_values = [o.value for o in VolumeMountOption] - for option in self.options_string.split(","): + for option in sorted(options_string.split(",")): if option not in option_values: errors.append( f"Unrecognized label: '{option}'," - " available labels include {oxfordcomma(option_values)}.", + f" available labels include" + f" {oxfordcomma(option_values, condition='and/or')}.", ) else: options.append(VolumeMountOption(option)) - self._options = options + 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: '{self.options_string}' is not a string.") + errors.append(f"Labels: '{options_string}' is not a string.") if errors: raise VolumeMountError(" ".join(errors)) @@ -142,8 +147,8 @@ def __post_init__(self): def to_string(self) -> str: """Render the volume mount in a way that (docker|podman) understands.""" out = f"{self.fs_source}:{self.fs_destination}" - if self._options: - joined_opts = ",".join(o.value for o in self._options) + if self.options: + joined_opts = ",".join(o.value for o in self.options) out += f":{joined_opts}" return out @@ -418,7 +423,7 @@ def execution_environment_volume_mounts( exit_messages: List[ExitMessage] = [] entry_name = entry.settings_file_path(prefix="") entry_source = entry.value.source - volume_mounts: List[VolumeMount] = [] + volume_mounts: Set[VolumeMount] = set() if entry_source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI): hint = ( @@ -437,7 +442,7 @@ def execution_environment_volume_mounts( exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT)) try: - volume_mounts.append( + volume_mounts.add( VolumeMount( fs_source=src, fs_destination=dest, @@ -467,7 +472,7 @@ def execution_environment_volume_mounts( else: for volume_mount in entry.value.current: try: - volume_mounts.append( + volume_mounts.add( VolumeMount( fs_source=volume_mount.get("src"), fs_destination=volume_mount.get("dest"), @@ -496,8 +501,8 @@ def execution_environment_volume_mounts( # new_mounts, C.PREVIOUS_CLI or C.NOT_SET if self.extra_volume_mounts: if not isinstance(entry.value.current, list): - entry.value.current = [] - entry.value.current.extend([v.to_string() for v in self.extra_volume_mounts]) + entry.value.current = set() + entry.value.current.extend(v.to_string() for v in self.extra_volume_mounts) return messages, exit_messages diff --git a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py index 62343d92a..a18802404 100644 --- a/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py +++ b/tests/unit/configuration_subsystem/post_processors/test_execution_environment_volume_mounts.py @@ -1,4 +1,5 @@ """Tests for the execution environment volume mount post processor.""" +from collections.abc import Iterable from dataclasses import dataclass from itertools import repeat from typing import Dict @@ -81,7 +82,7 @@ def __str__(self): ), Scenario( current=["/tmp:/tmp", "/tmp:/tmp"], - expected=["/tmp:/tmp", "/tmp:/tmp"], + expected=["/tmp:/tmp"], source=C.ENVIRONMENT_VARIABLE, ), Scenario( @@ -133,14 +134,14 @@ def __str__(self): current=[{"src": True, "dest": False, "label": 42}], exit_message_substr=( "Source: 'True' is not a string." - " Destination: 'False is not a string." + " Destination: 'False' is not a string." " Labels: '42' is not a string." ), source=C.USER_CFG, ), Scenario( current=list(repeat({"src": "/tmp", "dest": "/tmp", "label": "Z"}, 4)), - expected=["/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z", "/tmp:/tmp:Z"], + expected=["/tmp:/tmp:Z"], source=C.USER_CFG, ), Scenario( @@ -182,6 +183,11 @@ def test(data: Scenario): entry=entry, config=settings, ) - assert data.expected == entry.value.current + if isinstance(data.expected, Iterable) and isinstance(entry.value.current, Iterable): + assert sorted(data.expected) == sorted(entry.value.current) + else: + # A bool can't be sorted + assert data.expected == entry.value.current + if data.exit_message_substr: assert data.exit_message_substr in exit_messages[0].message From 3e9b7d16f592791807fe7f7694a070cc112eaa93 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 14:24:56 -0800 Subject: [PATCH 15/18] Docs fix --- docs/conf.py | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/conf.py b/docs/conf.py index fc30934d3..3c4458876 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -122,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"), From 5271d5866232771c9616fc32b1af2afe5a5888f6 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 15:29:54 -0800 Subject: [PATCH 16/18] Don't sort --- .../configuration_subsystem/navigator_post_processor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 734c9af36..083032c09 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -126,7 +126,7 @@ def __post_init__(self, options_string): if not options_string == "": options = [] option_values = [o.value for o in VolumeMountOption] - for option in sorted(options_string.split(",")): + for option in options_string.split(","): if option not in option_values: errors.append( f"Unrecognized label: '{option}'," From db680694bbe66121dac11b4f296a6445ad62c79a Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 16:02:20 -0800 Subject: [PATCH 17/18] Lists preserver order --- .../navigator_post_processor.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index 083032c09..d7b696697 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -423,7 +423,7 @@ def execution_environment_volume_mounts( exit_messages: List[ExitMessage] = [] entry_name = entry.settings_file_path(prefix="") entry_source = entry.value.source - volume_mounts: Set[VolumeMount] = set() + volume_mounts: List[VolumeMount] = [] if entry_source in (C.ENVIRONMENT_VARIABLE, C.USER_CLI): hint = ( @@ -442,7 +442,7 @@ def execution_environment_volume_mounts( exit_messages.append(ExitMessage(message=hint, prefix=ExitPrefix.HINT)) try: - volume_mounts.add( + volume_mounts.append( VolumeMount( fs_source=src, fs_destination=dest, @@ -472,7 +472,7 @@ def execution_environment_volume_mounts( else: for volume_mount in entry.value.current: try: - volume_mounts.add( + volume_mounts.append( VolumeMount( fs_source=volume_mount.get("src"), fs_destination=volume_mount.get("dest"), @@ -504,6 +504,10 @@ def execution_environment_volume_mounts( entry.value.current = set() entry.value.current.extend(v.to_string() for v in self.extra_volume_mounts) + # 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) + return messages, exit_messages @staticmethod From 23d091ca48a8a2e987af2b4d8671a9dbb88d89c4 Mon Sep 17 00:00:00 2001 From: cidrblock Date: Wed, 9 Mar 2022 16:12:07 -0800 Subject: [PATCH 18/18] No set --- .../configuration_subsystem/navigator_post_processor.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py index d7b696697..e7f623b62 100644 --- a/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py +++ b/src/ansible_navigator/configuration_subsystem/navigator_post_processor.py @@ -15,7 +15,6 @@ from itertools import repeat from pathlib import Path from typing import List -from typing import Set from typing import Tuple from typing import TypeVar