Fix catalog collections script to display all available modules in a …

…collection (#1738)

* Fix catalog collections script

* Parse examples, returndocs, metadata info

* Add test for get_doc_withast function

* Pre-commit and other logic fixes

* Improve code coverage

* Add test

* Fix test_catalog_collections.py

* More fixes in test_catalog_collections.py

* Add test_worker_with_mocked_get_docstring

* More fixes in test

.config/dictionary.txt

@@ -116,5 +116,6 @@ usefixtures
 userbase
 viewcode
 volmount
+withast
 workdir
 xmss

src/ansible_navigator/cli.py

@@ -194,7 +194,7 @@ def main() -> None:
 
     if args.execution_environment:
         pull_image(args)
-        cache_scripts()
+    cache_scripts()
 
     run_return = run(args)
     run_message = f"{run_return.message}\n"

src/ansible_navigator/data/catalog_collections.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import ast
 import hashlib
 import json
 import multiprocessing
@@ -360,6 +361,7 @@ def worker(
     :param completed_queue: The queue in which extracted documentation will be placed
     """
     # pylint: disable=import-outside-toplevel
+    # pylint: disable=too-many-locals
 
     # load the fragment_loader _after_ the path is set
     from ansible.plugins.loader import fragment_loader
@@ -383,10 +385,19 @@ def worker(
                     collection_name=collection_name,
                 )
 
-        except Exception as exc:  # noqa: BLE001
-            err_message = f"{type(exc).__name__} (get_docstring): {exc!s}"
-            completed_queue.put(("error", (checksum, plugin_path, err_message)))
-            continue
+        except Exception:  # noqa: BLE001
+            try:
+                with plugin_path.open(mode="r", encoding="utf-8") as f:
+                    content = f.read()
+                doc, examples, returndocs, metadata = get_doc_withast(content)
+                doc, examples, returndocs, metadata = (
+                    yaml.load(value, Loader=yaml.SafeLoader)
+                    for value in (doc, examples, returndocs, metadata)
+                )
+            except Exception as exc:  # noqa: BLE001
+                err_message = f"{type(exc).__name__} (get_docstring): {exc!s}"
+                completed_queue.put(("error", (checksum, plugin_path, err_message)))
+                continue
 
         try:
             q_message = {
@@ -532,6 +543,32 @@ def retrieve_docs(
             stats["cache_added_errors"] += 1
 
 
+def get_doc_withast(content: Any) -> tuple[Any, Any, Any, Any]:
+    """Get the documentation, examples, returndocs, and metadata from the content using ast.
+
+    :param content: The content to parse using ast.
+    :return: A tuple containing the documentation, examples, returndocs, and metadata.
+    """
+    doc, examples, returndocs, metadata = "", "", "", ""
+
+    # Parse the content using ast and walk through nodes
+    for node in ast.walk(ast.parse(content)):
+        if isinstance(node, ast.Assign) and isinstance(node.targets[0], ast.Name):
+            target_id = node.targets[0].id
+
+            # Check if node.value is an instance of ast.Constant and its value is a string
+            if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
+                if target_id == "DOCUMENTATION":
+                    doc = node.value.value.strip()
+                elif target_id == "EXAMPLES":
+                    examples = node.value.value.strip()
+                elif target_id == "RETURN":
+                    returndocs = node.value.value.strip()
+                elif target_id == "METADATA":
+                    metadata = node.value.value.strip()
+    return doc, examples, returndocs, metadata
+
+
 def run_command(cmd: list[str]) -> dict[str, str]:
     """Run a command using subprocess.
 

tests/fixtures/common/module_1.py

@@ -0,0 +1,17 @@
+"""An ansible test module."""
+
+DOCUMENTATION = r"""
+---
+module: vcenter_mod
+short_description: Gather info vCenter extensions
+description:
+- This module can be used to gather information about vCenter extension.
+author:
+- test
+extends_documentation_fragment:
+- community.vmware.vmware.documentation
+"""
+
+EXAMPLES = "Example usage here."
+RETURN = "This function returns a value."
+METADATA = "Author: John Doe"

tests/unit/test_catalog_collections.py

@@ -0,0 +1,54 @@
+"""Unit tests for catalog collections."""
+
+import multiprocessing
+
+from pathlib import Path
+from typing import Any
+
+from ansible_navigator.data.catalog_collections import worker
+
+
+def test_worker_with_failed_get_docstring() -> None:
+    """Test worker function when get_docstring fails and get_doc_withast method is used to parse the content."""
+    # Create the queues
+    pending_queue: multiprocessing.Queue[Any] = multiprocessing.Queue()
+    completed_queue: multiprocessing.Queue[Any] = multiprocessing.Queue()
+
+    plugin_path = Path("tests/fixtures/common/module_1.py")
+    collection_name = "microsoft.ad"
+    checksum = "12345"
+
+    # Add an entry to the pending queue
+    entry = collection_name, checksum, plugin_path
+    pending_queue.put(entry)
+
+    # Add a None entry to signal the end of processing
+    pending_queue.put(None)
+
+    worker(pending_queue, completed_queue)
+
+    plugin_path, data = completed_queue.get()
+    assert "vCenter" in data[1]
+
+
+def test_worker_with_invalid_plugin_path() -> None:
+    """Test the worker function when get_docstring has invalid plugin_path."""
+    pending_queue: multiprocessing.Queue[Any] = multiprocessing.Queue()
+    completed_queue: multiprocessing.Queue[Any] = multiprocessing.Queue()
+
+    plugin_path = Path("tests/fixtures/common/xyz.py")
+    collection_name = "microsoft.ad"
+    checksum = "12345"
+
+    # Add an entry to the pending queue
+    entry = collection_name, checksum, plugin_path
+    pending_queue.put(entry)
+
+    # Add a None entry to signal the end of processing
+    pending_queue.put(None)
+
+    worker(pending_queue, completed_queue)
+
+    plugin_path, data = completed_queue.get()
+    assert plugin_path == "error"
+    assert "FileNotFoundError (get_docstring)" in data[2]

tests/unit/utils/test_functions.py

@@ -11,6 +11,7 @@
 
 import pytest
 
+from ansible_navigator.data.catalog_collections import get_doc_withast
 from ansible_navigator.utils import functions
 
 
@@ -299,6 +300,30 @@ def test_now_iso(caplog: pytest.LogCaptureFixture, time_zone: str) -> None:
     ),
 )
 def test_unescape_moustaches(data: Any, output: Any) -> None:
-    """Tests unescape_moustaches."""
+    """Tests unescape_moustaches.
+
+    :param data: The input data.
+    :param output: The expected output.
+    """
     result = functions.unescape_moustaches(data)
     assert result == output
+
+
+def test_get_doc_withast() -> None:
+    """Test for the get_doc_withast function.
+
+    This test ensures that the get_doc_withast function correctly extracts the documentation,
+    examples, returndocs, and metadata from the module content.
+    """
+    module_content = """
+DOCUMENTATION = "This is a test documentation."
+EXAMPLES = "Example usage here."
+RETURN = "This function returns a value."
+METADATA = "Author: John Doe"
+"""
+
+    doc, examples, returndocs, metadata = get_doc_withast(module_content)
+    assert doc == "This is a test documentation."
+    assert examples == "Example usage here."
+    assert returndocs == "This function returns a value."
+    assert metadata == "Author: John Doe"