Allow more control over verbosity when calling ansible-galaxy

src/ansible_compat/runtime.py

@@ -165,6 +165,7 @@ def __init__(
         require_module: bool = False,
         max_retries: int = 0,
         environ: dict[str, str] | None = None,
+        verbosity: int = 0,
     ) -> None:
         """Initialize Ansible runtime environment.
 
@@ -186,12 +187,17 @@ def __init__(
                             Default is 0, no retries.
         :param environ: Environment dictionary to use, if undefined
                         ``os.environ`` will be copied and used.
+        :param verbosity: Verbosity level to use.
         """
         self.project_dir = project_dir or Path.cwd()
         self.isolated = isolated
         self.max_retries = max_retries
         self.environ = environ or os.environ.copy()
         self.plugins = Plugins(runtime=self)
+        self.verbosity = verbosity
+
+        self.initialize_logger(level=self.verbosity)
+
         # Reduce noise from paramiko, unless user already defined PYTHONWARNINGS
         # paramiko/transport.py:236: CryptographyDeprecationWarning: Blowfish has been deprecated
         # https://github.com/paramiko/paramiko/issues/2038
@@ -235,6 +241,21 @@ def warning(
         # Monkey patch ansible warning in order to use warnings module.
         Display.warning = warning
 
+    def initialize_logger(self, level: int = 0) -> None:
+        """Set up the global logging level based on the verbosity number."""
+        verbosity_map = {
+            -2: logging.CRITICAL,
+            -1: logging.ERROR,
+            0: logging.WARNING,
+            1: logging.INFO,
+            2: logging.DEBUG,
+        }
+        # Unknown logging level is treated as DEBUG
+        logging_level = verbosity_map.get(level, logging.DEBUG)
+        _logger.setLevel(logging_level)
+        # Use module-level _logger instance to validate it
+        _logger.debug("Logging initialized to level %s", logging_level)
+
     def _add_sys_path_to_collection_paths(self) -> None:
         """Add the sys.path to the collection paths."""
         if self.config.collections_scan_sys_path:
@@ -505,9 +526,11 @@ def install_requirements(  # noqa: C901
                 "ansible-galaxy",
                 "role",
                 "install",
-                "-vr",
+                "-r",
                 f"{requirement}",
             ]
+            if self.verbosity > 0:
+                cmd.extend(["-" + ("v" * self.verbosity)])
             if self.cache_dir:
                 cmd.extend(["--roles-path", f"{self.cache_dir}/roles"])
 
@@ -519,8 +542,9 @@ def install_requirements(  # noqa: C901
                 _logger.info("Running %s", " ".join(cmd))
 
                 result = self.run(cmd, retry=retry)
+                _logger.debug(result.stdout)
                 if result.returncode != 0:
-                    _logger.error(result.stdout)
+                    _logger.error(result.stderr)
                     raise AnsibleCommandError(result)
 
         # Run galaxy collection install works on v2 requirements.yml
@@ -529,8 +553,10 @@ def install_requirements(  # noqa: C901
                 "ansible-galaxy",
                 "collection",
                 "install",
-                "-v",
             ]
+            if self.verbosity > 0:
+                cmd.extend(["-" + ("v" * self.verbosity)])
+
             for collection in reqs_yaml["collections"]:
                 if isinstance(collection, dict) and collection.get("type", "") == "git":
                     _logger.info(
@@ -558,8 +584,8 @@ def install_requirements(  # noqa: C901
                     retry=retry,
                     env={**os.environ, "ANSIBLE_COLLECTIONS_PATH": ":".join(cpaths)},
                 )
+                _logger.debug(result.stdout)
                 if result.returncode != 0:
-                    _logger.error(result.stdout)
                     _logger.error(result.stderr)
                     raise AnsibleCommandError(result)
 

test/test_runtime.py

@@ -240,10 +240,11 @@ def cwd(path: Path) -> Iterator[None]:
         os.chdir(old_pwd)
 
 
-def test_prerun_reqs_v1(caplog: pytest.LogCaptureFixture, runtime: Runtime) -> None:
+def test_prerun_reqs_v1(caplog: pytest.LogCaptureFixture) -> None:
     """Checks that the linter can auto-install requirements v1 when found."""
+    runtime = Runtime(verbosity=1)
     path = Path(__file__).parent.parent / "examples" / "reqs_v1"
-    with cwd(path), caplog.at_level(logging.INFO):
+    with cwd(path):
         runtime.prepare_environment()
     assert any(
         msg.startswith("Running ansible-galaxy role install") for msg in caplog.messages
@@ -254,12 +255,12 @@ def test_prerun_reqs_v1(caplog: pytest.LogCaptureFixture, runtime: Runtime) -> N
     )
 
 
-def test_prerun_reqs_v2(caplog: pytest.LogCaptureFixture, runtime: Runtime) -> None:
+def test_prerun_reqs_v2(caplog: pytest.LogCaptureFixture) -> None:
     """Checks that the linter can auto-install requirements v2 when found."""
+    runtime = Runtime(verbosity=1)
     path = (Path(__file__).parent.parent / "examples" / "reqs_v2").resolve()
     with cwd(path):
-        with caplog.at_level(logging.INFO):
-            runtime.prepare_environment()
+        runtime.prepare_environment()
         assert any(
             msg.startswith("Running ansible-galaxy role install")
             for msg in caplog.messages
@@ -526,11 +527,10 @@ def test_install_galaxy_role(runtime_tmp: Runtime) -> None:
 
 
 def test_install_galaxy_role_unlink(
-    runtime_tmp: Runtime,
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """Test ability to unlink incorrect symlinked roles."""
-    caplog.set_level(logging.INFO)
+    runtime_tmp = Runtime(verbosity=1)
     runtime_tmp.prepare_environment()
     pathlib.Path(f"{runtime_tmp.cache_dir}/roles").mkdir(parents=True, exist_ok=True)
     pathlib.Path(f"{runtime_tmp.cache_dir}/roles/acme.get_rich").symlink_to("/dev/null")