Add isolation options

Author: cidrblock

.config/dictionary.txt

@@ -27,6 +27,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,7 @@
 
 from .arg_parser import parse
 from .config import Config
+from .definitions import AnsibleCfg
 from .output import Output
 from .utils import TermFeatures
 
@@ -92,8 +93,83 @@ 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()
+        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)
+            return False
+
+        cwd = AnsibleCfg(path=Path("./ansible.cfg"))
+        home = AnsibleCfg(path=Path("~/.ansible.cfg").expanduser().resolve())
+        system = AnsibleCfg(path=Path("/etc/ansible/ansible.cfg"))
+
+        if cwd.exists and cwd.collections_path_is_dot:
+            self.output.info(f"{cwd.path} has collections_path=. which isolates this workspace.")
+            return True
+        if cwd.exists and not cwd.collections_path_is_dot:
+            cwd.set_or_update_collection_path()
+            self.output.warning(
+                f"{cwd.path} has been updated with collections_path=. to isolate this workspace.",
+            )
+            return True
+        if home.exists and not home.collections_path_is_dot:
+            self.output.warning(
+                f"{home.path} has been updated with collections_path=. to isolate this and all workspaces.",
+            )
+            return True
+        if system.exists and system.collections_path_is_dot:
+            self.output.info(
+                f"{system.path} has collections_path=. which isolates this and all workspaces.",
+            )
+            return True
+
+        cwd.author_new()
+        self.output.info(
+            f"{cwd.path} has been created with collections_path=. to isolate this workspace.",
+        )
+        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.",
+        )
+        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 +203,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 +221,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 +247,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,86 @@
+"""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
+
+
+@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_collection_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.
+        """
+        config = ConfigParser()
+        config.read(self.path)
+        if not config.has_section("defaults"):
+            with self.path.open(mode="r+") as file:
+                content_str = file.read()
+                file.seek(0, 0)
+                file.truncate()
+                file.write("[defaults]\ncollections_path = .\n" + content_str)
+                file.write("\n")
+            return
+
+        if not config.has_option("defaults", "collections_path"):
+            with self.path.open(mode="r+") as file:
+                content_list = file.read().splitlines()
+                idx = content_list.index("[defaults]") + 1
+                content_list.insert(idx, "collections_path = .")
+                file.seek(0, 0)
+                file.truncate()
+                file.write("\n".join(content_list))
+                file.write("\n")
+            return
+
+        with self.path.open(mode="r+") as file:
+            content_list = file.read().splitlines()
+            idx = next(
+                i for i, line in enumerate(content_list) if line.startswith("collections_path")
+            )
+            content_list[idx] = "collections_path = ."
+            file.seek(0, 0)
+            file.truncate()
+            file.write("\n".join(content_list))
+            file.write("\n")
+        return
+
+    def author_new(self) -> None:
+        """Author the file and update it."""
+        config = ConfigParser()
+        config.add_section("defaults")
+        config.set("defaults", "collections_path", ".")
+        with self.path.open(mode="w") as f:
+            config.write(f)

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()