ansible-galaxy - Add timeout and progress indicator for publish and install (#60660)

* ansible-galaxy - Add timeout and progress indicator for publish

* add progress indicator to install phase as well

lib/ansible/cli/galaxy.py

@@ -285,6 +285,8 @@ class GalaxyCLI(CLI):
                                     help='The path to the collection tarball to publish.')
         publish_parser.add_argument('--no-wait', dest='wait', action='store_false', default=True,
                                     help="Don't wait for import validation results.")
+        publish_parser.add_argument('--import-timeout', dest='import_timeout', type=int, default=0,
+                                    help="The time to wait for the collection import process to finish.")
 
     def post_process_args(self, options):
         options = super(GalaxyCLI, self).post_process_args(options)
@@ -977,8 +979,9 @@ class GalaxyCLI(CLI):
         """
         collection_path = GalaxyCLI._resolve_path(context.CLIARGS['args'])
         wait = context.CLIARGS['wait']
+        timeout = context.CLIARGS['import_timeout']
 
-        publish_collection(collection_path, self.api, wait)
+        publish_collection(collection_path, self.api, wait, timeout)
 
     def execute_search(self):
         ''' searches for roles on the Ansible Galaxy server'''

lib/ansible/galaxy/collection.py

@@ -11,6 +11,7 @@ import os
 import shutil
 import tarfile
 import tempfile
+import threading
 import time
 import uuid
 import yaml
@@ -21,6 +22,11 @@ from hashlib import sha256
 from io import BytesIO
 from yaml.error import YAMLError
 
+try:
+    import queue
+except ImportError:
+    import Queue as queue  # Python 2
+
 import ansible.constants as C
 from ansible.errors import AnsibleError
 from ansible.galaxy import get_collections_galaxy_meta_info
@@ -383,13 +389,14 @@ def build_collection(collection_path, output_path, force):
     _build_collection_tar(b_collection_path, b_collection_output, collection_manifest, file_manifest)
 
 
-def publish_collection(collection_path, api, wait):
+def publish_collection(collection_path, api, wait, timeout):
     """
     Publish an Ansible collection tarball into an Ansible Galaxy server.
 
     :param collection_path: The path to the collection tarball to publish.
     :param api: A GalaxyAPI to publish the collection to.
     :param wait: Whether to wait until the import process is complete.
