IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-01-26 16:16:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

Turn LDAPEntry.single_value into a dictionary-like property.

Browse Source 
This change makes single_value consistent with the raw property.

https://fedorahosted.org/freeipa/ticket/3521
This commit is contained in:
Jan Cholasta 2013-09-10 10:20:24 +00:00 committed by Petr Viktorin
parent 989493979d
commit df5f4ee81d
22 changed files with 152 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
install/tools/ipa-csreplica-manage
View File

@ -100,9 +100,9 @@ def list_replicas(realm, host, replica, dirman_passwd, verbose):
try:
cadn = DN(('cn', 'CA'), DN(ent.dn))
entry = conn.get_entry(cadn)
peers[ent.single_value('cn')] = ['master', '']
peers[ent.single_value['cn']] = ['master', '']
except errors.NotFound:
peers[ent.single_value('cn')] = ['CA not configured', '']
peers[ent.single_value['cn']] = ['CA not configured', '']
except Exception, e:
sys.exit(
@ -124,19 +124,19 @@ def list_replicas(realm, host, replica, dirman_passwd, verbose):
entries = repl.find_replication_agreements()
for entry in entries:
print '%s' % entry.single_value('nsds5replicahost', None)
print '%s' % entry.single_value.get('nsds5replicahost')
if verbose:
print " last init status: %s" % entry.single_value(
'nsds5replicalastinitstatus', None)
print " last init status: %s" % entry.single_value.get(
'nsds5replicalastinitstatus')
print " last init ended: %s" % str(
ipautil.parse_generalized_time(
entry.single_value('nsds5replicalastinitend')))
print " last update status: %s" % entry.single_value(
'nsds5replicalastupdatestatus', None)
entry.single_value['nsds5replicalastinitend']))
print " last update status: %s" % entry.single_value.get(
'nsds5replicalastupdatestatus')
print " last update ended: %s" % str(
ipautil.parse_generalized_time(
entry.single_value('nsds5replicalastupdateend')))
entry.single_value['nsds5replicalastupdateend']))
def del_link(realm, replica1, replica2, dirman_passwd, force=False):
@ -152,7 +152,7 @@ def del_link(realm, replica1, replica2, dirman_passwd, force=False):
# Find the DN of the replication agreement to remove
replica1_dn = None
for e in repl_list1:
if e.single_value('nsDS5ReplicaHost', None) == replica2:
if e.single_value.get('nsDS5ReplicaHost') == replica2:
replica1_dn = e.dn
break
@ -188,7 +188,7 @@ def del_link(realm, replica1, replica2, dirman_passwd, force=False):
# Find the DN of the replication agreement to remove
replica2_dn = None
for e in repl_list:
if e.single_value('nsDS5ReplicaHost', None) == replica1:
if e.single_value.get('nsDS5ReplicaHost') == replica1:
replica2_dn = e.dn
break
@ -263,7 +263,7 @@ def del_master(realm, hostname, options):
replica_names = [options.host]
else:
replica_entries = delrepl.find_ipa_replication_agreements()
replica_names = [rep.single_value('nsds5replicahost', None)
replica_names = [rep.single_value.get('nsds5replicahost')
for rep in replica_entries]
# 5. Remove each agreement
@ -311,7 +311,7 @@ def add_link(realm, replica1, replica2, dirman_passwd, options):
dirman_passwd)
entries = repl1.find_replication_agreements()
for e in entries:
if e.single_value('nsDS5ReplicaHost', None) == replica2:
if e.single_value.get('nsDS5ReplicaHost') == replica2:
sys.exit('This replication agreement already exists.')
repl1.hostnames = [replica1, replica2]

4
install/tools/ipa-managed-entries
View File

@ -113,7 +113,7 @@ def main():
except Exception, e:
root_logger.debug("Search for managed entries failed: %s" % str(e))
sys.exit("Unable to find managed entries at %s" % managed_entry_definitions_dn)
managed_entries = [entry.single_value('cn') for entry in entries]
managed_entries = [entry.single_value['cn'] for entry in entries]
if managed_entries:
print "Available Managed Entry Definitions:"
for managed_entry in managed_entries:
@ -132,7 +132,7 @@ def main():
filter, ['originfilter'])
disable_attr = '(objectclass=disable)'
try:
org_filter = entry.single_value('originfilter', None)
org_filter = entry.single_value.get('originfilter')
disabled = re.search(r'%s' % disable_attr, org_filter)
except KeyError:
sys.exit("%s is not a valid Managed Entry" % def_dn)

62
install/tools/ipa-replica-manage
View File

