[6.13.z] Add Ruff rule group B (flake8-bugbear)

pyproject.toml

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

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

@@ -468,14 +468,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(
@@ -575,8 +575,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:
@@ -607,7 +608,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']
@@ -618,12 +619,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})
         lce_promoted = cv_info['lifecycle-environments']
@@ -639,7 +642,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(
@@ -658,7 +663,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:
@@ -709,15 +716,16 @@ def _setup_org_for_a_rh_repo(self, options=None):
             env_id = self.make_lifecycle_environment({'organization-id': org_id})['id']
         else:
             env_id = options['lifecycle-environment-id']
-        # Clone manifest and upload it
-        with clone() as manifest:
-            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}')
+        # 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:
+                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(
@@ -730,7 +738,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(
@@ -741,7 +749,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(
@@ -752,7 +760,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']
@@ -763,24 +771,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(
@@ -799,7 +811,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:
@@ -865,7 +879,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:
@@ -1075,7 +1089,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

@@ -771,7 +771,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 = [

robottelo/host_helpers/satellite_mixins.py

@@ -1,10 +1,10 @@
 import contextlib
-from functools import cache
 import io
 import os
 import random
 import re
 
+from cachetools import cachedmethod
 import requests
 
 from robottelo.cli.proxy import CapsuleTunnelError
@@ -187,7 +187,7 @@ def publish_content_view(self, org, repo_list):
 
         :returns: A dictionary containing the details of the published content view.
         """
-        repo = repo_list if type(repo_list) is list else [repo_list]
+        repo = repo_list if isinstance(repo_list, list) else [repo_list]
         content_view = self.api.ContentView(organization=org, repository=repo).create()
         content_view.publish()
         content_view = content_view.read()
@@ -234,9 +234,9 @@ def available_capsule_port(self):
         :rtype: int
         """
         port_pool_range = settings.fake_capsules.port_range
-        if type(port_pool_range) is str:
+        if isinstance(port_pool_range, list):
             port_pool_range = tuple(port_pool_range.split('-'))
-        if type(port_pool_range) is tuple and len(port_pool_range) == 2:
+        if isinstance(port_pool_range, tuple) and len(port_pool_range) == 2:
             port_pool = range(int(port_pool_range[0]), int(port_pool_range[1]))
         else:
             raise TypeError(
@@ -262,14 +262,14 @@ def available_capsule_port(self):
         except ValueError:
             raise CapsuleTunnelError(
                 f'Failed parsing the port numbers from stdout: {ss_cmd.stdout.splitlines()[:-1]}'
-            )
+            ) from None
         try:
             # take the list of available ports and return randomly selected one
             return random.choice([port for port in port_pool if port not in used_ports])
         except IndexError:
             raise CapsuleTunnelError(
                 'Failed to create ssh tunnel: No more ports available for mapping'
-            )
+            ) from None
 
     @contextlib.contextmanager
     def default_url_on_new_port(self, oldport, newport):
@@ -307,7 +307,7 @@ def validate_pulp_filepath(
         self,
         org,
         dir_path,
-        file_names=['*.json', '*.tar.gz'],
+        file_names=('*.json', '*.tar.gz'),
     ):
         """Checks the existence of certain files in a pulp dir"""
         extension_query = ' -o '.join([f'-name "{file}"' for file in file_names])
@@ -357,6 +357,6 @@ def api_factory(self):
             self._api_factory = APIFactory(self)
         return self._api_factory
 
-    @cache
+    @cachedmethod
     def ui_factory(self, session):
         return UIFactory(self, session=session)

robottelo/host_helpers/ui_factory.py

@@ -18,12 +18,14 @@ def __init__(self, satellite, session=None):
     def create_fake_host(
         self,
         host,
-        interface_id=gen_string('alpha'),
+        interface_id=None,
         global_parameters=None,
         host_parameters=None,
         extra_values=None,
         new_host_details=False,
     ):
+        if interface_id is None:
+            interface_id = gen_string('alpha')
         if extra_values is None:
             extra_values = {}
         os_name = f'{host.operatingsystem.name} {host.operatingsystem.major}'