+    :param timeout: The time in seconds to wait for the import process to finish, 0 is indefinite.
     """
     b_collection_path = to_bytes(collection_path, errors='surrogate_or_strict')
     if not os.path.exists(b_collection_path):
@@ -423,14 +430,16 @@ def publish_collection(collection_path, api, wait):
         raise AnsibleError("Error when publishing collection (HTTP Code: %d, Message: %s Code: %s)"
                            % (err.code, message, code))
 
-    display.vvv("Collection has been pushed to the Galaxy server %s %s" % (api.name, api.api_server))
     import_uri = resp['task']
     if wait:
-        _wait_import(import_uri, api)
-        display.display("Collection has been successfully published to the Galaxy server")
+        display.display("Collection has been published to the Galaxy server %s %s" % (api.name, api.api_server))
+        _wait_import(import_uri, api, timeout)
+        display.display("Collection has been successfully published and imported to the Galaxy server %s %s"
+                        % (api.name, api.api_server))
     else:
-        display.display("Collection has been pushed to the Galaxy server, not waiting until import has completed "
-                        "due to --no-wait being set. Import task results can be found at %s" % import_uri)
+        display.display("Collection has been pushed to the Galaxy server %s %s, not waiting until import has "
+                        "completed due to --no-wait being set. Import task results can be found at %s"
+                        % (api.name, api.api_server, import_uri))
 
 
 def install_collections(collections, output_path, apis, validate_certs, ignore_errors, no_deps, force, force_deps):
@@ -449,18 +458,22 @@ def install_collections(collections, output_path, apis, validate_certs, ignore_e
     existing_collections = _find_existing_collections(output_path)
 
     with _tempdir() as b_temp_path:
-        dependency_map = _build_dependency_map(collections, existing_collections, b_temp_path, apis, validate_certs,
-                                               force, force_deps, no_deps)
-
-        for collection in dependency_map.values():
-            try:
-                collection.install(output_path, b_temp_path)
-            except AnsibleError as err:
-                if ignore_errors:
-                    display.warning("Failed to install collection %s but skipping due to --ignore-errors being set. "
-                                    "Error: %s" % (to_text(collection), to_text(err)))
-                else:
-                    raise
+        display.display("Process install dependency map")
+        with _display_progress():
+            dependency_map = _build_dependency_map(collections, existing_collections, b_temp_path, apis,
+                                                   validate_certs, force, force_deps, no_deps)
+
+        display.display("Starting collection install process")
+        with _display_progress():
+            for collection in dependency_map.values():
+                try:
+                    collection.install(output_path, b_temp_path)
+                except AnsibleError as err:
+                    if ignore_errors:
+                        display.warning("Failed to install collection %s but skipping due to --ignore-errors being set. "
+                                        "Error: %s" % (to_text(collection), to_text(err)))
+                    else:
+                        raise
 
 
 def validate_collection_name(name):
@@ -491,6 +504,64 @@ def _tarfile_extract(tar, member):
     tar_obj.close()
 
 
+@contextmanager
+def _display_progress():
+    def progress(display_queue, actual_display):
+        actual_display.debug("Starting display_progress display thread")
+        t = threading.current_thread()
+
+        while True:
+            for c in "|/-\\":
+                actual_display.display(c + "\b", newline=False)
+                time.sleep(0.1)
+
+                # Display a message from the main thread
+                while True:
+                    try:
+                        method, args, kwargs = display_queue.get(block=False, timeout=0.1)
+                    except queue.Empty:
+                        break
+                    else:
+                        func = getattr(actual_display, method)
+                        func(*args, **kwargs)
+
+                if getattr(t, "finish", False):
+                    actual_display.debug("Received end signal for display_progress display thread")
+                    return
+
+    class DisplayThread(object):
+
+        def __init__(self, display_queue):
+            self.display_queue = display_queue
+
+        def __getattr__(self, attr):
+            def call_display(*args, **kwargs):
+                self.display_queue.put((attr, args, kwargs))
+
+            return call_display
+
+    # Temporary override the global display class with our own which add the calls to a queue for the thread to call.
+    global display
+    old_display = display
+    try:
+        display_queue = queue.Queue()
+        display = DisplayThread(display_queue)
+        t = threading.Thread(target=progress, args=(display_queue, old_display))
+        t.daemon = True
+        t.start()
+
+        try:
+            yield
+        finally:
+            t.finish = True
+            t.join()
+    except Exception:
+        # The exception is re-raised so we can sure the thread is finished and not using the display anymore
+        raise
+    finally:
+        display = old_display
+
+
 def _get_galaxy_yml(b_galaxy_yml_path):
     meta_info = get_collections_galaxy_meta_info()
 
@@ -729,27 +800,35 @@ def _get_mime_data(b_collection_path):
     return b"\r\n".join(form), content_type
 
 
-def _wait_import(task_url, api):
+def _wait_import(task_url, api, timeout):
     headers = api._auth_header()
 
-    display.vvv('Waiting until galaxy import task %s has completed' % task_url)
+    state = 'waiting'
+    resp = None
 
-    wait = 2
-    while True:
-        resp = json.load(open_url(to_native(task_url, errors='surrogate_or_strict'), headers=headers, method='GET',
-                                  validate_certs=api.validate_certs))
+    display.display("Waiting until Galaxy import task %s has completed" % task_url)
+    with _display_progress():
+        start = time.time()
+        wait = 2
 
-        if resp.get('finished_at', None):
-            break
-        elif wait > 20:
-            # We try for a maximum of ~60 seconds before giving up in case something has gone wrong on the server end.
-            raise AnsibleError("Timeout while waiting for the Galaxy import process to finish, check progress at '%s'"
-                               % to_native(task_url))
+        while timeout == 0 or (time.time() - start) < timeout:
+            resp = json.load(open_url(to_native(task_url, errors='surrogate_or_strict'), headers=headers,
+                                      method='GET', validate_certs=api.validate_certs))
+            state = resp.get('state', 'waiting')
+
+            if resp.get('finished_at', None):
+                break
+
+            display.vvv('Galaxy import process has a status of %s, wait %d seconds before trying again'
+                        % (state, wait))
+            time.sleep(wait)
+
+            # poor man's exponential backoff algo so we don't flood the Galaxy API, cap at 30 seconds.
+            wait = min(30, wait * 1.5)
 
-        status = resp.get('status', 'waiting')
-        display.vvv('Galaxy import process has a status of %s, wait %d seconds before trying again' % (status, wait))
-        time.sleep(wait)
-        wait *= 1.5  # poor man's exponential backoff algo so we don't flood the Galaxy API.
+    if state == 'waiting':
+        raise AnsibleError("Timeout while waiting for the Galaxy import process to finish, check progress at '%s'"
+                           % to_native(task_url))
 
     for message in resp.get('messages', []):
         level = message['level']
@@ -760,7 +839,7 @@ def _wait_import(task_url, api):
         else:
             display.vvv("Galaxy import message: %s - %s" % (level, message['message']))
 
-    if resp['state'] == 'failed':
+    if state == 'failed':
         code = to_native(resp['error'].get('code', 'UNKNOWN'))
         description = to_native(resp['error'].get('description', "Unknown error, see %s for more details" % task_url))
         raise AnsibleError("Galaxy import process failed: %s (Code: %s)" % (description, code))

lib/ansible/utils/display.py

@@ -129,7 +129,7 @@ class Display(with_metaclass(Singleton, object)):
                 if os.path.exists(b_cow_path):
                     self.b_cowsay = b_cow_path
 
-    def display(self, msg, color=None, stderr=False, screen_only=False, log_only=False):
+    def display(self, msg, color=None, stderr=False, screen_only=False, log_only=False, newline=True):
         """ Display a message to the user
 
         Note: msg *must* be a unicode string to prevent UnicodeError tracebacks.
