Add isolation options (#313)

* Add isolation options

* Cleanup logic

* Cleanup logic

* Add some tests

* No fail for ansible_config

* Additional tests

* Update test

* Update tests

Author: cidrblock

.config/dictionary.txt

@@ -1,3 +1,4 @@
+acfg
 argnames
 argvalues
 bindep
@@ -27,6 +28,7 @@ reqs
 sessionstart
 setenv
 treemaker
+unisolated
 usefixtures
 xdist
 xmltodict

src/ansible_dev_environment/arg_parser.py

@@ -203,6 +203,20 @@ def parse() -> argparse.Namespace:
         help="Install seed packages inside the virtual environment (ansible-dev-tools).",
     )
 
+    install.add_argument(
+        "--im",
+        "--isolation-mode",
+        dest="isolation_mode",
+        help=(
+            "Isolation mode to use. restrictive: Exit if collections are found in ansible home or the system collection directory. "
+            " cfg: Update or add an ansible.cfg file in the current working directory to isolate the workspace"
+            " none: No isolation, not recommended."
+        ),
+        default="restrictive",
+        choices=["restrictive", "cfg", "none"],
+        type=str,
+    )
+
     _uninstall = subparsers.add_parser(
         "uninstall",
         formatter_class=CustomHelpFormatter,

src/ansible_dev_environment/cli.py

@@ -13,6 +13,8 @@
 
 from .arg_parser import parse
 from .config import Config
+from .definitions import COLLECTIONS_PATH as CP
+from .definitions import AnsibleCfg
 from .output import Output
 from .utils import TermFeatures
 
@@ -30,6 +32,10 @@ def __init__(self) -> None:
         self.config: Config
         self.output: Output
         self.term_features: TermFeatures
+        self.acfg_cwd = AnsibleCfg(path=Path("./ansible.cfg"))
+        self.acfg_home = AnsibleCfg(path=Path("~/.ansible.cfg").expanduser().resolve())
+        self.acfg_system = AnsibleCfg(path=Path("/etc/ansible/ansible.cfg"))
+        self.acfg_trusted: Path | None
 
     def parse_args(self) -> None:
         """Parse the command line arguments."""
@@ -92,8 +98,92 @@ def args_sanity(self) -> None:
             err = "Editable can not be used with a requirements file."
             self.output.critical(err)
 
-    def ensure_isolated(self) -> None:
-        """Ensure the environment is isolated."""
+    def isolation_check(self) -> bool:
+        """Check the environment for isolation.
+
+        Returns:
+            True if ade can continue, false otherwise.
+        """
+        if not hasattr(self.args, "isolation_mode"):
+            return True
+        if self.args.isolation_mode == "restrictive":
+            return self.isolation_restrictive()
+        if self.args.isolation_mode == "cfg":
+            return self.isolation_cfg()
+        if self.args.isolation_mode == "none":
+            return self.isolation_none()
+        self.acfg_trusted = None
+        return False
+
+    def isolation_cfg(self) -> bool:
+        """Ensure the environment is isolated using cfg isolation.
+
+        Returns:
+            True if ade can continue, false otherwise.
+        """
+        if os.environ.get("ANSIBLE_CONFIG"):
+            err = "ANSIBLE_CONFIG is set"
+            self.output.error(err)
+            hint = "Run `unset ANSIBLE_CONFIG` to unset it using cfg isolation mode."
+            self.output.hint(hint)
+            self.acfg_trusted = None
+            return False
+
+        if self.acfg_cwd.exists:
+            if self.acfg_cwd.collections_path_is_dot:
+                msg = f"{self.acfg_cwd.path} has '{CP}' which isolates this workspace."
+                self.output.info(msg)
+            else:
+                self.acfg_cwd.set_or_update_collections_path()
+                msg = f"{self.acfg_cwd.path} updated with '{CP}' to isolate this workspace."
+                self.output.warning(msg)
+            self.acfg_trusted = self.acfg_cwd.path
+            return True
+
+        if self.acfg_home.exists:
+            if self.acfg_home.collections_path_is_dot:
+                msg = f"{self.acfg_home.path} has '{CP}' which isolates this and all workspaces."
+                self.output.info(msg)
+            else:
+                self.acfg_home.set_or_update_collections_path()
+                msg = (
+                    f"{self.acfg_home.path} updated with '{CP}' to isolate this and all workspaces."
+                )
+                self.output.warning(msg)
+            self.acfg_trusted = self.acfg_home.path
+            return True
+
+        if self.acfg_system.exists and self.acfg_system.collections_path_is_dot:
+            msg = f"{self.acfg_system.path} has '{CP}' which isolates this and all workspaces."
+            self.output.info(msg)
+            self.acfg_trusted = self.acfg_system.path
+            return True
+
+        self.acfg_cwd.author_new()
+        msg = f"{self.acfg_cwd.path} created with '{CP}' to isolate this workspace."
+        self.output.info(msg)
+        self.acfg_trusted = self.acfg_cwd.path
+        return True
+
+    def isolation_none(self) -> bool:
+        """No isolation.
+
+        Returns:
+            True if ade can continue, false otherwise.
+        """
+        self.output.warning(
+            "An unisolated development environment can cause issues with conflicting dependency"
+            " versions and the use of incompatible collections.",
+        )
+        self.acfg_trusted = None
+        return True
+
+    def isolation_restrictive(self) -> bool:
+        """Ensure the environment is isolated.
+
+        Returns:
+            True if ade can continue, false otherwise.
+        """
         env_vars = os.environ
         errored = False
         if "ANSIBLE_COLLECTIONS_PATHS" in env_vars:
@@ -127,11 +217,11 @@ def ensure_isolated(self) -> None:
             hint = "Run `sudo rm -rf /usr/share/ansible/collections` to remove them."
             self.output.hint(hint)
             errored = True
-
         if errored:
             err = "The development environment is not isolated, please resolve the above errors."
-
-            self.output.critical(err)
+            self.output.warning(err)
+            return False
+        return True
 
     def run(self) -> None:
         """Run the application."""
@@ -145,9 +235,9 @@ def run(self) -> None:
         subcommand_cls = getattr(subcommands, self.config.args.subcommand.capitalize())
         subcommand = subcommand_cls(config=self.config, output=self.output)
         subcommand.run()
-        self._exit()
+        self.exit()
 
-    def _exit(self) -> None:
+    def exit(self) -> None:
         """Exit the application setting the return code."""
         if self.output.call_count["error"]:
             sys.exit(1)
@@ -171,6 +261,7 @@ def main(*, dry: bool = False) -> None:
         cli.output.warning(str(warn.message))
     warnings.resetwarnings()
     cli.args_sanity()
-    cli.ensure_isolated()
+    if not cli.isolation_check():
+        cli.exit()
     if not dry:
         cli.run()

src/ansible_dev_environment/definitions.py

@@ -0,0 +1,68 @@
+"""Some common definitions."""
+
+from __future__ import annotations
+
+from configparser import ConfigParser
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+COLLECTIONS_PATH = "collections_path = ."
+
+
+@dataclass
+class AnsibleCfg:
+    """ansible.cfg file abstraction.
+
+    Attributes:
+        path: Path to the ansible.cfg file.
+    """
+
+    path: Path
+
+    @property
+    def exists(self) -> bool:
+        """Check if the ansible.cfg file exists."""
+        return self.path.exists()
+
+    @property
+    def collections_path_is_dot(self) -> bool:
+        """Check if the collection path is a dot.
+
+        Returns:
+            bool: True if the collection path is a dot.
+        """
+        config = ConfigParser()
+        config.read(self.path)
+        return config.get("defaults", "collections_path", fallback=None) == "."
+
+    def set_or_update_collections_path(self) -> None:
+        """Set or update the collection path in the ansible.cfg file.
+
+        The configparser doesn't preserve comments, so we need to read the file
+        and write it back with the new collection path.
+        """
+        contents = self.path.read_text().splitlines()
+
+        if "[defaults]" not in contents:
+            contents.insert(0, "[defaults]")
+
+        idx = [i for i, line in enumerate(contents) if line.startswith("collections_path")]
+
+        if idx:
+            contents[idx[0]] = COLLECTIONS_PATH
+        else:
+            insert_at = contents.index("[defaults]") + 1
+            contents.insert(insert_at, COLLECTIONS_PATH)
+
+        with self.path.open(mode="w") as file:
+            file.write("\n".join(contents) + "\n")
+
+    def author_new(self) -> None:
+        """Author the file and update it."""
+        contents = ["[defaults]", COLLECTIONS_PATH]
+        with self.path.open(mode="w") as file:
+            file.write("\n".join(contents) + "\n")

tests/conftest.py

@@ -212,7 +212,7 @@ def session_venv(session_dir: Path, monkey_session: pytest.MonkeyPatch) -> Confi
     cli.parse_args()
     cli.init_output()
     cli.args_sanity()
-    cli.ensure_isolated()
+    cli.isolation_check()
     with pytest.raises(SystemExit):
         cli.run()
     return cli.config
@@ -257,7 +257,7 @@ def function_venv(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Config:
     cli.parse_args()
     cli.init_output()
     cli.args_sanity()
-    cli.ensure_isolated()
+    cli.isolation_check()
     with pytest.raises(SystemExit):
         cli.run()
     return cli.config

tests/unit/test_cli.py

@@ -266,7 +266,7 @@ def test_exit_code_one(
     cli.init_output()
     cli.output.error("Test error")
     with pytest.raises(SystemExit) as excinfo:
-        cli._exit()
+        cli.exit()
     expected = 1
     assert excinfo.value.code == expected
     captured = capsys.readouterr()
@@ -292,7 +292,7 @@ def test_exit_code_two(
     cli.init_output()
     cli.output.warning("Test warning")
     with pytest.raises(SystemExit) as excinfo:
-        cli._exit()
+        cli.exit()
     expected = 2
     assert excinfo.value.code == expected
     captured = capsys.readouterr()