[stable-2.15] Fix TypeError when installing subdir collections. Fixes #80943 (#80949) (#80950)

(cherry picked from commit 0982d5f)

changelogs/fragments/80943-ansible-galaxy-collection-subdir-install.yml

@@ -0,0 +1,3 @@
+bugfixes:
+- ansible-galaxy - Fix variable type error when installing subdir collections
+  (https://github.com/ansible/ansible/issues/80943)

lib/ansible/galaxy/collection/__init__.py

@@ -283,11 +283,8 @@ def verify_local_collection(local_collection, remote_collection, artifacts_manag
         manifest_hash = get_hash_from_validation_source(MANIFEST_FILENAME)
     else:
         # fetch remote
-        b_temp_tar_path = (  # NOTE: AnsibleError is raised on URLError
+        # NOTE: AnsibleError is raised on URLError
-            artifacts_manager.get_artifact_path
+        b_temp_tar_path = artifacts_manager.get_artifact_path_from_unknown(remote_collection)
-            if remote_collection.is_concrete_artifact
-            else artifacts_manager.get_galaxy_artifact_path
-        )(remote_collection)
 
         display.vvv(
             u"Remote collection cached as '{path!s}'".format(path=to_text(b_temp_tar_path))
@@ -557,11 +554,7 @@ def download_collections(
                 format(coll=to_text(concrete_coll_pin), path=to_text(b_output_path)),
             )
 
-            b_src_path = (
+            b_src_path = artifacts_manager.get_artifact_path_from_unknown(concrete_coll_pin)
-                artifacts_manager.get_artifact_path
-                if concrete_coll_pin.is_concrete_artifact
-                else artifacts_manager.get_galaxy_artifact_path
-            )(concrete_coll_pin)
 
             b_dest_path = os.path.join(
                 b_output_path,
@@ -681,7 +674,7 @@ def install_collections(
     unsatisfied_requirements = set(
         chain.from_iterable(
             (
-                Requirement.from_dir_path(sub_coll, artifacts_manager)
+                Requirement.from_dir_path(to_bytes(sub_coll), artifacts_manager)
                 for sub_coll in (
                     artifacts_manager.
                     get_direct_collection_dependencies(install_req).
@@ -1485,10 +1478,7 @@ def install(collection, path, artifacts_manager):  # FIXME: mv to dataclasses?
     :param path: Collection dirs layout path.
     :param artifacts_manager: Artifacts manager.
     """
-    b_artifact_path = (
+    b_artifact_path = artifacts_manager.get_artifact_path_from_unknown(collection)
-        artifacts_manager.get_artifact_path if collection.is_concrete_artifact
-        else artifacts_manager.get_galaxy_artifact_path
-    )(collection)
 
     collection_path = os.path.join(path, collection.namespace, collection.name)
     b_collection_path = to_bytes(collection_path, errors='surrogate_or_strict')

lib/ansible/galaxy/collection/concrete_artifact_manager.py

@@ -21,7 +21,7 @@ from tempfile import mkdtemp
 
 if t.TYPE_CHECKING:
     from ansible.galaxy.dependency_resolution.dataclasses import (
-        Candidate, Requirement,
+        Candidate, Collection, Requirement,
     )
     from ansible.galaxy.token import GalaxyToken
 
@@ -190,7 +190,7 @@ class ConcreteArtifactsManager:
         return b_artifact_path
 
     def get_artifact_path(self, collection):
-        # type: (t.Union[Candidate, Requirement]) -> bytes
+        # type: (Collection) -> bytes
         """Given a concrete collection pointer, return a cached path.
 
         If it's not yet on disk, this method downloads the artifact first.
@@ -251,16 +251,22 @@ class ConcreteArtifactsManager:
         self._artifact_cache[collection.src] = b_artifact_path
         return b_artifact_path
 
+    def get_artifact_path_from_unknown(self, collection):
+        # type: (Candidate) -> bytes
+        if collection.is_concrete_artifact:
+            return self.get_artifact_path(collection)
+        return self.get_galaxy_artifact_path(collection)
+
     def _get_direct_collection_namespace(self, collection):
         # type: (Candidate) -> t.Optional[str]
         return self.get_direct_collection_meta(collection)['namespace']  # type: ignore[return-value]
 
     def _get_direct_collection_name(self, collection):
-        # type: (Candidate) -> t.Optional[str]
+        # type: (Collection) -> t.Optional[str]
         return self.get_direct_collection_meta(collection)['name']  # type: ignore[return-value]
 
     def get_direct_collection_fqcn(self, collection):
-        # type: (Candidate) -> t.Optional[str]
+        # type: (Collection) -> t.Optional[str]
         """Extract FQCN from the given on-disk collection artifact.
 
         If the collection is virtual, ``None`` is returned instead
@@ -276,7 +282,7 @@ class ConcreteArtifactsManager:
         ))
 
     def get_direct_collection_version(self, collection):
-        # type: (t.Union[Candidate, Requirement]) -> str
+        # type: (Collection) -> str
         """Extract version from the given on-disk collection artifact."""
         return self.get_direct_collection_meta(collection)['version']  # type: ignore[return-value]
 
@@ -289,7 +295,7 @@ class ConcreteArtifactsManager:
         return collection_dependencies  # type: ignore[return-value]
 
     def get_direct_collection_meta(self, collection):
-        # type: (t.Union[Candidate, Requirement]) -> dict[str, t.Union[str, dict[str, str], list[str], None, t.Type[Sentinel]]]
+        # type: (Collection) -> dict[str, t.Union[str, dict[str, str], list[str], None, t.Type[Sentinel]]]
         """Extract meta from the given on-disk collection artifact."""
         try:  # FIXME: use unique collection identifier as a cache key?
             return self._artifact_meta_cache[collection.src]

lib/ansible/galaxy/dependency_resolution/dataclasses.py

@@ -216,12 +216,15 @@ class _ComputedReqKindsMixin:
             return cls.from_dir_path_implicit(dir_path)
 
     @classmethod
-    def from_dir_path(cls, dir_path, art_mgr):
+    def from_dir_path(  # type: ignore[misc]
+            cls,  # type: t.Type[Collection]
+            dir_path,  # type: bytes
+            art_mgr,  # type: ConcreteArtifactsManager
+    ):  # type: (...)  -> Collection
         """Make collection from an directory with metadata."""
         if dir_path.endswith(to_bytes(os.path.sep)):
             dir_path = dir_path.rstrip(to_bytes(os.path.sep))
-        b_dir_path = to_bytes(dir_path, errors='surrogate_or_strict')
+        if not _is_collection_dir(dir_path):
-        if not _is_collection_dir(b_dir_path):
             display.warning(
                 u"Collection at '{path!s}' does not have a {manifest_json!s} "
                 u'file, nor has it {galaxy_yml!s}: cannot detect version.'.