@@ -140,7 +140,7 @@ class Display(with_metaclass(Singleton, object)):
             msg = stringc(msg, color)
 
         if not log_only:
-            if not msg.endswith(u'\n'):
+            if not msg.endswith(u'\n') and newline:
                 msg2 = msg + u'\n'
             else:
                 msg2 = msg

test/units/galaxy/test_collection.py

@@ -406,7 +406,7 @@ def test_publish_missing_file():
     expected = to_native("The collection path specified '%s' does not exist." % fake_path)
 
     with pytest.raises(AnsibleError, match=expected):
-        collection.publish_collection(fake_path, None, True)
+        collection.publish_collection(fake_path, None, True, 0)
 
 
 def test_publish_not_a_tarball():
@@ -417,7 +417,7 @@ def test_publish_not_a_tarball():
         temp_file.write(b"\x00")
         temp_file.flush()
         with pytest.raises(AnsibleError, match=expected.format(to_native(temp_file.name))):
-            collection.publish_collection(temp_file.name, None, True)
+            collection.publish_collection(temp_file.name, None, True, 0)
 
 
 def test_publish_no_wait(galaxy_server, collection_artifact, monkeypatch):
@@ -430,7 +430,7 @@ def test_publish_no_wait(galaxy_server, collection_artifact, monkeypatch):
     mock_open.return_value = StringIO(u'{"task":"%s"}' % fake_import_uri)
     expected_form, expected_content_type = collection._get_mime_data(to_bytes(artifact_path))
 
-    collection.publish_collection(artifact_path, galaxy_server, False)
+    collection.publish_collection(artifact_path, galaxy_server, False, 0)
 
     assert mock_open.call_count == 1
     assert mock_open.mock_calls[0][1][0] == '%s/api/v2/collections/' % galaxy_server.api_server
@@ -445,8 +445,9 @@ def test_publish_no_wait(galaxy_server, collection_artifact, monkeypatch):
     assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
         % (artifact_path, galaxy_server.name, galaxy_server.api_server)
     assert mock_display.mock_calls[1][1][0] == \
