test rex using values from global parameters

conftest.py

@@ -43,6 +43,7 @@
     'pytest_fixtures.component.contentview',
     'pytest_fixtures.component.domain',
     'pytest_fixtures.component.discovery',
+    'pytest_fixtures.component.global_params',
     'pytest_fixtures.component.host',
     'pytest_fixtures.component.hostgroup',
     'pytest_fixtures.component.http_proxy',

pytest_fixtures/component/global_params.py

@@ -0,0 +1,35 @@
+# Settings Fixtures
+import pytest
+
+
+@pytest.fixture
+def multi_global_param_update(request, target_sat):
+    """
+    This fixture is used to alter multiple global parameters in one batch.
+    """
+    key_vals = request.param
+    param_objects = []
+    for key_val in key_vals:
+        param, new_value = tuple(key_val.split('=')) if '=' in key_val else (key_val, None)
+        existing_params = target_sat.api.CommonParameter().search(query={'search': f'name={param}'})
+        if len(existing_params) > 0:
+            param_object = existing_params[0]
+            cleanup = False
+            default_param_value = param_object.value
+            if new_value is not None:
+                param_object.value = new_value
+                param_object.update({'value'})
+        else:
+            param_object = target_sat.api.CommonParameter(name=param, value=new_value).create()
+            cleanup = True
+            default_param_value = new_value
+        param_objects.append(
+            {'object': param_object, 'default': default_param_value, 'cleanup': cleanup}
+        )
+    yield [item['object'] for item in param_objects]
+    for item in param_objects:
+        if item['cleanup']:
+            item['object'].delete()
+        else:
+            item['object'].value = item['default']
+            item['object'].update({'value'})

pytest_fixtures/component/settings.py

@@ -19,3 +19,24 @@ def setting_update(request, target_sat):
     yield setting_object
     setting_object.value = default_setting_value
     setting_object.update({'value'})
+
+
+@pytest.fixture
+def multi_setting_update(request, target_sat):
+    """
+    This fixture is used to alter multiple settings in one batch.
+    """
+    key_vals = request.param
+    setting_objects = []
+    for key_val in key_vals:
+        setting, new_value = tuple(key_val.split('=')) if '=' in key_val else (key_val, None)
+        setting_object = target_sat.api.Setting().search(query={'search': f'name={setting}'})[0]
+        default_setting_value = '' if setting_object.value is None else setting_object.value
+        if new_value is not None:
+            setting_object.value = new_value
+            setting_object.update({'value'})
+        setting_objects.append({'object': setting_object, 'default': default_setting_value})
+    yield [item['object'] for item in setting_objects]
+    for item in setting_objects:
+        item['object'].value = item['object'].default
+        item['object'].update({'value'})

tests/foreman/cli/test_remoteexecution.py

@@ -311,6 +311,96 @@ def test_positive_run_job_effective_user(
                 }
             )
 
+    @pytest.mark.tier3
+    @pytest.mark.parametrize(
+        'multi_global_param_update',
+        [
+            [
+                'remote_execution_effective_user',
+                'remote_execution_effective_user_password',
+                'remote_execution_effective_user_method',
+                'remote_execution_ssh_user',
+                'remote_execution_ssh_password',
+            ],
+        ],
+        ids=["global-param-sudo"],
+        indirect=True,
+    )
+    @pytest.mark.rhel_ver_list([9])
+    def test_positive_run_job_global_ssh_user(
+        self,
+        rex_contenthost,
+        module_target_sat,
+        multi_global_param_update,
+        module_org,
+        module_ak_with_cv,
+    ):
+        """Run default job template with global ssh user, effective user and sudo
+
+        :id: 0adaf5a2-930a-4050-863b-62456234ce8c
+
+        :verifies: SAT-28443
+
+        :expectedresults: Verify global paremeters are used to set up rex during registration
+        :parametrized: yes
+        """
+        client = rex_contenthost
+
+        # configure global settings
+        ssh_username = f"sshuser_{gen_string('alpha')}"
+        ssh_password = gen_string('alpha')
+        username = f"effuser_{gen_string('alpha')}"
+        password = gen_string('alpha')
+        filename = gen_string('alpha')
+        multi_global_param_update[0].value = username
+        multi_global_param_update[1].value = password
+        multi_global_param_update[2].value = 'sudo'
+        multi_global_param_update[3].value = ssh_username
+        multi_global_param_update[4].value = ssh_password
+        for param in multi_global_param_update:
+            param.update({'value'})
+
+        # add users to the host
+        make_user_job = module_target_sat.cli_factory.job_invocation(
+            {
+                'job-template': 'Run Command - Script Default',
+                'inputs': f"command=useradd {ssh_username}; echo {ssh_username}:{ssh_password} | chpasswd; useradd {username}; echo {username}:{password} | chpasswd",
+                'search-query': f"name ~ {client.hostname}",
+                'effective-user': 'root',
+                'ssh-user': 'root',
+                'description-format': 'adding users',
+            }
+        )
+        assert_job_invocation_result(module_target_sat, make_user_job['id'], client.hostname)
+
+        # re-register host to run the remote_execution_ssh_keys snippet with new defaults
+        client.register(module_org, None, module_ak_with_cv.name, module_target_sat, force=True)
+
+        # check the sudoers.d entry was created by the snippet
+        result = client.execute(f'''stat -c "%a %n" /etc/sudoers.d/{ssh_username}''')
+        assert '440' in result.stdout
+
+        # create a file using global ssh-user and effective-user settings
+        invocation_command = module_target_sat.cli_factory.job_invocation(
+            {
+                'job-template': 'Run Command - Script Default',
+                'inputs': f"command=touch /home/{username}/{filename}",
+                'search-query': f"name ~ {client.hostname}",
+            }
+        )
+        assert_job_invocation_result(module_target_sat, invocation_command['id'], client.hostname)
+        # check the file owner
+        result = client.execute(
+            f'''stat -c '%U' /home/{username}/{filename}''',
+        )
+        # assert the file is owned by the effective user
+        assert username == result.stdout.strip('\n')
+        result = client.execute(
+            f'''stat -c '%G' /home/{username}/{filename}''',
+        )
+        # assert the file is in the effective user's group
+        assert username == result.stdout.strip('\n')
+
     @pytest.mark.tier3
     @pytest.mark.e2e
     @pytest.mark.rhel_ver_match('[^6].*')