IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2024-12-22 15:13:50 -06:00
Code Issues Packages Projects Releases Wiki Activity

pylint: Fix used-before-assignment

Browse Source 
> Emitted when a local variable is accessed before its assignment took
place. Assignments in try blocks are assumed not to have occurred when
evaluating associated except/finally blocks. Assignments in except
blocks are assumed not to have occurred when evaluating statements
outside the block, except when the associated try block contains a
return statement.

Fixes: https://pagure.io/freeipa/issue/9278
Signed-off-by: Stanislav Levin <slev@altlinux.org>
Reviewed-By: Stanislav Levin <slev@altlinux.org>
This commit is contained in:
Stanislav Levin 2022-12-16 19:23:32 +03:00 committed by Florence Blanc-Renaud
parent a8dd070992
commit 0e03315299
9 changed files with 39 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ipaclient/install/ipa_client_samba.py
View File

@ -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():

3
ipapython/ipaldap.py
View File

@ -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'

13
ipaserver/install/adtrustinstance.py
View File

@ -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. "

6
ipaserver/install/installutils.py
View File

@ -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)

3
ipaserver/install/krbinstance.py
View File

@ -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)

3
ipatests/test_cmdline/test_ipagetkeytab.py
View File

@ -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)

15
ipatests/test_integration/test_trust.py
View File

@ -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)

25
ipatests/test_xmlrpc/tracker/caacl_plugin.py
View File

@ -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(

13
ipatests/test_xmlrpc/xmlrpc_test.py
View File

@ -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,