verify job invocation for different user should show after job schedule

tests/foreman/api/test_ansible.py

@@ -602,3 +602,94 @@ def test_negative_ansible_job_timeout_to_kill(
         assert [i['output'] for i in result if i['output'] == termination_msg]
         assert [i['output'] for i in result if i['output'] == 'StandardError: Job execution failed']
         assert [i['output'] for i in result if i['output'] == 'Exit status: 120']
+
+    @pytest.mark.tier2
+    @pytest.mark.no_containers
+    @pytest.mark.rhel_ver_match('8')
+    @pytest.mark.parametrize('setting_update', ['remote_execution_effective_user'], indirect=True)
+    def test_positive_privilege_escalation_ansible_playbook(
+        self,
+        target_sat,
+        rhel_contenthost,
+        module_org,
+        module_location,
+        module_activation_key,
+        setting_update,
+    ):
+        """Privilege escalation defined inside ansible playbook tasks is working
+        when executing the playbook via Remote Execution
+
+        :id: 8c63fd1a-2121-4cce-9ec1-ae12817c9cc4
+
+        :steps:
+            1. Register a RHEL host to Satellite.
+            2. Setup a user on that host.
+            3. Create a playbook.
+            4. Set the SSH user to the created user, and unset the Effective user.
+            5. Run the playbook.
+
+        :expectedresults: In the playbook, created user is expected instead root user.
+        """
+        playbook = '''
+                    ---
+                    - name: Test Play
+                      hosts: all
+                      gather_facts: false
+                      tasks:
+                        - name: Check current user
+                          command: bash -c "whoami"
+                          register: def_user
+                        - debug:
+                            var: def_user.stdout
+                        - name: Check become user
+                          command: bash -c "whoami"
+                          become: true
+                          become_user: testing
+                          register: bec_user
+                        - debug:
+                            var: bec_user.stdout
+                '''
+        result = rhel_contenthost.register(
+            module_org, module_location, module_activation_key.name, target_sat
+        )
+        assert result.status == 0, f'Failed to register host: {result.stderr}'
+        assert rhel_contenthost.username == 'root'
+        one = rhel_contenthost.execute('useradd rexuser')
+        assert one.status == 0
+        two = rhel_contenthost.execute('echo dog8code | passwd rexuser --stdin')
+        assert 'passwd: all authentication tokens updated successfully.' in two.stdout
+        three = rhel_contenthost.execute(
+            'echo "rexuser   ALL=NOPASSWD:   ALL" | tee -a /etc/sudoers.d/rexuser'
+        )
+        assert three.status == 0
+        four = rhel_contenthost.execute('useradd testing')
+        assert four.status == 0
+        five = rhel_contenthost.execute('echo dog8code | passwd testing --stdin')
+        assert 'passwd: all authentication tokens updated successfully.' in five.stdout
+        six = rhel_contenthost.execute('su - testing -c "whoami"')
+        assert 'testing' in six.stdout or six.status == 0
+        setting_update.value = ''
+        setting_update = setting_update.update({'value'})
+        assert setting_update.value == ''
+        template_id = (
+            target_sat.api.JobTemplate()
+            .search(query={'search': 'name="Ansible - Run playbook"'})[0]
+            .id
+        )
+        job = target_sat.api.JobInvocation().run(
+            synchronous=False,
+            data={
+                'job_category': 'Ansible Playbook',
+                'job_template_id': template_id,
+                'targeting_type': 'static_query',
+                'search_query': f'name = {rhel_contenthost.hostname}',
+                'inputs': {'playbook': playbook},
+            },
+        )
+        task = target_sat.wait_for_tasks(
+            f'resource_type = JobInvocation and resource_id = {job["id"]}',
+            poll_timeout=1000,
+            must_succeed=False,
+        )
+        assert '"def_user.stdout": "root"' in task[0].humanized['output']
+        assert '"bec_user.stdout": "testing"' in task[0].humanized['output']