-        "Collection has been pushed to the Galaxy server, not waiting until import has completed due to --no-wait " \
-        "being set. Import task results can be found at %s" % fake_import_uri
+        "Collection has been pushed to the Galaxy server %s %s, not waiting until import has completed due to " \
+        "--no-wait being set. Import task results can be found at %s"\
+        % (galaxy_server.name, galaxy_server.api_server, fake_import_uri)
 
 
 def test_publish_dont_validate_cert(galaxy_server, collection_artifact):
@@ -455,7 +456,7 @@ def test_publish_dont_validate_cert(galaxy_server, collection_artifact):
 
     mock_open.return_value = StringIO(u'{"task":"https://galaxy.server.com/api/v2/import/1234"}')
 
-    collection.publish_collection(artifact_path, galaxy_server, False)
+    collection.publish_collection(artifact_path, galaxy_server, False, 0)
 
     assert mock_open.call_count == 1
     assert mock_open.mock_calls[0][2]['validate_certs'] is False
@@ -469,7 +470,7 @@ def test_publish_failure(galaxy_server, collection_artifact):
     expected = 'Error when publishing collection (HTTP Code: 500, Message: Unknown error returned by Galaxy ' \
                'server. Code: Unknown)'
     with pytest.raises(AnsibleError, match=re.escape(expected)):
-        collection.publish_collection(artifact_path, galaxy_server, True)
+        collection.publish_collection(artifact_path, galaxy_server, True, 0)
 
 
 def test_publish_failure_with_json_info(galaxy_server, collection_artifact):
@@ -480,16 +481,13 @@ def test_publish_failure_with_json_info(galaxy_server, collection_artifact):
 
     expected = 'Error when publishing collection (HTTP Code: 503, Message: Galaxy error message Code: GWE002)'
     with pytest.raises(AnsibleError, match=re.escape(expected)):
-        collection.publish_collection(artifact_path, galaxy_server, True)
+        collection.publish_collection(artifact_path, galaxy_server, True, 0)
 
 
 def test_publish_with_wait(galaxy_server, collection_artifact, monkeypatch):
     mock_display = MagicMock()
     monkeypatch.setattr(Display, 'display', mock_display)
 
-    mock_vvv = MagicMock()
-    monkeypatch.setattr(Display, 'vvv', mock_vvv)
-
     fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
 
     artifact_path, mock_open = collection_artifact
@@ -499,7 +497,7 @@ def test_publish_with_wait(galaxy_server, collection_artifact, monkeypatch):
         StringIO(u'{"finished_at":"some_time","state":"success"}')
     )
 
-    collection.publish_collection(artifact_path, galaxy_server, True)
+    collection.publish_collection(artifact_path, galaxy_server, True, 0)
 
     assert mock_open.call_count == 2
     assert mock_open.mock_calls[1][1][0] == fake_import_uri
@@ -507,28 +505,23 @@ def test_publish_with_wait(galaxy_server, collection_artifact, monkeypatch):
     assert mock_open.mock_calls[1][2]['validate_certs'] is True
     assert mock_open.mock_calls[1][2]['method'] == 'GET'
 
-    assert mock_display.call_count == 2
+    assert mock_display.call_count == 5
     assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
         % (artifact_path, galaxy_server.name, galaxy_server.api_server)
-    assert mock_display.mock_calls[1][1][0] == 'Collection has been successfully published to the Galaxy server'
-
-    assert mock_vvv.call_count == 3
-    assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
+    assert mock_display.mock_calls[1][1][0] == 'Collection has been published to the Galaxy server %s %s'\
         % (galaxy_server.name, galaxy_server.api_server)
-    assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
+    assert mock_display.mock_calls[2][1][0] == 'Waiting until Galaxy import task %s has completed' % fake_import_uri
+    assert mock_display.mock_calls[4][1][0] == 'Collection has been successfully published and imported to the ' \
+                                               'Galaxy server %s %s' % (galaxy_server.name, galaxy_server.api_server)
 
 
-def test_publish_with_wait_timeout(collection_artifact, monkeypatch):
+def test_publish_with_wait_timeout(galaxy_server, collection_artifact, monkeypatch):
     monkeypatch.setattr(time, 'sleep', MagicMock())
 
