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

* 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.

* Update type check

---------

Co-authored-by: Gaurav Talreja <[email protected]>

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, str):
             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}'