diff --git a/ipaclient/install/ipa_client_samba.py b/ipaclient/install/ipa_client_samba.py index 82d945f2c..81d670c34 100755 --- a/ipaclient/install/ipa_client_samba.py +++ b/ipaclient/install/ipa_client_samba.py @@ -39,8 +39,8 @@ logger.setLevel(logging.DEBUG) @contextmanager def use_api_as_principal(principal, keytab): with ipautil.private_ccache() as ccache_file: + old_principal = getattr(context, "principal", None) try: - old_principal = getattr(context, "principal", None) name = gssapi.Name(principal, gssapi.NameType.kerberos_principal) store = {"ccache": ccache_file, "client_keytab": keytab} gssapi.Credentials(name=name, usage="initiate", store=store) @@ -63,9 +63,7 @@ def use_api_as_principal(principal, keytab): finally: if api.Backend.rpcclient.isconnected(): api.Backend.rpcclient.disconnect() - # pylint: disable=used-before-assignment setattr(context, "principal", old_principal) - # pylint: enable=used-before-assignment def parse_options(): diff --git a/ipapython/ipaldap.py b/ipapython/ipaldap.py index 2f7ef3f94..c20641563 100644 --- a/ipapython/ipaldap.py +++ b/ipapython/ipaldap.py @@ -1082,6 +1082,7 @@ class LDAPClient: def error_handler(self, arg_desc=None): """Context manager that handles LDAPErrors """ + desc = None try: try: yield @@ -1101,7 +1102,7 @@ class LDAPClient: except ldap.TYPE_OR_VALUE_EXISTS: # attribute type or attribute value already exists, usually only # occurs, when two machines try to write at the same time. - raise errors.DuplicateEntry(message=desc) # pylint: disable=E0601 + raise errors.DuplicateEntry(message=desc) except ldap.CONSTRAINT_VIOLATION: # This error gets thrown by the uniqueness plugin _msg = 'Another entry with the same attribute value already exists' diff --git a/ipaserver/install/adtrustinstance.py b/ipaserver/install/adtrustinstance.py index 518a30a07..eb632fdb5 100644 --- a/ipaserver/install/adtrustinstance.py +++ b/ipaserver/install/adtrustinstance.py @@ -388,13 +388,14 @@ class ADTRUSTInstance(service.Service): # Abort if RID bases are too close local_range = ranges_with_no_rid_base[0] + size_value = local_range.single_value.get('ipaIDRangeSize') try: - size = int(local_range.single_value.get('ipaIDRangeSize')) - except ValueError: - # pylint: disable=used-before-assignment - raise RuntimeError('ipaIDRangeSize is set to a non-integer ' - 'value or is not set at all (got {val})' - .format(val=size)) + size = int(size_value) + except (ValueError, TypeError): + raise RuntimeError( + "ipaIDRangeSize is set to a non-integer value or is not set" + f" at all (got {size_value!r})" + ) from None if abs(self.rid_base - self.secondary_rid_base) < size: self.print_msg("Primary and secondary RID base are too close. " diff --git a/ipaserver/install/installutils.py b/ipaserver/install/installutils.py index dd289368e..3a31f8a98 100644 --- a/ipaserver/install/installutils.py +++ b/ipaserver/install/installutils.py @@ -1559,10 +1559,8 @@ def default_ca_subject_dn(subject_base): def validate_mask(): - try: - mask = os.umask(0) - finally: - os.umask(mask) # pylint: disable=used-before-assignment + mask = os.umask(0) + os.umask(mask) mask_str = None if mask & 0b111101101 > 0: mask_str = "{:04o}".format(mask) diff --git a/ipaserver/install/krbinstance.py b/ipaserver/install/krbinstance.py index 4980de15c..002660b0b 100644 --- a/ipaserver/install/krbinstance.py +++ b/ipaserver/install/krbinstance.py @@ -436,8 +436,8 @@ class KrbInstance(service.Service): krbtgt = "krbtgt/" + self.realm + "@" + self.realm certpath = (paths.KDC_CERT, paths.KDC_KEY) + prev_helper = None try: - prev_helper = None # on the first CA-ful master without '--no-pkinit', we issue the # certificate by contacting Dogtag directly ca_instances = find_providing_servers( @@ -482,7 +482,6 @@ class KrbInstance(service.Service): logger.error("Failed to initiate the request: %s", e) return finally: - # pylint: disable-next=used-before-assignment if prev_helper is not None: certmonger.modify_ca_helper(certmonger_ca, prev_helper) diff --git a/ipatests/test_cmdline/test_ipagetkeytab.py b/ipatests/test_cmdline/test_ipagetkeytab.py index 9667699d1..c4678a76d 100755 --- a/ipatests/test_cmdline/test_ipagetkeytab.py +++ b/ipatests/test_cmdline/test_ipagetkeytab.py @@ -44,8 +44,8 @@ from contextlib import contextmanager @contextmanager def use_keytab(principal, keytab): with private_ccache() as ccache_file: + old_principal = getattr(context, 'principal', None) try: - old_principal = getattr(context, 'principal', None) name = gssapi.Name(principal, gssapi.NameType.kerberos_principal) store = {'ccache': ccache_file, 'client_keytab': keytab} @@ -60,7 +60,6 @@ def use_keytab(principal, keytab): 'principal %s in %s: %s' % (principal, keytab, str(e))) finally: - # pylint: disable-next=used-before-assignment setattr(context, 'principal', old_principal) diff --git a/ipatests/test_integration/test_trust.py b/ipatests/test_integration/test_trust.py index 653d86690..c4b3b99ce 100644 --- a/ipatests/test_integration/test_trust.py +++ b/ipatests/test_integration/test_trust.py @@ -657,14 +657,15 @@ class TestTrust(BaseTestTrust): try: # stop sssd_be, needed to simulate a timeout in the extdom plugin. - stop_sssdbe = self.master.run_command('kill -STOP %s' % pid) - client.run_command(test_id) - error = 'ldap_extended_operation result: No such object(32)' - sssd_log2 = client.get_file_contents(log_file)[logsize:] - assert error.encode() not in sssd_log2 - finally: - if stop_sssdbe.returncode == 0: # pylint: disable=E0601 + self.master.run_command('kill -STOP %s' % pid) + try: + client.run_command(test_id) + error = 'ldap_extended_operation result: No such object(32)' + sssd_log2 = client.get_file_contents(log_file)[logsize:] + assert error.encode() not in sssd_log2 + finally: self.master.run_command('kill -CONT %s' % pid) + finally: # reconnect and set back to default extdom plugin conn = self.master.ldap_connect() entry = conn.get_entry(extdom_dn) diff --git a/ipatests/test_xmlrpc/tracker/caacl_plugin.py b/ipatests/test_xmlrpc/tracker/caacl_plugin.py index 5dcc630c3..5f5881154 100644 --- a/ipatests/test_xmlrpc/tracker/caacl_plugin.py +++ b/ipatests/test_xmlrpc/tracker/caacl_plugin.py @@ -49,7 +49,7 @@ class CAACLTracker(Tracker): u'usercategory', u'hostcategory', u'ipacacategory', u'servicecategory', u'ipaenabledflag', u'objectclass', u'ipauniqueid'} - update_keys = create_keys - {u'dn'} + update_keys = retrieve_keys - {"dn"} def __init__(self, name, ipacertprofile_category=None, user_category=None, service_category=None, host_category=None, @@ -158,7 +158,7 @@ class CAACLTracker(Tracker): def make_update_command(self, updates): return self.make_command('caacl_mod', self.name, **updates) - def update(self, updates, expected_updates=None, silent=False): + def update(self, updates, expected_updates=None): """If removing a category, delete it from tracker as well""" # filter out empty categories and track changes @@ -166,11 +166,7 @@ class CAACLTracker(Tracker): for key, value in updates.items(): if key in self.category_keys: if not value: - try: - del self.attrs[key] - except IndexError: - if silent: - pass + del self.attrs[key] else: # if there is a value, prepare the pair for update filtered_updates.update({key: value}) @@ -185,14 +181,13 @@ class CAACLTracker(Tracker): try: result = command() except errors.EmptyModlist: - if silent: - self.attrs.update(filtered_updates) - self.attrs.update(expected_updates) - # pylint: disable=used-before-assignment - self.check_update(result, - extra_keys=set(self.update_keys) | - set(expected_updates.keys())) - # pylint: enable=used-before-assignment + pass + else: + self.attrs.update(filtered_updates) + self.attrs.update(expected_updates) + self.check_update(result, + extra_keys=set(self.update_keys) | + set(expected_updates.keys())) def check_update(self, result, extra_keys=()): assert_deepequal(dict( diff --git a/ipatests/test_xmlrpc/xmlrpc_test.py b/ipatests/test_xmlrpc/xmlrpc_test.py index abf3b8beb..cf11721bf 100644 --- a/ipatests/test_xmlrpc/xmlrpc_test.py +++ b/ipatests/test_xmlrpc/xmlrpc_test.py @@ -136,13 +136,13 @@ def fuzzy_set_optional_oc(s, oc): == set(y.lower() for y in s if y != oc)) +server_available = None try: if not api.Backend.rpcclient.isconnected(): api.Backend.rpcclient.connect() - res = api.Command['user_show'](u'notfound') -except errors.NetworkError: - server_available = False -except IOError: + api.Command["user_show"]("notfound") + server_available = True +except (errors.NetworkError, IOError): server_available = False except errors.NotFound: server_available = True @@ -202,7 +202,7 @@ class XMLRPC_test: """ @pytest.fixture(autouse=True, scope="class") def xmlrpc_setup(self, request): - if not server_available: # pylint: disable=used-before-assignment + if not server_available: pytest.skip('%r: Server not available: %r' % (request.cls.__module__, api.env.xmlrpc_uri)) @@ -345,6 +345,7 @@ class Declarative(XMLRPC_test): def check_exception(self, nice, cmd, args, options, expected): klass = expected.__class__ expected_name = klass.__name__ + got = None try: output = api.Command[cmd](*args, **options) except Exception as e: @@ -353,7 +354,7 @@ class Declarative(XMLRPC_test): raise AssertionError( EXPECTED % (cmd, expected_name, args, options, output) ) - if not isinstance(got, klass): # pylint: disable=used-before-assignment + if not isinstance(got, klass): raise AssertionError( UNEXPECTED % (cmd, expected_name, args, options, expected_name, expected,