@ -168,7 +168,7 @@ def list_replicas(realm, host, replica, dirman_passwd, verbose, nolookup=False):
return
else:
for ent in entries:
peers[ent.single_value('cn')] = ['master', '']
peers[ent.single_value['cn']] = ['master', '']
dn = DN(('cn', 'replicas'), ('cn', 'ipa'), ('cn', 'etc'), ipautil.realm_to_suffix(realm))
try:
@ -177,8 +177,8 @@ def list_replicas(realm, host, replica, dirman_passwd, verbose, nolookup=False):
pass
else:
for ent in entries:
config_string = ent.single_value('ipaConfigString')
peers[ent.single_value('cn')] = config_string.split(':')
config_string = ent.single_value['ipaConfigString']
peers[ent.single_value['cn']] = config_string.split(':')
if not replica:
for k, p in peers.iteritems():
@ -215,18 +215,18 @@ def list_replicas(realm, host, replica, dirman_passwd, verbose, nolookup=False):
return
for entry in entries:
print '%s: %s' % (entry.single_value('nsds5replicahost', None), ent_type)
print '%s: %s' % (entry.single_value.get('nsds5replicahost'), ent_type)
if verbose:
print " last init status: %s" % entry.single_value(
'nsds5replicalastinitstatus', None)
print " last init status: %s" % entry.single_value.get(
'nsds5replicalastinitstatus')
print " last init ended: %s" % str(ipautil.parse_generalized_time(
entry.single_value('nsds5replicalastinitend')))
print " last update status: %s" % entry.single_value(
'nsds5replicalastupdatestatus', None)
entry.single_value['nsds5replicalastinitend']))
print " last update status: %s" % entry.single_value.get(
'nsds5replicalastupdatestatus')
print " last update ended: %s" % str(
ipautil.parse_generalized_time(
entry.single_value('nsds5replicalastupdateend')))
entry.single_value['nsds5replicalastupdateend']))
def del_link(realm, replica1, replica2, dirman_passwd, force=False):
"""
@ -480,12 +480,12 @@ def list_clean_ruv(realm, host, dirman_passwd, verbose, nolookup=False):
else:
print "CLEANALLRUV tasks"
for entry in entries:
name = entry.single_value('cn').replace('clean ', '')
status = entry.single_value('nsTaskStatus', None)
name = entry.single_value['cn'].replace('clean ', '')
status = entry.single_value.get('nsTaskStatus')
print "RID %s: %s" % (name, status)
if verbose:
print str(dn)
print entry.single_value('nstasklog', None)
print entry.single_value.get('nstasklog')
print
@ -497,12 +497,12 @@ def list_clean_ruv(realm, host, dirman_passwd, verbose, nolookup=False):
else:
print "Abort CLEANALLRUV tasks"
for entry in entries:
name = entry.single_value('cn').replace('abort ', '')
status = entry.single_value('nsTaskStatus', None)
name = entry.single_value['cn'].replace('abort ', '')
status = entry.single_value.get('nsTaskStatus')
print "RID %s: %s" % (name, status)
if verbose:
print str(dn)
print entry.single_value('nstasklog', None)
print entry.single_value.get('nstasklog')
def check_last_link(delrepl, realm, dirman_passwd, force):
"""
@ -527,7 +527,7 @@ def check_last_link(delrepl, realm, dirman_passwd, force):
"""
replica_entries = delrepl.find_ipa_replication_agreements()
replica_names = [rep.single_value('nsds5replicahost', None)
replica_names = [rep.single_value.get('nsds5replicahost')
for rep in replica_entries]
orphaned = []
@ -543,7 +543,7 @@ def check_last_link(delrepl, realm, dirman_passwd, force):
continue
entries = repl.find_ipa_replication_agreements()
names = [rep.single_value('nsds5replicahost', None)
names = [rep.single_value.get('nsds5replicahost')
for rep in entries]
if len(names) == 1 and names[0] == delrepl.hostname:
@ -614,7 +614,7 @@ def del_master(realm, hostname, options):
dn, thisrepl.conn.SCOPE_ONELEVEL)
replica_names = []
for entry in entries:
replica_names.append(entry.single_value('cn'))
replica_names.append(entry.single_value['cn'])
# The host we're removing gets included in this list, remove it.
# Otherwise we try to delete an agreement from the host to itself.
try:
@ -624,7 +624,7 @@ def del_master(realm, hostname, options):
else:
# Get list of agreements.
replica_entries = delrepl.find_ipa_replication_agreements()
replica_names = [rep.single_value('nsds5replicahost', None)
replica_names = [rep.single_value.get('nsds5replicahost')
for rep in replica_entries]
else:
# WINSYNC replica, delete agreement from current host
@ -666,14 +666,14 @@ def del_master(realm, hostname, options):
this_services = []
other_services = []
for master_cn in [m.single_value('cn') for m in masters]:
for master_cn in [m.single_value['cn'] for m in masters]:
master_dn = DN(('cn', master_cn), ('cn', 'masters'), ('cn', 'ipa'), ('cn', 'etc'), ipautil.realm_to_suffix(realm))
try:
services = delrepl.conn.get_entries(master_dn,
delrepl.conn.SCOPE_ONELEVEL)
except errors.NotFound:
continue
services_cns = [s.single_value('cn') for s in services]
services_cns = [s.single_value['cn'] for s in services]
if master_cn == hostname:
this_services = services_cns
@ -766,7 +766,7 @@ def add_link(realm, replica1, replica2, dirman_passwd, options):
if repl.get_agreement_type(replica2) == replication.WINSYNC:
agreement = repl.get_replication_agreement(replica2)
sys.exit("winsync agreement already exists on subtree %s" %
agreement.single_value('nsds7WindowsReplicaSubtree', None))
agreement.single_value.get('nsds7WindowsReplicaSubtree'))
else:
sys.exit("A replication agreement to %s already exists" % replica2)
except errors.NotFound:
@ -864,7 +864,7 @@ def re_initialize(realm, thishost, fromhost, dirman_passwd, nolookup=False):
# If the agreement doesn't have nsDS5ReplicatedAttributeListTotal it means
# we did not replicate memberOf, do so now.
if not agreement.single_value('nsDS5ReplicatedAttributeListTotal', None):
if not agreement.single_value.get('nsDS5ReplicatedAttributeListTotal'):
ds = dsinstance.DsInstance(realm_name = realm, dm_password = dirman_passwd)
ds.ldapi = os.getegid() == 0
ds.init_memberof()
@ -918,7 +918,7 @@ def show_DNA_ranges(hostname, master, realm, dirman_passwd, nextrange=False,
return False
for ent in entries:
remote = ent.single_value('cn')
remote = ent.single_value['cn']
if master is not None and remote != master:
continue
try:
@ -974,7 +974,7 @@ def store_DNA_range(repl, range_start, range_max, deleted_master, realm,
return False
for ent in entries:
candidate = ent.single_value('cn')
candidate = ent.single_value['cn']
if candidate == deleted_master:
continue
try:
@ -1066,7 +1066,7 @@ def set_DNA_range(hostname, range, realm, dirman_passwd, next_range=False,
sys.exit("Failed to read master data from '%s': %s" % (repl.conn.host, str(e)))
else:
for ent in entries:
master = ent.single_value('cn')
master = ent.single_value['cn']
if master == hostname and not next_range:
continue
try:
@ -1100,8 +1100,8 @@ def set_DNA_range(hostname, range, realm, dirman_passwd, next_range=False,
sys.exit('Unable to load IPA ranges: %s' % e.message)
for ent in entries:
entry_start = int(ent.single_value('ipabaseid'))
entry_max = entry_start + int(ent.single_value('ipaidrangesize'))
entry_start = int(ent.single_value['ipabaseid'])
entry_max = entry_start + int(ent.single_value['ipaidrangesize'])
if dna_next >= entry_start and dna_max <= entry_max:
break
else:
@ -1115,8 +1115,8 @@ def set_DNA_range(hostname, range, realm, dirman_passwd, next_range=False,
entries = []
for ent in entries:
entry_start = int(ent.single_value('ipabaseid'))
entry_max = entry_start + int(ent.single_value('ipaidrangesize'))
entry_start = int(ent.single_value['ipabaseid'])
entry_max = entry_start + int(ent.single_value['ipaidrangesize'])
if range_intersection(dna_next, dna_max, entry_start, entry_max):
sys.exit("New range overlaps with a Trust range. See ipa help idrange command")

4
install/tools/ipactl
View File

@ -138,7 +138,7 @@ def get_config(dirsrv):
masters_list.append("No master found because of error: %s" % str(e))
else:
for dn, master_entry in entries:
masters_list.append(master_entry.single_value('cn'))
masters_list.append(master_entry.single_value['cn'])
masters = "\n".join(masters_list)
@ -151,7 +151,7 @@ def get_config(dirsrv):
svc_list = []
for entry in res:
name = entry.single_value('cn')
name = entry.single_value['cn']
for p in entry['ipaConfigString']:
if p.startswith('startOrder '):
order = p.split()[1]

2
ipa-client/ipaclient/ipadiscovery.py
View File

@ -376,7 +376,7 @@ class IPADiscovery(object):
for lres in lret:
root_logger.debug("Found: %s", lres.dn)
lrealms.append(lres.single_value('cn'))
lrealms.append(lres.single_value['cn'])
if trealm:
for r in lrealms:

6
ipalib/plugins/host.py
View File

@ -585,7 +585,7 @@ class host_del(LDAPDelete):
(dn, entry_attrs) = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
cert = entry_attrs.single_value('usercertificate', None)
cert = entry_attrs.single_value.get('usercertificate')
if cert:
cert = x509.normalize_certificate(cert)
try:
@ -667,7 +667,7 @@ class host_mod(LDAPUpdate):
if self.api.env.enable_ra:
x509.verify_cert_subject(ldap, keys[-1], cert)
(dn, entry_attrs_old) = ldap.get_entry(dn, ['usercertificate'])
oldcert = entry_attrs_old.single_value('usercertificate', None)
oldcert = entry_attrs_old.single_value.get('usercertificate')
if oldcert:
oldcert = x509.normalize_certificate(oldcert)
try:
@ -945,7 +945,7 @@ class host_disable(LDAPQuery):
(dn, entry_attrs) = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
cert = entry_attrs.single_value('usercertificate', None)
cert = entry_attrs.single_value.get('usercertificate')
if cert:
if self.api.env.enable_ra:
cert = x509.normalize_certificate(cert)

6
ipalib/plugins/service.py
View File

@ -259,7 +259,7 @@ def update_krbticketflags(ldap, entry_attrs, attrs_list, options, existing):
old_entry_attrs = entry_attrs
try:
ticket_flags = old_entry_attrs.single_value('krbticketflags')
ticket_flags = old_entry_attrs.single_value['krbticketflags']
ticket_flags = int(ticket_flags)
except (KeyError, ValueError):
ticket_flags = _ticket_flags_default
@ -275,8 +275,8 @@ def set_kerberos_attrs(entry_attrs, options):
return
try:
ticket_flags = entry_attrs.single_value('krbticketflags',
_ticket_flags_default)
ticket_flags = entry_attrs.single_value.get('krbticketflags',
_ticket_flags_default)
ticket_flags = int(ticket_flags)
except ValueError:
return

52
ipapython/ipaldap.py
View File

@ -620,7 +620,7 @@ class IPASimpleLDAPObject(object):
# r[1] == r.data
class LDAPEntry(collections.MutableMapping):
__slots__ = ('_conn', '_dn', '_names', '_nice', '_raw', '_sync',
'_not_list', '_orig', '_raw_view')
'_not_list', '_orig', '_raw_view', '_single_value_view')
def __init__(self, _conn, _dn=None, _obj=None, **kwargs):
"""
@ -638,6 +638,8 @@ class LDAPEntry(collections.MutableMapping):
Keyword arguments can be used to override values of specific attributes.
"""
super(LDAPEntry, self).__init__()
if isinstance(_conn, LDAPEntry):
assert _dn is None
_dn = _conn
@ -662,6 +664,7 @@ class LDAPEntry(collections.MutableMapping):
self._not_list = set()
self._orig = self
self._raw_view = None
self._single_value_view = None
if isinstance(_obj, LDAPEntry):
#pylint: disable=E1103
@ -699,6 +702,12 @@ class LDAPEntry(collections.MutableMapping):
self._raw_view = RawLDAPEntryView(self)
return self._raw_view
@property
def single_value(self):
if self._single_value_view is None:
self._single_value_view = SingleValueLDAPEntryView(self)
return self._single_value_view
@property
def data(self):
# FIXME: for backwards compatibility only
@ -911,27 +920,6 @@ class LDAPEntry(collections.MutableMapping):
return self._get_nice(name)
def single_value(self, name, default=_missing):
"""Return a single attribute value
Checks that the attribute really has one and only one value
If the entry is missing and default is given, return the default.
If the entry is missing and default is not given, raise KeyError.
"""
try:
values = self[name]
except KeyError:
if default is _missing:
raise
return default
if not isinstance(values, list): # TODO: remove when we enforce lists
return values
if len(values) != 1:
raise ValueError(
'%s has %s values, one expected' % (name, len(values)))
return values[0]
def __delitem__(self, name):
name = self._get_attr_name(name)
@ -1047,6 +1035,26 @@ class RawLDAPEntryView(LDAPEntryView):
def __setitem__(self, name, value):
self._entry._set_raw(name, value)
class SingleValueLDAPEntryView(LDAPEntryView):
def __getitem__(self, name):
value = self._entry[name]
if not isinstance(value, list):
# FIXME: remove when we enforce lists
return value
elif not value:
return None
elif len(value) == 1:
return value[0]
else:
raise ValueError(
'%s has %s values, one expected' % (name, len(value)))
def __setitem__(self, name, value):
if value is None:
self._entry[name] = None
else:
self._entry[name] = [value]
class LDAPClient(object):
"""LDAP backend class

