Add require_collection method

Adds utility method that detects if a collection is installed or if
it outdated and exits. This functionality is not directly used by
the linter yet but putting this code near similar prerun method makes
it easier to reuse in other related projects.

Related: ansible-community/molecule-podman#38

src/ansiblelint/prerun.py

@@ -1,13 +1,15 @@
 """Utilities for configuring ansible runtime environment."""
+import json
 import logging
 import os
 import pathlib
 import re
 import subprocess
 import sys
 from functools import lru_cache
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Type, Union
 
+import packaging
 import tenacity
 from packaging import version
 
@@ -371,3 +373,69 @@ def _perform_mockings() -> None:
         if link_path.exists():
             link_path.unlink()
         link_path.symlink_to(target, target_is_directory=True)
+
+
+def ansible_config_get(key: str, kind: Type[Any] = str) -> Union[str, List[str], None]:
+    """Return configuration item from ansible config."""
+    env = os.environ.copy()
+    # Avoid possible ANSI garbage
+    env["ANSIBLE_FORCE_COLOR"] = "0"
+    # Avoid our own override as this prevents returning system paths.
+    env.pop('ANSIBLE_COLLECTIONS_PATH')
+
+    config = subprocess.check_output(
+        ["ansible-config", "dump"], universal_newlines=True, env=env
+    )
+
+    if kind == str:
+        result = re.search(rf"^{key}.* = (.*)$", config, re.MULTILINE)
+        if result:
+            return result.groups()[0]
+    elif kind == list:
+        result = re.search(rf"^{key}.* = (\[.*\])$", config, re.MULTILINE)
+        if result:
+            val = eval(result.groups()[0])  # pylint: disable=eval-used
+            if not isinstance(val, list):
+                raise RuntimeError(f"Unexpected data read for {key}: {val}")
+            return val
+    else:
+        raise RuntimeError("Unknown data type.")
+    return None
+
+
+def require_collection(name: str, version: Optional[str] = None) -> None:
+    """Check if a minimal collection version is present or exits.
+
+    In the future this method may attempt to install a missing or outdated
+    collection before failing.
+    """
+    try:
+        ns, coll = name.split('.', 1)
+    except ValueError:
+        sys.exit("Invalid collection name supplied: %s" % name)
+
+    paths = ansible_config_get('COLLECTIONS_PATHS', list)
+    if not paths or not isinstance(paths, list):
+        sys.exit(f"Unable to determine ansible collection paths. ({paths})")
+    for path in paths:
+        collpath = os.path.join(path, 'ansible_collections', ns, coll)
+        if os.path.exists(collpath):
+            mpath = os.path.join(collpath, 'MANIFEST.json')
+            if not os.path.exists(mpath):
+                sys.exit(
+                    "Found collection at '%s' but missing MANIFEST.json, cannot get info."
+                    % collpath
+                )
+
+            with open(mpath, 'r') as f:
+                manifest = json.loads(f.read())
+                found_version = packaging.version.parse(
+                    manifest['collection_info']['version']
+                )
+                if version and found_version < packaging.version.parse(version):
+                    sys.exit(
+                        f"Found {name} collection {found_version} but {version} or newer is required."
+                    )
+            break
+    else:
+        sys.exit("Collection '%s' not found in '%s'" % (name, paths))

test/test_prerun.py

@@ -1,5 +1,6 @@
 """Tests related to prerun part of the linter."""
 import os
+import subprocess
 from typing import List
 
 import pytest
@@ -145,3 +146,44 @@ def test__update_env(
     prerun._update_env("DUMMY_VAR", value)
 
     assert os.environ["DUMMY_VAR"] == result
+
+
+def test_require_collection_wrong_version() -> None:
+    """Tests behaviour of require_collection."""
+    subprocess.check_output(
+        [
+            "ansible-galaxy",
+            "collection",
+            "install",
+            "containers.podman",
+            "-p",
+            "~/.ansible/collections",
+        ]
+    )
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        prerun.require_collection("containers.podman", '9999.9.9')
+    assert pytest_wrapped_e.type == SystemExit
+    assert 'Found containers.podman collection' in pytest_wrapped_e.value.code  # type: ignore
+    assert 'but 9999.9.9 or newer is required.' in pytest_wrapped_e.value.code  # type: ignore
+
+
+@pytest.mark.parametrize(
+    ("name", "version"),
+    (
+        ("this.is.sparta", None),
+        ("this.is.sparta", "9999.9.9"),
+    ),
+)
+def test_require_collection_missing(name: str, version: str) -> None:
+    """Tests behaviour of require_collection, missing case."""
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        prerun.require_collection(name, version)
+    assert pytest_wrapped_e.type == SystemExit
+    assert f"Collection '{name}' not found in" in pytest_wrapped_e.value.code  # type: ignore
+
+
+def test_ansible_config_get() -> None:
+    """Check test_ansible_config_get."""
+    paths = prerun.ansible_config_get("COLLECTIONS_PATHS", list)
+    assert isinstance(paths, list)
+    assert len(paths) > 0