Add validate_collection_path function (#66441)

* Add validate_collection_path function

Utility function for ensuring a collection target ends with 'ansible_collection'

* Fix bad syntax

* Correct docstring

changelogs/fragments/galaxy-add-path-validation-utility-function.yaml

@@ -0,0 +1,2 @@
+minor_changes:
+  - ansible-galaxy - add ``validate_collection_path()`` utility function ()

lib/ansible/cli/galaxy.py

@@ -22,8 +22,13 @@ from ansible.cli.arguments import option_helpers as opt_help
 from ansible.errors import AnsibleError, AnsibleOptionsError
 from ansible.galaxy import Galaxy, get_collections_galaxy_meta_info
 from ansible.galaxy.api import GalaxyAPI
-from ansible.galaxy.collection import build_collection, install_collections, publish_collection, \
-    validate_collection_name
+from ansible.galaxy.collection import (
+    build_collection,
+    install_collections,
+    publish_collection,
+    validate_collection_name,
+    validate_collection_path,
+)
 from ansible.galaxy.login import GalaxyLogin
 from ansible.galaxy.role import GalaxyRole
 from ansible.galaxy.token import BasicAuthToken, GalaxyToken, KeycloakToken, NoTokenSentinel
@@ -827,9 +832,7 @@ class GalaxyCLI(CLI):
                                 "collections paths '%s'. The installed collection won't be picked up in an Ansible "
                                 "run." % (to_text(output_path), to_text(":".join(collections_path))))
 
-            if os.path.split(output_path)[1] != 'ansible_collections':
-                output_path = os.path.join(output_path, 'ansible_collections')
-
+            output_path = validate_collection_path(output_path)
             b_output_path = to_bytes(output_path, errors='surrogate_or_strict')
             if not os.path.exists(b_output_path):
                 os.makedirs(b_output_path)

lib/ansible/galaxy/collection.py

@@ -466,6 +466,19 @@ def validate_collection_name(name):
                        "characters from [a-zA-Z0-9_] only." % name)
 
 
+def validate_collection_path(collection_path):
+    """ Ensure a given path ends with 'ansible_collections'
+
+    :param collection_path: The path that should end in 'ansible_collections'
+    :return: collection_path ending in 'ansible_collections' if it does not already.
+    """
+
+    if os.path.split(collection_path)[1] != 'ansible_collections':
+        return os.path.join(collection_path, 'ansible_collections')
+
+    return collection_path
+
+
 @contextmanager
 def _tempdir():
     b_temp_path = tempfile.mkdtemp(dir=to_bytes(C.DEFAULT_LOCAL_TMP, errors='surrogate_or_strict'))