Add Ruff rule group B (flake8-bugbear)

This change includes the rule addition and mostly manual changes to
existing code in order to bring it in line with the new rules and avoid
as many ignores as is reasonable.

I added a number of {{from err}} statements to the end of exceptions that currently include the error message. The
error message being included in the raised string can be removed later, but keeping for consistency for now.

In cases where I raised from None, the raised exception is clear enough that the additional context of the parent
exception isn't helpful.

cli/test\_ping.py has a zip that should probably be strict.

pyproject.toml

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

robottelo/cli/base.py

@@ -216,15 +216,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/factory.py

@@ -100,8 +100,8 @@ def create_object(cli_object, options, values):
         diff = set(values.keys()).difference(set(options.keys()))
         if diff:
             logger.debug(
-                "Option(s) {} not supported by CLI factory. Please check for "
-                "a typo or update default options".format(diff)
+                f"Option(s) {diff} not supported by CLI factory. Please check for "
+                "a typo or update default options"
             )
         for key in set(options.keys()).intersection(set(values.keys())):
             options[key] = values[key]
@@ -114,11 +114,11 @@ def create_object(cli_object, options, values):
             '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 result
@@ -513,8 +513,8 @@ def make_product_wait(options=None, wait_for=5):
             product = Product.info(
                 {'name': options.get('name'), 'organization-id': options.get('organization-id')}
             )
-        except CLIReturnCodeError:
-            raise err
+        except CLIReturnCodeError as cli_err:
+            raise err from cli_err
         if not product:
             raise err
     return product
@@ -1631,7 +1631,9 @@ def activationkey_add_subscription_to_repo(options=None):
                     }
                 )
             except CLIReturnCodeError as err:
-                raise CLIFactoryError(f'Failed to add subscription to activation key\n{err.msg}')
+                raise CLIFactoryError(
+                    f'Failed to add subscription to activation key\n{err.msg}'
+                ) from err
 
 
 def setup_org_for_a_custom_repo(options=None):
@@ -1672,7 +1674,7 @@ def setup_org_for_a_custom_repo(options=None):
     try:
         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 = make_content_view({'organization-id': org_id})['id']
@@ -1683,12 +1685,12 @@ def setup_org_for_a_custom_repo(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:
         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 content view info
     cv_info = ContentView.info({'id': cv_id})
     lce_promoted = cv_info['lifecycle-environments']
@@ -1700,7 +1702,7 @@ def setup_org_for_a_custom_repo(options=None):
                 {'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 = make_activation_key(
@@ -1719,7 +1721,7 @@ def setup_org_for_a_custom_repo(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 subscription to activation-key
     activationkey_add_subscription_to_repo(
         {
@@ -1788,7 +1790,7 @@ def _setup_org_for_a_rh_repo(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 = Repository.info(
@@ -1799,7 +1801,7 @@ def _setup_org_for_a_rh_repo(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:
         Repository.synchronize(
@@ -1810,7 +1812,7 @@ def _setup_org_for_a_rh_repo(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 = make_content_view({'organization-id': org_id})['id']
@@ -1821,24 +1823,24 @@ def _setup_org_for_a_rh_repo(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:
         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 = 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:
         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 = make_activation_key(
@@ -1857,7 +1859,7 @@ def _setup_org_for_a_rh_repo(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 subscription to activation-key
     if constants.DEFAULT_SUBSCRIPTION_NAME not in ActivationKey.subscriptions(
         {'id': activationkey_id, 'organization-id': org_id}
@@ -1919,7 +1921,7 @@ def setup_org_for_a_rh_repo(options=None, force_manifest_upload=False, force_use
                     {'file': manifest.filename, 'organization-id': result.get('organization-id')}
                 )
             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
             # attach the default subscription to activation key
             activationkey_add_subscription_to_repo(
                 {
@@ -2125,7 +2127,7 @@ def setup_cdn_and_custom_repos_content(
         try:
             target_sat.upload_manifest(org_id, clone(), 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 = setup_cdn_and_custom_repositories(
         org_id=org_id, repos=repos, download_policy=download_policy

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

@@ -109,8 +109,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'})
@@ -340,14 +340,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(

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']}")