Merge branch 'master' into api-errata-eval

.github/workflows/auto_cherry_pick_merge.yaml

@@ -67,7 +67,7 @@ jobs:
 
       - id: automerge
         name: Auto merge of cherry-picked PRs.
-        uses: "pascalgn/automerge-action@v0.15.6"
+        uses: "pascalgn/automerge-action@v0.16.0"
         env:
           GITHUB_TOKEN: "${{ secrets.CHERRYPICK_PAT }}"
           MERGE_LABELS: "AutoMerge_Cherry_Picked, Auto_Cherry_Picked"

.github/workflows/dependency_merge.yml

@@ -61,7 +61,7 @@ jobs:
 
       - id: automerge
         name: Auto merge of dependabot PRs.
-        uses: "pascalgn/automerge-action@v0.15.6"
+        uses: "pascalgn/automerge-action@v0.16.0"
         env:
           GITHUB_TOKEN: "${{ secrets.CHERRYPICK_PAT }}"
           MERGE_LABELS: "dependencies"

pyproject.toml

@@ -20,6 +20,7 @@ target-version = "py311"
 fixable = ["ALL"]
 
 select = [
+    "B",  # bugbear
     # "C90", # mccabe
     "E",  # pycodestyle
     "F",  # flake8

pytest_fixtures/core/ui.py

@@ -2,6 +2,7 @@
 import pytest
 from requests.exceptions import HTTPError
 
+from robottelo.hosts import Satellite
 from robottelo.logging import logger
 
 
@@ -70,3 +71,31 @@ def test_foo(autosession):
     """
     with target_sat.ui_session(test_name, ui_user.login, ui_user.password) as started_session:
         yield started_session
+
+
+@pytest.fixture(autouse=True)
+def ui_session_record_property(request, record_property):
+    """
+    Autouse fixture to set the record_property attribute for Satellite instances in the test.
+
+    This fixture iterates over all fixtures in the current test node
+    (excluding the current fixture) and sets the record_property attribute
+    for instances of the Satellite class.
+
+    Args:
+        request: The pytest request object.
+        record_property: The value to set for the record_property attribute.
+    """
+    test_directories = [
+        'tests/foreman/destructive',
+        'tests/foreman/ui',
+        'tests/foreman/sanity',
+        'tests/foreman/virtwho',
+    ]
+    test_file_path = request.node.fspath.strpath
+    if any(directory in test_file_path for directory in test_directories):
+        for fixture in request.node.fixturenames:
+            if request.fixturename != fixture:
+                if isinstance(request.getfixturevalue(fixture), Satellite):
+                    sat = request.getfixturevalue(fixture)
+                    sat.record_property = record_property

robottelo/cli/base.py

@@ -169,15 +169,9 @@ def _get_username_password(cls, username=None, password=None):
 
         """
         if username is None:
-            try:
-                username = getattr(cls, 'foreman_admin_username')
-            except AttributeError:
-                username = settings.server.admin_username
+            username = getattr(cls, 'foreman_admin_username', settings.server.admin_username)
         if password is None:
-            try:
-                password = getattr(cls, 'foreman_admin_password')
-            except AttributeError:
-                password = settings.server.admin_password
+            password = getattr(cls, 'foreman_admin_password', settings.server.admin_password)
 
         return (username, password)
 

robottelo/cli/hammer.py

@@ -42,7 +42,7 @@ def parse_csv(output):
     # Generate the key names, spaces will be converted to dashes "-"
     keys = [_normalize(header) for header in next(reader)]
     # For each entry, create a dict mapping each key with each value
-    return [dict(zip(keys, values)) for values in reader if len(values) > 0]
+    return [dict(zip(keys, values, strict=True)) for values in reader if len(values) > 0]
 
 
 def parse_help(output):

robottelo/host_helpers/api_factory.py

@@ -469,14 +469,14 @@ def create_role_permissions(
                     if entity_permission.name != name:
                         raise self._satellite.api.APIResponseError(
                             'the returned permission is different from the'
-                            ' requested one "{} != {}"'.format(entity_permission.name, name)
+                            f' requested one "{entity_permission.name} != {name}"'
                         )
                     permissions_entities.append(entity_permission)
             else:
                 if not permissions_name:
                     raise ValueError(
-                        'resource type "{}" empty. You must select at'
-                        ' least one permission'.format(resource_type)
+                        f'resource type "{resource_type}" empty. You must select at'
+                        ' least one permission'
                     )
 
                 resource_type_permissions_entities = self._satellite.api.Permission().search(
@@ -576,8 +576,8 @@ def satellite_setting(self, key_val: str):
                 setting = self._satellite.api.Setting().search(
                     query={'search': f'name={name.strip()}'}
                 )[0]
-            except IndexError:
-                raise KeyError(f'The setting {name} in not available in satellite.')
+            except IndexError as err:
+                raise KeyError(f'The setting {name} in not available in satellite.') from err
             old_value = setting.value
             setting.value = value.strip()
             setting.update({'value'})

robottelo/host_helpers/capsule_mixins.py

@@ -61,10 +61,12 @@ def wait_for_tasks(
             raise AssertionError(f"No task was found using query '{search_query}'")
         return tasks
 
-    def wait_for_sync(self, timeout=600, start_time=datetime.utcnow()):
+    def wait_for_sync(self, timeout=600, start_time=None):
         """Wait for capsule sync to finish and assert the sync task succeeded"""
         # Assert that a task to sync lifecycle environment to the capsule
         # is started (or finished already)
+        if start_time is None:
+            start_time = datetime.utcnow()
         logger.info(f"Waiting for capsule {self.hostname} sync to finish ...")
         sync_status = self.nailgun_capsule.content_get_sync()
         logger.info(f"Active tasks {sync_status['active_sync_tasks']}")

robottelo/host_helpers/cli_factory.py

@@ -4,7 +4,7 @@
 example: my_satellite.cli_factory.make_org()
 """
 import datetime
-from functools import lru_cache, partial
+from functools import partial
 import inspect
 import os
 from os import chmod
@@ -14,6 +14,7 @@
 from time import sleep
 
 from box import Box
+from cachetools import cachedmethod
 from fauxfactory import (
     gen_alpha,
     gen_alphanumeric,
@@ -59,9 +60,9 @@ def create_object(cli_object, options, values=None, credentials=None):
             'Failed to create {} with data:\n{}\n{}'.format(
                 cli_object.__name__, pprint.pformat(options, indent=2), err.msg
             )
-        )
+        ) from err
     # Sometimes we get a list with a dictionary and not a dictionary.
-    if type(result) is list and len(result) > 0:
+    if isinstance(result, list) and len(result) > 0:
         result = result[0]
     return Box(result)
 
@@ -294,7 +295,7 @@ def _evaluate_functions(self, iterable):
                 if not key.startswith('_')
             }
 
-    @lru_cache
+    @cachedmethod
     def _find_entity_class(self, entity_name):
         entity_name = entity_name.replace('_', '').lower()
         for name, class_obj in self._satellite.cli.__dict__.items():
@@ -394,8 +395,8 @@ def make_product_wait(self, options=None, wait_for=5):
                 product = self._satellite.cli.Product.info(
                     {'name': options.get('name'), 'organization-id': options.get('organization-id')}
                 )
-            except CLIReturnCodeError:
-                raise err
+            except CLIReturnCodeError as nested_err:
+                raise nested_err from err
             if not product:
                 raise err
         return product
@@ -503,7 +504,7 @@ def make_proxy(self, options=None):
                     args['url'] = url
                     return create_object(self._satellite.cli.Proxy, args, options)
             except CapsuleTunnelError as err:
-                raise CLIFactoryError(f'Failed to create ssh tunnel: {err}')
+                raise CLIFactoryError(f'Failed to create ssh tunnel: {err}') from None
         args['url'] = options['url']
         return create_object(self._satellite.cli.Proxy, args, options)
 
@@ -569,7 +570,7 @@ def activationkey_add_subscription_to_repo(self, options=None):
                 except CLIReturnCodeError as err:
                     raise CLIFactoryError(
                         f'Failed to add subscription to activation key\n{err.msg}'
-                    )
+                    ) from err
 
     def setup_org_for_a_custom_repo(self, options=None):
         """Sets up Org for the given custom repo by:
@@ -608,7 +609,7 @@ def setup_org_for_a_custom_repo(self, options=None):
         try:
             self._satellite.cli.Repository.synchronize({'id': custom_repo['id']})
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to synchronize repository\n{err.msg}')
+            raise CLIFactoryError(f'Failed to synchronize repository\n{err.msg}') from err
         # Create CV if needed and associate repo with it
         if options.get('content-view-id') is None:
             cv_id = self.make_content_view({'organization-id': org_id})['id']
@@ -619,12 +620,14 @@ def setup_org_for_a_custom_repo(self, options=None):
                 {'id': cv_id, 'organization-id': org_id, 'repository-id': custom_repo['id']}
             )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to add repository to content view\n{err.msg}')
+            raise CLIFactoryError(f'Failed to add repository to content view\n{err.msg}') from err
         # Publish a new version of CV
         try:
             self._satellite.cli.ContentView.publish({'id': cv_id})
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to publish new version of content view\n{err.msg}')
+            raise CLIFactoryError(
+                f'Failed to publish new version of content view\n{err.msg}'
+            ) from err
         # Get the version id
         cv_info = self._satellite.cli.ContentView.info({'id': cv_id})
         assert len(cv_info['versions']) > 0
@@ -642,7 +645,9 @@ def setup_org_for_a_custom_repo(self, options=None):
                     }
                 )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to promote version to next environment\n{err.msg}')
+            raise CLIFactoryError(
+                f'Failed to promote version to next environment\n{err.msg}'
+            ) from err
         # Create activation key if needed and associate content view with it
         if options.get('activationkey-id') is None:
             activationkey_id = self.make_activation_key(
@@ -661,7 +666,9 @@ def setup_org_for_a_custom_repo(self, options=None):
                     {'content-view-id': cv_id, 'id': activationkey_id, 'organization-id': org_id}
                 )
             except CLIReturnCodeError as err:
-                raise CLIFactoryError(f'Failed to associate activation-key with CV\n{err.msg}')
+                raise CLIFactoryError(
+                    f'Failed to associate activation-key with CV\n{err.msg}'
+                ) from err
 
         # Add custom_product subscription to activation-key, if SCA mode is disabled
         if self._satellite.is_sca_mode_enabled(org_id) is False:
@@ -720,10 +727,13 @@ def _setup_org_for_a_rh_repo(self, options=None):
         # If manifest does not exist, clone and upload it
         if len(self._satellite.cli.Subscription.exists({'organization-id': org_id})) == 0:
             with clone() as manifest:
-                try:
-                    self._satellite.upload_manifest(org_id, manifest.content)
-                except CLIReturnCodeError as err:
-                    raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}')
+                self._satellite.put(manifest.path, manifest.name)
+            try:
+                self._satellite.cli.Subscription.upload(
+                    {'file': manifest.name, 'organization-id': org_id}
+                )
+            except CLIReturnCodeError as err:
+                raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}') from err
         # Enable repo from Repository Set
         try:
             self._satellite.cli.RepositorySet.enable(
@@ -736,7 +746,7 @@ def _setup_org_for_a_rh_repo(self, options=None):
                 }
             )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to enable repository set\n{err.msg}')
