fileglob: make fileglob work with globbed path components again

PR ansible#16630 broke globbing in with_fileglob for globbed path components
by changing from path_dwim_relative to path_dwim_relative_stack. While
the former does something like this (for a path specification like
"/tmp/*/some.conf", see Issue ansible#17136 for an example playbook)

search = ["/tmp/*"]
for candidate in search:
    if os.path.exists(candidate):
        break
return candidate

the check inside path_dwim_relative_stack works slightly differently:

result = None
if os.path.exists("/tmp/*"):
    result = "/tmp/*"
return result

The way the loop in the former example works makes path_dwim_relative
return the last element from the 'search' list even if os.path.exists
never returned True, while path_dwim_relative_stack returns None in
the same case. This ultimately makes the fileglob plugin fail because
find_file_in_search_path returns None now.

To allow asterisks in paths again, this change breaks the search for
potential targets out of path_dwim_relative_stack. This function can
then be called from the fileglob plugin to find all potential target
directories without checking if these dirs (like "/tmp/*") actually
exist. fileglob will now apply globbing first and check for the actual
existence of the files _after_ globbing.

lib/ansible/parsing/dataloader.py

@@ -226,6 +226,46 @@ def path_dwim(self, given):
             basedir = to_text(self._basedir, errors='surrogate_or_strict')
             return os.path.abspath(os.path.join(basedir, given))
 
+    def path_dwim_get_candidates(self, paths, dirname, b_source, is_role=False):
+        '''
+        Get search target candidates for given paths in a directory dirname.
+        '''
+        b_dirname = to_bytes(dirname)
+        search = []
+        display.debug(u'evaluation_path:\n\t%s' % '\n\t'.join(paths))
+        for path in paths:
+            upath = unfrackpath(path)
+            b_upath = to_bytes(upath, errors='surrogate_or_strict')
+            b_mydir = os.path.dirname(b_upath)
+
+            # FIXME: this detection fails with non main.yml roles
+            # if path is in role and 'tasks' not there already, add it into the search
+            if is_role or (b_upath.endswith(b'tasks') and os.path.exists(os.path.join(b_upath, b'main.yml')) \
+                    or os.path.exists(os.path.join(b_upath, b'tasks/main.yml')) \
+                    or os.path.exists(os.path.join(b_mydir, b'tasks/main.yml'))):
+                if b_mydir.endswith(b'tasks'):
+                    search.append(os.path.join(os.path.dirname(b_mydir), b_dirname, b_source))
+                    search.append(os.path.join(b_mydir, b_source))
+                else:
+                    # don't add dirname if user already is using it in source
+                    if b_source.split(b'/')[0] != b_dirname:
+                        search.append(os.path.join(b_upath, b_dirname, b_source))
+                    search.append(os.path.join(b_upath, b_source))
+
+            elif b_dirname not in b_source.split(b'/'):
+                # don't add dirname if user already is using it in source
+                if b_source.split(b'/')[0] != dirname:
+                    search.append(os.path.join(b_upath, b_dirname, b_source))
+                search.append(os.path.join(b_upath, b_source))
+
+        # always append basedir as last resort
+        # don't add dirname if user already is using it in source
+        if b_source.split(b'/')[0] != dirname:
+            search.append(os.path.join(to_bytes(self.get_basedir()), b_dirname, b_source))
+        search.append(os.path.join(to_bytes(self.get_basedir()), b_source))
+
+        return search
+
     def path_dwim_relative(self, path, dirname, source, is_role=False):
         '''
         find one file in either a role or playbook dir with or without
@@ -289,7 +329,6 @@ def path_dwim_relative_stack(self, paths, dirname, source, is_role=False):
         :rtype: A text string
         :returns: An absolute path to the filename ``source``
         '''
-        b_dirname = to_bytes(dirname)
         b_source = to_bytes(source)
 
         result = None
@@ -301,38 +340,7 @@ def path_dwim_relative_stack(self, paths, dirname, source, is_role=False):
             if os.path.exists(to_bytes(test_path, errors='surrogate_or_strict')):
                 result = test_path
         else:
-            search = []
-            display.debug(u'evaluation_path:\n\t%s' % '\n\t'.join(paths))
-            for path in paths:
-                upath = unfrackpath(path)
-                b_upath = to_bytes(upath, errors='surrogate_or_strict')
-                b_mydir = os.path.dirname(b_upath)
-
-                # FIXME: this detection fails with non main.yml roles
-                # if path is in role and 'tasks' not there already, add it into the search
-                if is_role or (b_upath.endswith(b'tasks') and os.path.exists(os.path.join(b_upath, b'main.yml')) \
-                        or os.path.exists(os.path.join(b_upath, b'tasks/main.yml')) \
-                        or os.path.exists(os.path.join(b_mydir, b'tasks/main.yml'))):
-                    if b_mydir.endswith(b'tasks'):
-                        search.append(os.path.join(os.path.dirname(b_mydir), b_dirname, b_source))
-                        search.append(os.path.join(b_mydir, b_source))
-                    else:
-                        # don't add dirname if user already is using it in source
-                        if b_source.split(b'/')[0] != b_dirname:
-                            search.append(os.path.join(b_upath, b_dirname, b_source))
-                        search.append(os.path.join(b_upath, b_source))
-
-                elif b_dirname not in b_source.split(b'/'):
-                    # don't add dirname if user already is using it in source
-                    if b_source.split(b'/')[0] != dirname:
-                        search.append(os.path.join(b_upath, b_dirname, b_source))
-                    search.append(os.path.join(b_upath, b_source))
-
-            # always append basedir as last resort
-            # don't add dirname if user already is using it in source
-            if b_source.split(b'/')[0] != dirname:
-                search.append(os.path.join(to_bytes(self.get_basedir()), b_dirname, b_source))
-            search.append(os.path.join(to_bytes(self.get_basedir()), b_source))
+            search = self.path_dwim_get_candidates(paths, dirname, b_source, is_role)
 
             display.debug(u'search_path:\n\t%s' % to_text(b'\n\t'.join(search)))
             for b_candidate in search:

lib/ansible/plugins/lookup/__init__.py

@@ -45,6 +45,12 @@ def get_basedir(self, variables):
         else:
             return self._loader.get_basedir()
 
+    def get_search_path(self, variables):
+        if 'ansible_search_path' in variables:
+            return variables['ansible_search_path']
+        else:
+            return self.get_basedir(variables)
+
     @staticmethod
     def _flatten(terms):
         ret = []
@@ -107,10 +113,7 @@ def find_file_in_search_path(self, myvars, subdir, needle, ignore_missing=False)
         Return a file (needle) in the task's expected search path.
         '''
 
-        if 'ansible_search_path' in myvars:
-            paths = myvars['ansible_search_path']
-        else:
-            paths = self.get_basedir(myvars)
+        paths = self.get_search_path(myvars)
 
         result = self._loader.path_dwim_relative_stack(paths, subdir, needle)
         if result is None and not ignore_missing:

lib/ansible/plugins/lookup/fileglob.py

@@ -23,6 +23,7 @@
 from ansible.plugins.lookup import LookupBase
 from ansible.errors import AnsibleFileNotFound
 from ansible.module_utils._text import to_bytes, to_text
+from ansible.utils.path import unfrackpath
 
 
 class LookupModule(LookupBase):
@@ -32,8 +33,16 @@ def run(self, terms, variables=None, **kwargs):
         ret = []
         for term in terms:
             term_file = os.path.basename(term)
-            dwimmed_path = self.find_file_in_search_path(variables, 'files', os.path.dirname(term))
-            if dwimmed_path:
+            term_dir = os.path.dirname(term)
+            if term_dir.startswith('~') or term_dir.startswith(os.path.sep):
+                basepath = unfrackpath(to_text(term_dir))
+                candidates = [basepath]
+            else:
+                paths = self.get_search_path(variables)
+                candidates = self._loader.path_dwim_get_candidates(paths, 'files', term_dir)
+
+            for b_candidate in candidates:
+                dwimmed_path = to_text(b_candidate)
                 globbed = glob.glob(to_bytes(os.path.join(dwimmed_path, term_file), errors='surrogate_or_strict'))
                 ret.extend(to_text(g, errors='surrogate_or_strict') for g in globbed if os.path.isfile(g))
         return ret

test/integration/targets/lookups/tasks/main.yml

@@ -291,3 +291,41 @@
 - assert:
     that:
       - "hello_world|trim == 'Hello world!'"
+
+# Fileglob lookup
+- name: Prepare directories for fileglob test
+  file: path={{output_dir}}/fileglob_dir_{{item}} state=directory
+  with_items: [one, two]
+
+- name: Create files for fileglob test
+  copy: dest={{output_dir}}/{{item}} mode=0644 content=""
+  with_items: [fileglob_dir_one/some.conf, fileglob_dir_one/other.conf, fileglob_dir_two/some.conf]
+
+- name: Test globbing of filenames
+  debug: var=item
+  with_fileglob: "{{output_dir}}/fileglob_dir_one/*.conf"
+  register: fileglob_files
+
+- name: Verify globbing of filenames
+  assert:
+    that:
+        - fileglob_files.results|length == 2
+
+- name: Test globbing of files with asterisk in path component
+  debug: var=item
+  with_fileglob: "{{output_dir}}/fileglob_dir_*/some.conf"
+  register: fileglob_dirs
+
+- name: Verify globbing of files with asterisk in path component
+  assert:
+    that:
+        - fileglob_dirs.results|length == 2
+
+- name: Cleanup fileglob test directories
+  file: dest={{item}} state=absent
+  with_items:
+  - "{{output_dir}}/fileglob_dir_one/some.conf"
+  - "{{output_dir}}/fileglob_dir_one/other.conf"
+  - "{{output_dir}}/fileglob_dir_two/some.conf"
+  - "{{output_dir}}/fileglob_dir_one"
+  - "{{output_dir}}/fileglob_dir_two"