-    mock_display = MagicMock()
-    monkeypatch.setattr(Display, 'display', mock_display)
-
     mock_vvv = MagicMock()
     monkeypatch.setattr(Display, 'vvv', mock_vvv)
 
     fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
-    server = 'https://galaxy.server.com'
 
     artifact_path, mock_open = collection_artifact
 
@@ -538,37 +531,26 @@ def test_publish_with_wait_timeout(collection_artifact, monkeypatch):
         StringIO(u'{"finished_at":"some_time","state":"success"}')
     )
 
-    collection.publish_collection(artifact_path, server, 'key', True, True)
+    collection.publish_collection(artifact_path, galaxy_server, True, 60)
 
     assert mock_open.call_count == 3
     assert mock_open.mock_calls[1][1][0] == fake_import_uri
     assert mock_open.mock_calls[1][2]['headers']['Authorization'] == 'Token key'
-    assert mock_open.mock_calls[1][2]['validate_certs'] is False
+    assert mock_open.mock_calls[1][2]['validate_certs'] is True
     assert mock_open.mock_calls[1][2]['method'] == 'GET'
     assert mock_open.mock_calls[2][1][0] == fake_import_uri
     assert mock_open.mock_calls[2][2]['headers']['Authorization'] == 'Token key'
-    assert mock_open.mock_calls[2][2]['validate_certs'] is False
+    assert mock_open.mock_calls[2][2]['validate_certs'] is True
     assert mock_open.mock_calls[2][2]['method'] == 'GET'
 
