sartography
/
cr-connect-workflow
mirror of https://github.com/sartography/cr-connect-workflow.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into rrt/production

This commit is contained in:
Aaron Louie 2020-05-29 17:27:20 -04:00
parent 5d4cba0b0a 5046d3a669
commit eeae3d1038
2 changed files with 34 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
crc/services/ldap_service.py
View File

@ -1,5 +1,7 @@
import os import os
from ldap3.core.exceptions import LDAPExceptionError
from crc import app from crc import app
from ldap3 import Connection, Server, MOCK_SYNC from ldap3 import Connection, Server, MOCK_SYNC
@ -38,7 +40,9 @@ class LdapService(object):
attributes = ['uid', 'cn', 'sn', 'displayName', 'givenName', 'mail', 'objectClass', 'UvaDisplayDepartment', attributes = ['uid', 'cn', 'sn', 'displayName', 'givenName', 'mail', 'objectClass', 'UvaDisplayDepartment',
'telephoneNumber', 'title', 'uvaPersonIAMAffiliation', 'uvaPersonSponsoredType'] 'telephoneNumber', 'title', 'uvaPersonIAMAffiliation', 'uvaPersonSponsoredType']
uid_search_string = "(&(objectclass=person)(uid=%s))" uid_search_string = "(&(objectclass=person)(uid=%s))"
user_or_last_name_search_string = "(&(objectclass=person)(|(uid=%s*)(sn=%s*)))" user_or_last_name_search = "(&(objectclass=person)(|(uid=%s*)(sn=%s*)))"
cn_single_search = '(&(objectclass=person)(cn=%s*))'
cn_double_search = '(&(objectclass=person)(&(cn=%s*)(cn=*%s*)))'
def __init__(self): def __init__(self):
if app.config['TESTING']: if app.config['TESTING']:
@ -67,18 +71,33 @@ class LdapService(object):
return LdapUserInfo.from_entry(entry) return LdapUserInfo.from_entry(entry)
def search_users(self, query, limit): def search_users(self, query, limit):
if len(query) < 3: return [] if len(query.strip()) < 3:
search_string = LdapService.user_or_last_name_search_string % (query, query) return []
elif query.endswith(' '):
search_string = LdapService.cn_single_search % (query.strip())
elif query.strip().count(',') == 1:
f, l = query.split(",")
search_string = LdapService.cn_double_search % (l.strip(), f.strip())
elif query.strip().count(' ') == 1:
f,l = query.split(" ")
search_string = LdapService.cn_double_search % (f, l)
else:
# Search by user_id or last name
search_string = LdapService.user_or_last_name_search % (query, query)
results = []
print(search_string)
try:
self.conn.search(LdapService.search_base, search_string, attributes=LdapService.attributes) self.conn.search(LdapService.search_base, search_string, attributes=LdapService.attributes)
# Entries are returned as a generator, accessing entries # Entries are returned as a generator, accessing entries
# can make subsequent calls to the ldap service, so limit # can make subsequent calls to the ldap service, so limit
# those here. # those here.
count = 0 count = 0
results = []
for entry in self.conn.entries: for entry in self.conn.entries:
if count > limit: if count > limit:
break break
results.append(LdapUserInfo.from_entry(entry)) results.append(LdapUserInfo.from_entry(entry))
count += 1 count += 1
except LDAPExceptionError as le:
app.logger.info("Failed to execute ldap search. %s", str(le))
return results return results

1
tests/test_lookup_service.py
View File

@ -115,6 +115,7 @@ class TestLookupService(BaseTest):
self.assertEquals("1 Something", results[0].label, "special characters don't flake out") self.assertEquals("1 Something", results[0].label, "special characters don't flake out")
# 1018 10000 Something Industry # 1018 10000 Something Industry
# 1019 1000 Something Industry # 1019 1000 Something Industry
# 1020 1 Something Industry # 1020 1 Something Industry