Add tests for ansible-tower/AAP integration with Satellite

Signed-off-by: Gaurav Talreja <gtalreja@redhat.com>

robottelo/hosts.py

@@ -622,6 +622,8 @@ def register(
         force=False,
         insecure=True,
         hostgroup=None,
+        auth_username=None,
+        auth_password=None,
     ):
         """Registers content host to the Satellite or Capsule server
         using a global registration template.
@@ -642,6 +644,8 @@ def register(
         :param force: Register the content host even if it's already registered.
         :param insecure: Don't verify server authenticity.
         :param hostgroup: hostgroup to register with
+        :param auth_username: username required if non-admin user
+        :param auth_password: password required if non-admin user
         :return: SSHCommandResult instance filled with the result of the registration
         """
         options = {
@@ -692,7 +696,20 @@ def register(
             options['force'] = str(force).lower()
 
         self._satellite = target.satellite
-        cmd = target.satellite.cli.HostRegistration.generate_command(options)
+        if auth_username and auth_password:
+            user = target.satellite.cli.User.list({'search': f'login={auth_username}'})
+            if user:
+                register_role = target.satellite.cli.Role.info({'name': 'Register hosts'})
+                target.satellite.cli.User.add_role(
+                    {'id': user[0]['id'], 'role-id': register_role['id']}
+                )
+                cmd = target.satellite.cli.HostRegistration.with_user(
+                    auth_username, auth_password
+                ).generate_command(options)
+            else:
+                raise CLIFactoryError(f'User {auth_username} doesn\'t exist')
+        else:
+            cmd = target.satellite.cli.HostRegistration.generate_command(options)
         return self.execute(cmd.strip('\n'))
 
     def api_register(self, target, **kwargs):

tests/foreman/cli/test_ansible.py

@@ -11,11 +11,13 @@
 
 from time import sleep
 
+import awxkit
 from fauxfactory import gen_string
 import pytest
 
 from robottelo.config import settings
 from robottelo.exceptions import CLIFactoryError
+from robottelo.utils.issue_handlers import is_open
 
 
 def assert_job_invocation_result(
@@ -561,3 +563,126 @@ def test_positive_install_ansible_collection(
         assert result['success'] == '1'
         collection_path = client.execute('ls ~/ansible_collections').stdout
         assert 'oasis_roles' in collection_path
+
+
+@pytest.mark.upgrade
+class TestAnsibleAAPIntegration:
+    """Test class for Satellite integration with Ansible Automation Controller
+
+    :CaseComponent: Ansible-ConfigurationManagement
+    """
+
+    @pytest.fixture(scope='class')
+    def aap_client(self):
+        client = awxkit.api.client.Connection(settings.AAP_INTEGRATION.AAP_URL)
+        client.login(settings.AAP_INTEGRATION.USERNAME, settings.AAP_INTEGRATION.PASSWORD)
+        yield client
+        client.logout()
+
+    @pytest.mark.parametrize('auth_type', ['admin', 'non-admin'])
+    @pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
+    def test_positive_dynamic_inventory(
+        self,
+        request,
+        target_sat,
+        module_org,
+        module_location,
+        module_ak_with_cv,
+        aap_client,
+        rhel_contenthost,
+        auth_type,
+    ):
+        """Test whether Ansible collection can be installed via Ansible REX
+
+        :id: ad25aee5-4ea3-4743-a301-1c6271856g19
+
+        :steps:
+            1. Upload a manifest.
+            2. Register content host to Satellite with REX setup
+            3. Enable Ansible repo on content host.
+            4. Install ansible or ansible-core package
+            5. Run REX job to install Ansible collection on content host.
+
+        :expectedresults: Ansible collection can be installed on content host via REX.
+        """
+        creds_name = settings.AAP_INTEGRATION.satellite_credentials
+        inventory_name = settings.AAP_INTEGRATION.satellite_inventory
+
+        password = settings.server.admin_password
+        if auth_type == 'admin':
+            login = gen_string('alpha')
+            user = target_sat.api.User(
+                admin=True,
+                login=login,
+                password=password,
+                organization=[module_org],
+                location=[module_location],
+            ).create()
+        else:
+            login = gen_string('alpha')
+            role = target_sat.api.Role(
+                name=gen_string('alpha'), location=[module_location], organization=[module_org]
+            ).create()
+            for perm in ['view_hosts', 'view_hostgroups', 'view_facts']:
+                permission = target_sat.api.Permission().search(query={'search': f'name={perm}'})
+                target_sat.api.Filter(permission=permission, role=role).create()
+            user = target_sat.api.User(
+                role=[role],
+                admin=False,
+                login=login,
+                password=password,
+                organization=[module_org],
+                location=[module_location],
+            ).create()
+            target_sat.cli.User.add_role(
+                {'login': user.login, 'role': 'Ansible Tower Inventory Reader'}
+            )
+        result = rhel_contenthost.register(
+            module_org,
+            module_location,
+            module_ak_with_cv.name,
+            target_sat,
+            auth_username=user.login,
+            auth_password=password,
+            force=True,
+        )
+        assert result.status == 0, f'Failed to register host: {result.stderr}'
+
+        if is_open('SAT-28613') and auth_type == 'non-admin':
+            host = rhel_contenthost.nailgun_host
+            host.location = module_location
+            host.update(['location'])
+
+        # Find the Satellite credentials in AAP and update it for target_sat.hostname and user credentials
+        creds_list = aap_client.get(
+            '/api/v2/credentials/', query_parameters=f'name={creds_name}'
+        ).json()
+        new_creds = {
+            'inputs': {
+                'host': f'https://{target_sat.hostname}',
+                'username': login,
+                'password': settings.server.admin_password,
+            }
+        }
+        response = aap_client.patch(
+            f'/api/v2/credentials/{creds_list["results"][0]["id"]}/', json=new_creds
+        )
+        assert response.ok
+
+        # Find the Satellite inventory in AAP and update it for target_sat and user credentials
+        inv_list = aap_client.get(
+            '/api/v2/inventories/', query_parameters=f'name={inventory_name}'
+        ).json()
+        inv_source_list = aap_client.get(
+            f'/api/v2/inventories/{inv_list["results"][0]["id"]}/inventory_sources/'
+        ).json()
+        sync_response = aap_client.post(
+            f'/api/v2/inventory_sources/{inv_source_list["results"][0]["id"]}/update/'
+        )
+        assert sync_response.ok
+        sleep(180)
+        # Find the hosts in Satellite inventory in AAP and verify if target_sat is listed in inventory
+        hosts_list = aap_client.get(
+            f'/api/v2/inventories/{inv_list["results"][0]["id"]}/hosts/'
+        ).json()
+        assert rhel_contenthost.hostname in [host['name'] for host in hosts_list['results']]