-    assert mock_display.call_count == 2
-    assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s" % (artifact_path, server)
-    assert mock_display.mock_calls[1][1][0] == 'Collection has been successfully published to the Galaxy server'
-
-    assert mock_vvv.call_count == 3
-    assert mock_vvv.mock_calls[0][1][0] == 'Collection has been pushed to the Galaxy server %s' % server
-    assert mock_vvv.mock_calls[1][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
-    assert mock_vvv.mock_calls[2][1][0] == \
+    assert mock_vvv.call_count == 2
+    assert mock_vvv.mock_calls[1][1][0] == \
         'Galaxy import process has a status of waiting, wait 2 seconds before trying again'
 
 
-def test_publish_with_wait_timeout(galaxy_server, collection_artifact, monkeypatch):
-    galaxy_server.validate_certs = False
-
+def test_publish_with_wait_timeout_failure(galaxy_server, collection_artifact, monkeypatch):
     monkeypatch.setattr(time, 'sleep', MagicMock())
 
-    mock_display = MagicMock()
-    monkeypatch.setattr(Display, 'display', mock_display)
-
     mock_vvv = MagicMock()
     monkeypatch.setattr(Display, 'vvv', mock_vvv)
 
@@ -576,45 +558,33 @@ def test_publish_with_wait_timeout(galaxy_server, collection_artifact, monkeypat
 
     artifact_path, mock_open = collection_artifact
 
-    mock_open.side_effect = (
-        StringIO(u'{"task":"%s"}' % fake_import_uri),
-        StringIO(u'{"finished_at":null}'),
-        StringIO(u'{"finished_at":null}'),
-        StringIO(u'{"finished_at":null}'),
-        StringIO(u'{"finished_at":null}'),
-        StringIO(u'{"finished_at":null}'),
-        StringIO(u'{"finished_at":null}'),
-        StringIO(u'{"finished_at":null}'),
-    )
+    first_call = True
+
+    def open_value(*args, **kwargs):
+        if first_call:
+            return StringIO(u'{"task":"%s"}' % fake_import_uri)
+        else:
+            return StringIO(u'{"finished_at":null}')
+
+    mock_open.side_effect = open_value
 
     expected = "Timeout while waiting for the Galaxy import process to finish, check progress at '%s'" \
         % fake_import_uri
     with pytest.raises(AnsibleError, match=expected):
-        collection.publish_collection(artifact_path, galaxy_server, True)
-
-    assert mock_open.call_count == 8
-    for i in range(7):
-        mock_call = mock_open.mock_calls[i + 1]
-        assert mock_call[1][0] == fake_import_uri
-        assert mock_call[2]['headers']['Authorization'] == 'Token key'
-        assert mock_call[2]['validate_certs'] is False
-        assert mock_call[2]['method'] == 'GET'
-
-    assert mock_display.call_count == 1
-    assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
-        % (artifact_path, galaxy_server.name, galaxy_server.api_server)
+        collection.publish_collection(artifact_path, galaxy_server, True, 2)
 
+    # While the seconds exceed the time we are testing that the exponential backoff gets to 30 and then sits there
+    # Because we mock time.sleep() there should be thousands of calls here
     expected_wait_msg = 'Galaxy import process has a status of waiting, wait {0} seconds before trying again'
-    assert mock_vvv.call_count == 9
-    assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
-        % (galaxy_server.name, galaxy_server.api_server)
-    assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
-    assert mock_vvv.mock_calls[3][1][0] == expected_wait_msg.format(2)
-    assert mock_vvv.mock_calls[4][1][0] == expected_wait_msg.format(3)
-    assert mock_vvv.mock_calls[5][1][0] == expected_wait_msg.format(4)
-    assert mock_vvv.mock_calls[6][1][0] == expected_wait_msg.format(6)
-    assert mock_vvv.mock_calls[7][1][0] == expected_wait_msg.format(10)
-    assert mock_vvv.mock_calls[8][1][0] == expected_wait_msg.format(15)
+    assert mock_vvv.call_count > 9
+    assert mock_vvv.mock_calls[1][1][0] == expected_wait_msg.format(2)
+    assert mock_vvv.mock_calls[2][1][0] == expected_wait_msg.format(3)
+    assert mock_vvv.mock_calls[3][1][0] == expected_wait_msg.format(4)
+    assert mock_vvv.mock_calls[4][1][0] == expected_wait_msg.format(6)
+    assert mock_vvv.mock_calls[5][1][0] == expected_wait_msg.format(10)
+    assert mock_vvv.mock_calls[6][1][0] == expected_wait_msg.format(15)
+    assert mock_vvv.mock_calls[7][1][0] == expected_wait_msg.format(22)
+    assert mock_vvv.mock_calls[8][1][0] == expected_wait_msg.format(30)
 
 
 def test_publish_with_wait_and_failure(galaxy_server, collection_artifact, monkeypatch):
@@ -665,7 +635,7 @@ def test_publish_with_wait_and_failure(galaxy_server, collection_artifact, monke
 
     expected = 'Galaxy import process failed: Because I said so! (Code: GW001)'
     with pytest.raises(AnsibleError, match=re.escape(expected)):
-        collection.publish_collection(artifact_path, galaxy_server, True)
+        collection.publish_collection(artifact_path, galaxy_server, True, 0)
 
     assert mock_open.call_count == 2
     assert mock_open.mock_calls[1][1][0] == fake_import_uri
@@ -673,15 +643,15 @@ def test_publish_with_wait_and_failure(galaxy_server, collection_artifact, monke
     assert mock_open.mock_calls[1][2]['validate_certs'] is True
     assert mock_open.mock_calls[1][2]['method'] == 'GET'
 
-    assert mock_display.call_count == 1
-    assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
+    assert mock_display.call_count == 4
+    assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s"\
         % (artifact_path, galaxy_server.name, galaxy_server.api_server)
-
-    assert mock_vvv.call_count == 4
-    assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
+    assert mock_display.mock_calls[1][1][0] == 'Collection has been published to the Galaxy server %s %s'\
         % (galaxy_server.name, galaxy_server.api_server)
-    assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
-    assert mock_vvv.mock_calls[3][1][0] == 'Galaxy import message: info - Some info'
+    assert mock_display.mock_calls[2][1][0] == 'Waiting until Galaxy import task %s has completed' % fake_import_uri
+
+    assert mock_vvv.call_count == 2
+    assert mock_vvv.mock_calls[1][1][0] == 'Galaxy import message: info - Some info'
 
     assert mock_warn.call_count == 1
     assert mock_warn.mock_calls[0][1][0] == 'Galaxy import warning message: Some warning'
@@ -734,7 +704,7 @@ def test_publish_with_wait_and_failure_and_no_error(galaxy_server, collection_ar
 
     expected = 'Galaxy import process failed: Unknown error, see %s for more details (Code: UNKNOWN)' % fake_import_uri
     with pytest.raises(AnsibleError, match=re.escape(expected)):
-        collection.publish_collection(artifact_path, galaxy_server, True)
+        collection.publish_collection(artifact_path, galaxy_server, True, 0)
 
     assert mock_open.call_count == 2
     assert mock_open.mock_calls[1][1][0] == fake_import_uri
@@ -742,15 +712,15 @@ def test_publish_with_wait_and_failure_and_no_error(galaxy_server, collection_ar
     assert mock_open.mock_calls[1][2]['validate_certs'] is True
     assert mock_open.mock_calls[1][2]['method'] == 'GET'
 
-    assert mock_display.call_count == 1
-    assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
+    assert mock_display.call_count == 4
+    assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s"\
         % (artifact_path, galaxy_server.name, galaxy_server.api_server)
-
-    assert mock_vvv.call_count == 4
-    assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
+    assert mock_display.mock_calls[1][1][0] == 'Collection has been published to the Galaxy server %s %s'\
         % (galaxy_server.name, galaxy_server.api_server)
-    assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
-    assert mock_vvv.mock_calls[3][1][0] == 'Galaxy import message: info - Some info'
+    assert mock_display.mock_calls[2][1][0] == 'Waiting until Galaxy import task %s has completed' % fake_import_uri
+
+    assert mock_vvv.call_count == 2
+    assert mock_vvv.mock_calls[1][1][0] == 'Galaxy import message: info - Some info'
 
     assert mock_warn.call_count == 1
     assert mock_warn.mock_calls[0][1][0] == 'Galaxy import warning message: Some warning'

test/units/galaxy/test_collection_install.py

@@ -672,9 +672,12 @@ def test_install_collections_from_tar(collection_artifact, monkeypatch):
     assert actual_manifest['collection_info']['name'] == 'collection'
     assert actual_manifest['collection_info']['version'] == '0.1.0'
 
-    assert mock_display.call_count == 1
-    assert mock_display.mock_calls[0][1][0] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" \
-        % to_text(collection_path)
+    # Filter out the progress cursor display calls.
+    display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2]]
+    assert len(display_msgs) == 3
+    assert display_msgs[0] == "Process install dependency map"
+    assert display_msgs[1] == "Starting collection install process"
+    assert display_msgs[2] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" % to_text(collection_path)
 
 
 def test_install_collections_existing_without_force(collection_artifact, monkeypatch):
@@ -694,10 +697,14 @@ def test_install_collections_existing_without_force(collection_artifact, monkeyp
     actual_files.sort()
     assert actual_files == [b'README.md', b'docs', b'galaxy.yml', b'playbooks', b'plugins', b'roles']
 
-    assert mock_display.call_count == 2
+    # Filter out the progress cursor display calls.
+    display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2]]
+    assert len(display_msgs) == 4
     # Msg1 is the warning about not MANIFEST.json, cannot really check message as it has line breaks which varies based
     # on the path size
-    assert mock_display.mock_calls[1][1][0] == "Skipping 'ansible_namespace.collection' as it is already installed"
+    assert display_msgs[1] == "Process install dependency map"
+    assert display_msgs[2] == "Starting collection install process"
+    assert display_msgs[3] == "Skipping 'ansible_namespace.collection' as it is already installed"
 
 
 # Makes sure we don't get stuck in some recursive loop
@@ -728,6 +735,9 @@ def test_install_collection_with_circular_dependency(collection_artifact, monkey
     assert actual_manifest['collection_info']['name'] == 'collection'
     assert actual_manifest['collection_info']['version'] == '0.1.0'
 
-    assert mock_display.call_count == 1
-    assert mock_display.mock_calls[0][1][0] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" \
-        % to_text(collection_path)
+    # Filter out the progress cursor display calls.
+    display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2]]
+    assert len(display_msgs) == 3
+    assert display_msgs[0] == "Process install dependency map"
+    assert display_msgs[1] == "Starting collection install process"
+    assert display_msgs[2] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" % to_text(collection_path)