From e44194087dc7bbc9239081b715e798f45be2d501 Mon Sep 17 00:00:00 2001
From: Sorin Sbarnea <ssbarnea@redhat.com>
Date: Tue, 15 Jun 2021 14:50:29 +0100
Subject: [PATCH] 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: https://github.com/ansible-community/molecule-podman/pull/38
---
 src/ansiblelint/prerun.py | 70 ++++++++++++++++++++++++++++++++++++++-
 test/test_prerun.py       | 42 +++++++++++++++++++++++
 2 files changed, 111 insertions(+), 1 deletion(-)

diff --git a/src/ansiblelint/prerun.py b/src/ansiblelint/prerun.py
index 9d75bfa6255..905d3355efe 100644
--- a/src/ansiblelint/prerun.py
+++ b/src/ansiblelint/prerun.py
@@ -1,4 +1,5 @@
 """Utilities for configuring ansible runtime environment."""
+import json
 import logging
 import os
 import pathlib
@@ -6,8 +7,9 @@
 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))
diff --git a/test/test_prerun.py b/test/test_prerun.py
index ae4d96930b6..b26ebe35713 100644
--- a/test/test_prerun.py
+++ b/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