Fix connection/become task loop settings (#78565) (#78935)

* Fix connection/become task loop settings

* Remove old code

(cherry picked from commit ba6da65a0f3baefda7a058ebbd0a8dcafb8512f5)

8 files changed, 119 insertions, 11 deletions

diff --git a/changelogs/fragments/become-loop-setting.yml b/changelogs/fragments/become-loop-setting.yml
new file mode 100644
index 0000000000..7aa295fa9e
--- /dev/null
+++ b/changelogs/fragments/become-loop-setting.yml
@@ -0,0 +1,3 @@
+bugfixes:
+- Fix reusing a connection in a task loop that uses a redirected or aliased name - https://github.com/ansible/ansible/issues/78425
+- Fix setting become activation in a task loop - https://github.com/ansible/ansible/issues/78425
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 2531c15989..6718f31085 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -551,6 +551,7 @@ class TaskExecutor:
             # if connection is reused, its _play_context is no longer valid and needs
             # to be replaced with the one templated above, in case other data changed
             self._connection._play_context = self._play_context
+            self._set_become_plugin(cvars, templar, self._connection)
 
         plugin_vars = self._set_connection_options(cvars, templar)
 
@@ -952,6 +953,14 @@ class TaskExecutor:
         if not connection:
             raise AnsibleError("the connection plugin '%s' was not found" % conn_type)
 
+        self._set_become_plugin(cvars, templar, connection)
+
+        # Also backwards compat call for those still using play_context
+        self._play_context.set_attributes_from_plugin(connection)
+
+        return connection
+
+    def _set_become_plugin(self, cvars, templar, connection):
         # load become plugin if needed
         if cvars.get('ansible_become') is not None:
             become = boolean(templar.template(cvars['ansible_become']))
@@ -964,16 +973,22 @@ class TaskExecutor:
             else:
                 become_plugin = self._get_become(self._task.become_method)
 
-            try:
-                connection.set_become_plugin(become_plugin)
-            except AttributeError:
-                # Older connection plugin that does not support set_become_plugin
-                pass
+        else:
+            # If become is not enabled on the task it needs to be removed from the connection plugin
+            # https://github.com/ansible/ansible/issues/78425
+            become_plugin = None
 
+        try:
+            connection.set_become_plugin(become_plugin)
+        except AttributeError:
+            # Older connection plugin that does not support set_become_plugin
+            pass
+
+        if become_plugin:
             if getattr(connection.become, 'require_tty', False) and not getattr(connection, 'has_tty', False):
                 raise AnsibleError(
                     "The '%s' connection does not provide a TTY which is required for the selected "
-                    "become plugin: %s." % (conn_type, become_plugin.name)
+                    "become plugin: %s." % (connection._load_name, become_plugin.name)
                 )
 
             # Backwards compat for connection plugins that don't support become plugins
@@ -981,11 +996,6 @@ class TaskExecutor:
             # AttributeError above later
             self._play_context.set_become_plugin(become_plugin.name)
 
-        # Also backwards compat call for those still using play_context
-        self._play_context.set_attributes_from_plugin(connection)
-
-        return connection
-
     def _set_plugin_options(self, plugin_type, variables, templar, task_keys):
         try:
             plugin = getattr(self._connection, '_%s' % plugin_type)
diff --git a/test/integration/targets/loop-connection/aliases b/test/integration/targets/loop-connection/aliases
new file mode 100644
index 0000000000..498fedd558
--- /dev/null
+++ b/test/integration/targets/loop-connection/aliases
@@ -0,0 +1,2 @@
+shippable/posix/group4
+context/controller
diff --git a/test/integration/targets/loop-connection/collections/ansible_collections/ns/name/meta/runtime.yml b/test/integration/targets/loop-connection/collections/ansible_collections/ns/name/meta/runtime.yml
new file mode 100644
index 0000000000..09322a9deb
--- /dev/null
+++ b/test/integration/targets/loop-connection/collections/ansible_collections/ns/name/meta/runtime.yml
@@ -0,0 +1,4 @@
+plugin_routing:
+  connection:
+    redirected_dummy:
+      redirect: ns.name.dummy
\ No newline at end of file
diff --git a/test/integration/targets/loop-connection/collections/ansible_collections/ns/name/plugins/connection/dummy.py b/test/integration/targets/loop-connection/collections/ansible_collections/ns/name/plugins/connection/dummy.py
new file mode 100644
index 0000000000..cb14991fc3
--- /dev/null
+++ b/test/integration/targets/loop-connection/collections/ansible_collections/ns/name/plugins/connection/dummy.py
@@ -0,0 +1,50 @@
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = '''
+name: dummy
+short_description: Used for loop-connection tests
+description:
+- See above
+author: ansible (@core)
+'''
+
+from ansible.errors import AnsibleError
+from ansible.plugins.connection import ConnectionBase
+
+
+class Connection(ConnectionBase):
+
+    transport = 'ns.name.dummy'
+
+    def __init__(self, *args, **kwargs):
+        self._cmds_run = 0
+        super().__init__(*args, **kwargs)
+
+    @property
+    def connected(self):
+        return True
+
+    def _connect(self):
+        return
+
+    def exec_command(self, cmd, in_data=None, sudoable=True):
+        if 'become_test' in cmd:
+            stderr = f"become - {self.become.name if self.become else None}"
+
+        elif 'connected_test' in cmd:
+            self._cmds_run += 1
+            stderr = f"ran - {self._cmds_run}"
+
+        else:
+            raise AnsibleError(f"Unknown test cmd {cmd}")
+
+        return 0, cmd.encode(), stderr.encode()
+
+    def put_file(self, in_path, out_path):
+        return
+
+    def fetch_file(self, in_path, out_path):
+        return
+
+    def close(self):
+        return
diff --git a/test/integration/targets/loop-connection/main.yml b/test/integration/targets/loop-connection/main.yml
new file mode 100644
index 0000000000..fbffe309dd
--- /dev/null
+++ b/test/integration/targets/loop-connection/main.yml
@@ -0,0 +1,33 @@
+- hosts: localhost
+  gather_facts: false
+  tasks:
+  - name: test changing become activation on the same connection
+    raw: become_test
+    register: become_test
+    become: '{{ item }}'
+    vars:
+      ansible_connection: ns.name.dummy
+    loop:
+    - true
+    - false
+
+  - assert:
+      that:
+      - become_test.results[0].stderr == "become - sudo"
+      - become_test.results[0].stdout.startswith("sudo ")
+      - become_test.results[1].stderr == "become - None"
+      - become_test.results[1].stdout == "become_test"
+
+  - name: test loop reusing connection with redirected plugin name
+    raw: connected_test
+    register: connected_test
+    vars:
+      ansible_connection: ns.name.redirected_dummy
+    loop:
+    - 1
+    - 2
+
+  - assert:
+      that:
+      - connected_test.results[0].stderr == "ran - 1"
+      - connected_test.results[1].stderr == "ran - 2"
\ No newline at end of file
diff --git a/test/integration/targets/loop-connection/runme.sh b/test/integration/targets/loop-connection/runme.sh
new file mode 100755
index 0000000000..db4031df3b
--- /dev/null
+++ b/test/integration/targets/loop-connection/runme.sh
@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+set -eux -o pipefail
+
+ansible-playbook main.yml "$@"
diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py
index f71629781f..315d26ae3a 100644
--- a/test/units/executor/test_task_executor.py
+++ b/test/units/executor/test_task_executor.py
@@ -315,6 +315,7 @@ class TestTaskExecutor(unittest.TestCase):
         mock_task = MagicMock()
         mock_task.action = 'mock.action'
         mock_task.args = dict()
+        mock_task.become = False
         mock_task.retries = 0
         mock_task.delay = -1
         mock_task.register = 'foo'