+            raise CLIFactoryError(f'Failed to enable repository set\n{err.msg}') from err
         # Fetch repository info
         try:
             rhel_repo = self._satellite.cli.Repository.info(
@@ -747,7 +757,7 @@ def _setup_org_for_a_rh_repo(self, options=None):
                 }
             )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to fetch repository info\n{err.msg}')
+            raise CLIFactoryError(f'Failed to fetch repository info\n{err.msg}') from err
         # Synchronize the RH repository
         try:
             self._satellite.cli.Repository.synchronize(
@@ -758,7 +768,7 @@ def _setup_org_for_a_rh_repo(self, options=None):
                 }
             )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to synchronize repository\n{err.msg}')
+            raise CLIFactoryError(f'Failed to synchronize repository\n{err.msg}') from err
         # Create CV if needed and associate repo with it
         if options.get('content-view-id') is None:
             cv_id = self.make_content_view({'organization-id': org_id})['id']
@@ -769,24 +779,28 @@ def _setup_org_for_a_rh_repo(self, options=None):
                 {'id': cv_id, 'organization-id': org_id, 'repository-id': rhel_repo['id']}
             )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to add repository to content view\n{err.msg}')
+            raise CLIFactoryError(f'Failed to add repository to content view\n{err.msg}') from err
         # Publish a new version of CV
         try:
             self._satellite.cli.ContentView.publish({'id': cv_id})
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to publish new version of content view\n{err.msg}')
+            raise CLIFactoryError(
+                f'Failed to publish new version of content view\n{err.msg}'
+            ) from err
         # Get the version id
         try:
             cvv = self._satellite.cli.ContentView.info({'id': cv_id})['versions'][-1]
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to fetch content view info\n{err.msg}')
+            raise CLIFactoryError(f'Failed to fetch content view info\n{err.msg}') from err
         # Promote version1 to next env
         try:
             self._satellite.cli.ContentView.version_promote(
                 {'id': cvv['id'], 'organization-id': org_id, 'to-lifecycle-environment-id': env_id}
             )
         except CLIReturnCodeError as err:
-            raise CLIFactoryError(f'Failed to promote version to next environment\n{err.msg}')
+            raise CLIFactoryError(
+                f'Failed to promote version to next environment\n{err.msg}'
+            ) from err
         # Create activation key if needed and associate content view with it
         if options.get('activationkey-id') is None:
             activationkey_id = self.make_activation_key(
@@ -805,7 +819,9 @@ def _setup_org_for_a_rh_repo(self, options=None):
                     {'id': activationkey_id, 'organization-id': org_id, 'content-view-id': cv_id}
                 )
             except CLIReturnCodeError as err:
-                raise CLIFactoryError(f'Failed to associate activation-key with CV\n{err.msg}')
+                raise CLIFactoryError(
+                    f'Failed to associate activation-key with CV\n{err.msg}'
+                ) from err
 
         # Add default subscription to activation-key, if SCA mode is disabled
         if self._satellite.is_sca_mode_enabled(org_id) is False:
@@ -876,7 +892,7 @@ def setup_org_for_a_rh_repo(
                         }
                     )
                 except CLIReturnCodeError as err:
-                    raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}')
+                    raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}') from err
 
                 # Add default subscription to activation-key, if SCA mode is disabled
                 if self._satellite.is_sca_mode_enabled(result['organization-id']) is False:
@@ -1086,7 +1102,7 @@ def setup_cdn_and_custom_repos_content(
             try:
                 self._satellite.upload_manifest(org_id, interface='CLI')
             except CLIReturnCodeError as err:
-                raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}')
+                raise CLIFactoryError(f'Failed to upload manifest\n{err.msg}') from err
 
         custom_product, repos_info = self.setup_cdn_and_custom_repositories(
             org_id=org_id, repos=repos, download_policy=download_policy

robottelo/host_helpers/repository_mixins.py

@@ -826,7 +826,7 @@ def setup_virtual_machine(
             patch_os_release_distro = self.os_repo.distro
         rh_repo_ids = []
         if enable_rh_repos:
-            rh_repo_ids = [getattr(repo, 'rh_repository_id') for repo in self.rh_repos]
+            rh_repo_ids = [repo.rh_repository_id for repo in self.rh_repos]
         repo_labels = []
         if enable_custom_repos:
             repo_labels = [