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 'rrt/testing' into rrt/staging

This commit is contained in:
Aaron Louie 2020-05-29 21:42:11 -04:00
parent b8203402b7 ab7fd60108
commit 99278f1beb
2 changed files with 34 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
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 []
self.conn.search(LdapService.search_base, search_string, attributes=LdapService.attributes) elif query.endswith(' '):
search_string = LdapService.cn_single_search % (query.strip())
# Entries are returned as a generator, accessing entries elif query.strip().count(',') == 1:
# can make subsequent calls to the ldap service, so limit f, l = query.split(",")
# those here. search_string = LdapService.cn_double_search % (l.strip(), f.strip())
count = 0 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 = [] results = []
for entry in self.conn.entries: print(search_string)
if count > limit: try:
break self.conn.search(LdapService.search_base, search_string, attributes=LdapService.attributes)
results.append(LdapUserInfo.from_entry(entry)) # Entries are returned as a generator, accessing entries
count += 1 # can make subsequent calls to the ldap service, so limit
# those here.
count = 0
for entry in self.conn.entries:
if count > limit:
break
results.append(LdapUserInfo.from_entry(entry))
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