4
ipapython/ipautil.py
View File

@ -895,7 +895,7 @@ def get_ipa_basedn(conn):
contexts = entry['namingcontexts']
if 'defaultnamingcontext' in entry:
# If there is a defaultNamingContext examine that one first
default = entry.single_value('defaultnamingcontext')
default = entry.single_value['defaultnamingcontext']
if default in contexts:
contexts.remove(default)
contexts.insert(0, default)
@ -908,7 +908,7 @@ def get_ipa_basedn(conn):
root_logger.debug("LDAP server did not return info attribute to "
"check for IPA version")
continue
info = entry.single_value('info').lower()
info = entry.single_value['info'].lower()
if info != IPA_BASEDN_INFO:
root_logger.debug("Detected IPA server version (%s) did not match the client (%s)" \
% (info, IPA_BASEDN_INFO))

18
ipaserver/install/adtrustinstance.py
View File

@ -201,7 +201,7 @@ class ADTRUSTInstance(service.Service):
self.print_msg("Samba domain object not found")
return
dom_sid = dom_entry.single_value(self.ATTR_SID, None)
dom_sid = dom_entry.single_value.get(self.ATTR_SID)
if not dom_sid:
self.print_msg("Samba domain object does not have a SID")
return
@ -218,7 +218,7 @@ class ADTRUSTInstance(service.Service):
self.print_msg("IPA admin group object not found")
return
if admin_entry.single_value(self.ATTR_SID, None):
if admin_entry.single_value.get(self.ATTR_SID):
self.print_msg("Admin SID already set, nothing to do")
else:
try:
@ -228,7 +228,7 @@ class ADTRUSTInstance(service.Service):
except:
self.print_msg("Failed to modify IPA admin object")
if admin_group_entry.single_value(self.ATTR_SID, None):
if admin_group_entry.single_value.get(self.ATTR_SID):
self.print_msg("Admin group SID already set, nothing to do")
else:
try:
@ -259,7 +259,7 @@ class ADTRUSTInstance(service.Service):
self.print_msg("Samba domain object not found")
return
if dom_entry.single_value(self.ATTR_FALLBACK_GROUP, None):
if dom_entry.single_value.get(self.ATTR_FALLBACK_GROUP):
self.print_msg("Fallback group already set, nothing to do")
return
@ -303,8 +303,8 @@ class ADTRUSTInstance(service.Service):
# Filter out ranges where RID base is already set
no_rid_base_set = lambda r: not any((
r.single_value('ipaBaseRID', None),
r.single_value('ipaSecondaryBaseRID', None)))
r.single_value.get('ipaBaseRID'),
r.single_value.get('ipaSecondaryBaseRID')))
ranges_with_no_rid_base = filter(no_rid_base_set, ranges)
@ -321,7 +321,7 @@ class ADTRUSTInstance(service.Service):
# Abort if RID bases are too close
local_range = ranges_with_no_rid_base[0]
size = local_range.single_value('ipaIDRangeSize', None)
size = local_range.single_value.get('ipaIDRangeSize')
if abs(self.rid_base - self.secondary_rid_base) > size:
self.print_msg("Primary and secondary RID base are too close. "
@ -699,7 +699,7 @@ class ADTRUSTInstance(service.Service):
if len(res) > 1:
# there are other CIFS services defined, we are not alone
for entry in res:
managedBy = entry.single_value('managedBy', None)
managedBy = entry.single_value.get('managedBy')
if managedBy:
fqdn = DN(managedBy)['fqdn']
if fqdn != unicode(self.fqdn):
@ -810,7 +810,7 @@ class ADTRUSTInstance(service.Service):
raise ValueError("No local ID range and no admins group found.\n" \
"Add local ID range manually and try again!")
base_id = int(entry.single_value('gidNumber'))
base_id = int(entry.single_value['gidNumber'])
id_range_size = 200000
id_filter = "(&" \

2
ipaserver/install/dsinstance.py
View File

@ -894,7 +894,7 @@ class DsInstance(service.Service):
dn = DN(('cn', 'default'), ('ou', 'profile'), self.suffix)
try:
entry = self.admin_conn.get_entry(dn)
srvlist = entry.single_value('defaultServerList', '')
srvlist = entry.single_value.get('defaultServerList', '')
srvlist = srvlist.split()
if not self.fqdn in srvlist:
srvlist.append(self.fqdn)

2
ipaserver/install/ipa_backup.py
View File

@ -504,7 +504,7 @@ class Backup(admintool.AdminTool):
self.log.error("Failed to read services from '%s': %s" %
(conn.host, e))
else:
services_cns = [s.single_value('cn') for s in services]
services_cns = [s.single_value['cn'] for s in services]
config.set('ipa', 'services', ','.join(services_cns))
with open(self.header, 'w') as fd:

8
ipaserver/install/ipa_restore.py
View File

@ -349,7 +349,7 @@ class Restore(admintool.AdminTool):
raise admintool.ScriptError(
"Failed to read master data: %s" % e)
else:
masters = [ent.single_value('cn') for ent in entries]
masters = [ent.single_value['cn'] for ent in entries]
for master in masters:
if master == api.env.host:
@ -368,10 +368,10 @@ class Restore(admintool.AdminTool):
except errors.NotFound:
continue
services_cns = [s.single_value('cn') for s in services]
services_cns = [s.single_value['cn'] for s in services]
host_entries = repl.find_ipa_replication_agreements()
hosts = [rep.single_value('nsds5replicahost', None)
hosts = [rep.single_value.get('nsds5replicahost')
for rep in host_entries]
for host in hosts:
@ -386,7 +386,7 @@ class Restore(admintool.AdminTool):
self.log.critical("Unable to disable agreement on %s: %s" % (master, e))
host_entries = repl.find_ipa_replication_agreements()
hosts = [rep.single_value('nsds5replicahost', None)
hosts = [rep.single_value.get('nsds5replicahost')
for rep in host_entries]
for host in hosts:
self.log.info('Disabling CA replication agreement on %s to %s' % (master, host))

2
ipaserver/install/ipa_server_certinstall.py
View File

@ -117,7 +117,7 @@ class ServerCertInstall(admintool.AdminTool):
entry = conn.get_entry(DN(('cn', 'RSA'), ('cn', 'encryption'),
('cn', 'config')),
['nssslpersonalityssl'])
old_cert = entry.single_value('nssslpersonalityssl')
old_cert = entry.single_value['nssslpersonalityssl']
server_cert = self.import_cert(dirname, self.options.pin,
old_cert, 'ldap/%s' % api.env.host,

2
ipaserver/install/krbinstance.py
View File

@ -365,7 +365,7 @@ class KrbInstance(service.Service):
raise e
krbMKey = pyasn1.codec.ber.decoder.decode(
entry.single_value('krbmkey', None))
entry.single_value.get('krbmkey'))
keytype = int(krbMKey[0][1][0])
keydata = str(krbMKey[0][1][1])

4
ipaserver/install/ldapupdate.py
View File

@ -482,7 +482,7 @@ class LDAPUpdate:
self.error("Task lookup failure %s", e)
return
status = entry.single_value('nstaskstatus', None)
status = entry.single_value.get('nstaskstatus')
if status is None:
# task doesn't have a status yet
time.sleep(1)
@ -842,7 +842,7 @@ class LDAPUpdate:
if entry.dn.endswith(DN(('cn', 'index'), ('cn', 'userRoot'),
('cn', 'ldbm database'), ('cn', 'plugins'),
('cn', 'config'))) and (added or updated):
taskid = self.create_index_task(entry.single_value('cn'))
taskid = self.create_index_task(entry.single_value['cn'])
self.monitor_index_task(taskid)
return

4
ipaserver/install/plugins/fix_replica_agreements.py
View File

@ -52,7 +52,7 @@ class update_replica_attribute_lists(PreUpdate):
self.log.debug("Found %d agreement(s)", len(ipa_replicas))
for replica in ipa_replicas:
self.log.debug(replica.single_value('description', None))
self.log.debug(replica.single_value.get('description'))
self._update_attr(repl, replica,
'nsDS5ReplicatedAttributeList',
@ -81,7 +81,7 @@ class update_replica_attribute_lists(PreUpdate):
:param values: List of values the attribute should hold
:param template: Template to use when adding attribute
"""
attrlist = replica.single_value(attribute, None)
attrlist = replica.single_value.get(attribute)
if attrlist is None:
self.log.debug("Adding %s", attribute)

50
ipaserver/install/replication.py
View File

@ -104,7 +104,7 @@ def enable_replication_version_checking(hostname, realm, dirman_passwd):
entry = conn.get_entry(DN(('cn', 'IPA Version Replication'),
('cn', 'plugins'),
('cn', 'config')))
if entry.single_value('nsslapd-pluginenabled', None) == 'off':
if entry.single_value.get('nsslapd-pluginenabled') == 'off':
conn.modify_s(entry.dn, [(ldap.MOD_REPLACE, 'nsslapd-pluginenabled', 'on')])
conn.unbind()
serverid = "-".join(realm.split("."))
@ -126,8 +126,8 @@ def wait_for_task(conn, dn):
'nsTaskTotalItems']
while True:
entry = conn.get_entry(dn, attrlist)
if entry.single_value('nsTaskExitCode', None):
exit_code = int(entry.single_value('nsTaskExitCode'))
if entry.single_value.get('nsTaskExitCode'):
exit_code = int(entry.single_value['nsTaskExitCode'])
break
time.sleep(1)
return exit_code
@ -224,8 +224,8 @@ class ReplicationManager(object):
except errors.NotFound:
pass
else:
if replica.single_value('nsDS5ReplicaId', None):
return int(replica.single_value('nsDS5ReplicaId'))
if replica.single_value.get('nsDS5ReplicaId'):
return int(replica.single_value['nsDS5ReplicaId'])
# Ok, either the entry doesn't exist or the attribute isn't set
# so get it from the other master
@ -237,12 +237,12 @@ class ReplicationManager(object):
root_logger.debug("Unable to retrieve nsDS5ReplicaId from remote server")
raise
else:
if replica.single_value('nsDS5ReplicaId', None) is None:
if replica.single_value.get('nsDS5ReplicaId') is None:
root_logger.debug("Unable to retrieve nsDS5ReplicaId from remote server")
raise RuntimeError("Unable to retrieve nsDS5ReplicaId from remote server")
# Now update the value on the master
retval = int(replica.single_value('nsDS5ReplicaId'))
retval = int(replica.single_value['nsDS5ReplicaId'])
mod = [(ldap.MOD_REPLACE, 'nsDS5ReplicaId', str(retval + 1))]
try:
@ -421,7 +421,7 @@ class ReplicationManager(object):
('cn', 'config'), ('cn', 'ldbm database'),
('cn', 'plugins'), ('cn', 'config')),
['nsslapd-directory'])
dbdir = os.path.dirname(ent.single_value('nsslapd-directory', None))
dbdir = os.path.dirname(ent.single_value.get('nsslapd-directory'))
entry = conn.make_entry(
DN(('cn', 'changelog5'), ('cn', 'config')),
@ -501,7 +501,7 @@ class ReplicationManager(object):
DN(('cn', 'Multimaster Replication Plugin'), ('cn', 'plugins'),
('cn', 'config')),
['nsslapd-pluginPath'])
path = plgent.single_value('nsslapd-pluginPath', None)
path = plgent.single_value.get('nsslapd-pluginPath')
mod = [(ldap.MOD_REPLACE, 'nsslapd-state', 'backend'),
(ldap.MOD_ADD, 'nsslapd-backend', bename),
@ -802,10 +802,9 @@ class ReplicationManager(object):
print "Error reading status from agreement", agmtdn
hasError = 1
else:
refresh = entry.single_value('nsds5BeginReplicaRefresh', None)
inprogress = entry.single_value('nsds5replicaUpdateInProgress',
None)
status = entry.single_value('nsds5ReplicaLastInitStatus', None)
refresh = entry.single_value.get('nsds5BeginReplicaRefresh')
inprogress = entry.single_value.get('nsds5replicaUpdateInProgress')
status = entry.single_value.get('nsds5ReplicaLastInitStatus')
if not refresh: # done - check status
if not status:
print "No status yet"
@ -843,15 +842,14 @@ class ReplicationManager(object):
print "Error reading status from agreement", agmtdn
hasError = 1
else:
inprogress = entry.single_value('nsds5replicaUpdateInProgress',
None)
status = entry.single_value('nsds5ReplicaLastUpdateStatus', None)
inprogress = entry.single_value.get('nsds5replicaUpdateInProgress')
status = entry.single_value.get('nsds5ReplicaLastUpdateStatus')
try:
start = int(entry.single_value('nsds5ReplicaLastUpdateStart'))
start = int(entry.single_value['nsds5ReplicaLastUpdateStart'])
except (ValueError, TypeError, KeyError):
start = 0
try:
end = int(entry.single_value('nsds5ReplicaLastUpdateEnd'))
end = int(entry.single_value['nsds5ReplicaLastUpdateEnd'])
except (ValueError, TypeError, KeyError):
end = 0
# incremental update is done if inprogress is false and end >= start
@ -1095,7 +1093,7 @@ class ReplicationManager(object):
root_logger.error("Using the first one only (%s)" % entries[0].dn)
dn = entries[0].dn
schedule = entries[0].single_value('nsds5replicaupdateschedule', None)
schedule = entries[0].single_value.get('nsds5replicaupdateschedule')
# On the remote chance of a match. We force a synch to happen right
# now by setting the schedule to something and quickly removing it.
@ -1215,7 +1213,7 @@ class ReplicationManager(object):
try:
dn = DN(('cn', 'default'), ('ou', 'profile'), self.suffix)
ret = self.conn.get_entry(dn)
srvlist = ret.single_value('defaultServerList', '')
srvlist = ret.single_value.get('defaultServerList', '')
srvlist = srvlist[0].split()
if replica in srvlist:
srvlist.remove(replica)
@ -1321,15 +1319,15 @@ class ReplicationManager(object):
"""
entry = self.conn.get_entry(DNA_DN)
nextvalue = int(entry.single_value("dnaNextValue", 0))
maxvalue = int(entry.single_value("dnaMaxValue", 0))
nextvalue = int(entry.single_value.get("dnaNextValue", 0))
maxvalue = int(entry.single_value.get("dnaMaxValue", 0))
sharedcfgdn = entry.single_value("dnaSharedCfgDN", None)
sharedcfgdn = entry.single_value.get("dnaSharedCfgDN")
if sharedcfgdn is not None:
sharedcfgdn = DN(sharedcfgdn)
shared_entry = self.conn.get_entry(sharedcfgdn)
remaining = int(shared_entry.single_value("dnaRemainingValues", 0))
remaining = int(shared_entry.single_value.get("dnaRemainingValues", 0))
else:
remaining = 0
@ -1352,7 +1350,7 @@ class ReplicationManager(object):
"""
entry = self.conn.get_entry(DNA_DN)
range = entry.single_value("dnaNextRange", None)
range = entry.single_value.get("dnaNextRange")
if range is None:
return (None, None)
@ -1378,7 +1376,7 @@ class ReplicationManager(object):
"""
entry = self.conn.get_entry(DNA_DN)
range = entry.single_value("dnaNextRange", None)
range = entry.single_value.get("dnaNextRange")
if range is not None and next_start != 0 and next_max != 0:
return False

26
ipatests/test_install/test_updates.py
View File

@ -112,7 +112,7 @@ class test_update(unittest.TestCase):
for item in ('top', 'nsContainer'):
self.assertTrue(item in objectclasses)
self.assertEqual(entry.single_value('cn'), 'test')
self.assertEqual(entry.single_value['cn'], 'test')
entries = self.ld.get_entries(
self.user_dn, self.ld.SCOPE_BASE, 'objectclass=*', ['*'])
@ -123,10 +123,10 @@ class test_update(unittest.TestCase):
for item in ('top', 'person', 'posixaccount', 'krbprincipalaux', 'inetuser'):
self.assertTrue(item in objectclasses)
self.assertEqual(entry.single_value('loginshell'), '/bin/bash')
self.assertEqual(entry.single_value('sn'), 'User')
self.assertEqual(entry.single_value('uid'), 'tuser')
self.assertEqual(entry.single_value('cn'), 'Test User')
self.assertEqual(entry.single_value['loginshell'], '/bin/bash')
self.assertEqual(entry.single_value['sn'], 'User')
self.assertEqual(entry.single_value['uid'], 'tuser')
self.assertEqual(entry.single_value['cn'], 'Test User')
def test_2_update(self):
@ -140,7 +140,7 @@ class test_update(unittest.TestCase):
self.user_dn, self.ld.SCOPE_BASE, 'objectclass=*', ['*'])
self.assertEqual(len(entries), 1)
entry = entries[0]
self.assertEqual(entry.single_value('gecos'), 'Test User')
self.assertEqual(entry.single_value['gecos'], 'Test User')
def test_3_update(self):
"""
@ -153,7 +153,7 @@ class test_update(unittest.TestCase):
self.user_dn, self.ld.SCOPE_BASE, 'objectclass=*', ['*'])
self.assertEqual(len(entries), 1)
entry = entries[0]
self.assertEqual(entry.single_value('gecos'), 'Test User New')
self.assertEqual(entry.single_value['gecos'], 'Test User New')
def test_4_update(self):
"""
@ -166,7 +166,7 @@ class test_update(unittest.TestCase):
self.user_dn, self.ld.SCOPE_BASE, 'objectclass=*', ['*'])
self.assertEqual(len(entries), 1)
entry = entries[0]
self.assertEqual(entry.single_value('gecos'), 'Test User New2')
self.assertEqual(entry.single_value['gecos'], 'Test User New2')
def test_5_update(self):
"""
@ -296,7 +296,7 @@ class test_update(unittest.TestCase):
for item in ('top', 'nsContainer'):
self.assertTrue(item in objectclasses)
self.assertEqual(entry.single_value('cn'), 'test')
self.assertEqual(entry.single_value['cn'], 'test')
entries = self.ld.get_entries(
self.user_dn, self.ld.SCOPE_BASE, 'objectclass=*', ['*'])
@ -307,10 +307,10 @@ class test_update(unittest.TestCase):
for item in ('top', 'person', 'posixaccount', 'krbprincipalaux', 'inetuser'):
self.assertTrue(item in objectclasses)
self.assertEqual(entry.single_value('loginshell'), '/bin/bash')
self.assertEqual(entry.single_value('sn'), 'User')
self.assertEqual(entry.single_value('uid'), 'tuser')
self.assertEqual(entry.single_value('cn'), 'Test User')
self.assertEqual(entry.single_value['loginshell'], '/bin/bash')
self.assertEqual(entry.single_value['sn'], 'User')
self.assertEqual(entry.single_value['uid'], 'tuser')
self.assertEqual(entry.single_value['cn'], 'Test User')
# Now delete

4
ipatests/test_integration/tasks.py
View File

@ -620,11 +620,11 @@ def wait_for_replication(ldap, timeout=30):
filter='(objectclass=nsds5replicationagreement)',
attrs_list=[status_attr, progress_attr])
log.debug('Replication agreements: \n%s', _entries_to_ldif(entries))
if any(not e.single_value(status_attr).startswith('0 ')
if any(not e.single_value[status_attr].startswith('0 ')
for e in entries):
log.error('Replication error')
continue
if any(e.single_value(progress_attr) == 'TRUE' for e in entries):
if any(e.single_value[progress_attr] == 'TRUE' for e in entries):
log.debug('Replication in progress (waited %s/%ss)',
i, timeout)
else:

2
ipatests/test_integration/test_caless.py
View File

@ -313,7 +313,7 @@ class CALessBase(IntegrationTest):
ldap = host.ldap_connect()
entry = ldap.get_entry(DN(('cn', 'CACert'), ('cn', 'ipa'),
('cn', 'etc'), host.domain.basedn))
cert_from_ldap = entry.single_value('cACertificate')
cert_from_ldap = entry.single_value['cACertificate']
self.log.debug('CA cert from LDAP on %s:\n%r',
host, cert_from_ldap)
assert cert_from_ldap == expected_binary_cacrt

8
ipatests/test_ipaserver/test_ldap.py
View File

@ -253,10 +253,10 @@ class test_LDAPEntry(object):
def test_single_value(self):
e = self.entry
assert e.single_value('cn') == self.cn1[0]
assert e.single_value('commonname') == self.cn1[0]
assert e.single_value('COMMONNAME', 'default') == self.cn1[0]
assert e.single_value('bad key', 'default') == 'default'
assert e.single_value['cn'] == self.cn1[0]
assert e.single_value['commonname'] == self.cn1[0]
assert e.single_value.get('COMMONNAME', 'default') == self.cn1[0]
assert e.single_value.get('bad key', 'default') == 'default'
def test_sync(self):
e = self.entry