forked from DGNum/gestioCOF
Switch to python-ldap (instead of ldap3)
This commit is contained in:
parent
3ca8b45014
commit
028b6f6cb7
4 changed files with 41 additions and 31 deletions
|
@ -317,10 +317,11 @@ class RegistrationAutocompleteViewTests(MockLDAPMixin, ViewTestCaseMixin, TestCa
|
||||||
|
|
||||||
self._test("aa bb", [], [], [Clipper("uid", "first last")])
|
self._test("aa bb", [], [], [Clipper("uid", "first last")])
|
||||||
|
|
||||||
mock_ldap.search.assert_called_once_with(
|
mock_ldap.ldap_obj.search_s.assert_called_once_with(
|
||||||
"dc=spi,dc=ens,dc=fr",
|
"dc=spi,dc=ens,dc=fr",
|
||||||
|
mock_ldap.SCOPE_SUBTREE,
|
||||||
"(&(|(cn=*aa*)(uid=*aa*))(|(cn=*bb*)(uid=*bb*)))",
|
"(&(|(cn=*aa*)(uid=*aa*))(|(cn=*bb*)(uid=*bb*)))",
|
||||||
attributes=["cn", "uid"],
|
["cn", "uid"],
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_clipper_escaped(self):
|
def test_clipper_escaped(self):
|
||||||
|
@ -328,7 +329,7 @@ class RegistrationAutocompleteViewTests(MockLDAPMixin, ViewTestCaseMixin, TestCa
|
||||||
|
|
||||||
self._test("; & | (", [], [], [])
|
self._test("; & | (", [], [], [])
|
||||||
|
|
||||||
mock_ldap.search.assert_not_called()
|
mock_ldap.ldap_obj.search_s.assert_not_called()
|
||||||
|
|
||||||
def test_clipper_no_duplicate(self):
|
def test_clipper_no_duplicate(self):
|
||||||
self.mockLDAP([("uid", "abc")])
|
self.mockLDAP([("uid", "abc")])
|
||||||
|
|
|
@ -11,4 +11,4 @@ asgiref==1.1.1
|
||||||
daphne==1.3.0
|
daphne==1.3.0
|
||||||
|
|
||||||
# ldap bindings
|
# ldap bindings
|
||||||
ldap3
|
python-ldap
|
||||||
|
|
|
@ -22,32 +22,33 @@ class MockLDAPMixin:
|
||||||
appeler `with Connection(*args, **kwargs) as foo` pour que le test fonctionne.
|
appeler `with Connection(*args, **kwargs) as foo` pour que le test fonctionne.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
class MockLDAPModule:
|
||||||
|
SCOPE_SUBTREE = None # whatever
|
||||||
|
|
||||||
|
def __init__(self, ldap_obj):
|
||||||
|
self.ldap_obj = ldap_obj
|
||||||
|
|
||||||
|
def initialize(self, *args):
|
||||||
|
"""Always return the same ldap object."""
|
||||||
|
return self.ldap_obj
|
||||||
|
|
||||||
def mockLDAP(self, results):
|
def mockLDAP(self, results):
|
||||||
class Elt:
|
entries = [
|
||||||
def __init__(self, value):
|
("whatever", {"cn": [name.encode("utf-8")], "uid": [uid.encode("utf-8")]})
|
||||||
self.value = value
|
for uid, name in results
|
||||||
|
]
|
||||||
|
# Mock ldap object whose `search_s` method always returns the same results.
|
||||||
|
mock_ldap_obj = mock.Mock()
|
||||||
|
mock_ldap_obj.search_s = mock.Mock(return_value=entries)
|
||||||
|
|
||||||
class Entry:
|
# Mock ldap module whose `initialize_method` always return the same ldap object.
|
||||||
def __init__(self, **kwargs):
|
mock_ldap_module = self.MockLDAPModule(mock_ldap_obj)
|
||||||
for k, v in kwargs.items():
|
|
||||||
setattr(self, k, Elt(v))
|
|
||||||
|
|
||||||
results_as_ldap = [Entry(uid=uid, cn=name) for uid, name in results]
|
patcher = mock.patch("shared.views.autocomplete.ldap", new=mock_ldap_module)
|
||||||
|
|
||||||
mock_connection = mock.MagicMock()
|
|
||||||
mock_connection.entries = results_as_ldap
|
|
||||||
|
|
||||||
# Connection is used as a context manager.
|
|
||||||
mock_context_manager = mock.MagicMock()
|
|
||||||
mock_context_manager.return_value.__enter__.return_value = mock_connection
|
|
||||||
|
|
||||||
patcher = mock.patch(
|
|
||||||
"shared.views.autocomplete.Connection", new=mock_context_manager
|
|
||||||
)
|
|
||||||
patcher.start()
|
patcher.start()
|
||||||
self.addCleanup(patcher.stop)
|
self.addCleanup(patcher.stop)
|
||||||
|
|
||||||
return mock_connection
|
return mock_ldap_module
|
||||||
|
|
||||||
|
|
||||||
class CSVResponseMixin:
|
class CSVResponseMixin:
|
||||||
|
|
|
@ -5,11 +5,11 @@ from django.conf import settings
|
||||||
from django.db.models import Q
|
from django.db.models import Q
|
||||||
|
|
||||||
if getattr(settings, "LDAP_SERVER_URL", None):
|
if getattr(settings, "LDAP_SERVER_URL", None):
|
||||||
from ldap3 import Connection
|
import ldap
|
||||||
else:
|
else:
|
||||||
# shared.tests.testcases.TestCaseMixin.mockLDAP needs
|
# shared.tests.testcases.TestCaseMixin.mockLDAP needs
|
||||||
# Connection to be defined
|
# an ldap object to be in the scope
|
||||||
Connection = None
|
ldap = None
|
||||||
|
|
||||||
|
|
||||||
class SearchUnit:
|
class SearchUnit:
|
||||||
|
@ -125,12 +125,20 @@ class LDAPSearch(SearchUnit):
|
||||||
|
|
||||||
query = self.get_ldap_query(keywords)
|
query = self.get_ldap_query(keywords)
|
||||||
|
|
||||||
if Connection is None or query == "(&)":
|
if ldap is None or query == "(&)":
|
||||||
return []
|
return []
|
||||||
|
|
||||||
with Connection(self.ldap_server_url) as conn:
|
ldap_obj = ldap.initialize(self.ldap_server_url)
|
||||||
conn.search(self.domain_component, query, attributes=self.search_fields)
|
res = ldap_obj.search_s(
|
||||||
return [Clipper(entry.uid.value, entry.cn.value) for entry in conn.entries]
|
self.domain_component, ldap.SCOPE_SUBTREE, query, self.search_fields
|
||||||
|
)
|
||||||
|
return [
|
||||||
|
Clipper(
|
||||||
|
clipper=attrs["uid"][0].decode("utf-8"),
|
||||||
|
fullname=attrs["cn"][0].decode("utf-8"),
|
||||||
|
)
|
||||||
|
for (_, attrs) in res
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
# ---
|
# ---
|
||||||
|
|
Loading…
Reference in a new issue