Python ldap
Flavien
Bonjour,
Pour le travail je travaille sur la bibliothèque python ldap. J'aimerai écrire des tests unitaires pour une fonctione de ce type :
def set_account_ldap(self, dn: AnyStr, newStatus):
"""Modification du statut d'une entrée LDAP"""
annuaire: LDAP = LDAP(configuration=self.__configuration)
# extraire l'uid du nom distingué (dn)
uid = self.__extract_uid(dn=dn)
self.logger.info("[LDAP] : activation de compte %s", uid)
annuaire.update(
dn=dn, name=self.attribut_account_status, value=newStatus
)
Voila le test que j'ai écrit pour le moment (il ne fonctionne pas).
@pytest.fixture
def mock_ldap(monkeypatch):
# Mock the LDAP class
ldap_mock = Mock()
monkeypatch.setattr("mce_sync.core.ldap.LDAP", ldap_mock)
return ldap_mock
def test_set_account_ldap(mock_ldap):
# Create a mock instance of LDAP
annuaire_mock = Mock()
mock_ldap.return_value = annuaire_mock
# Créez un mock pour représenter un objet Config
config_mock = Mock()
config_mock.ldap_write.uri = "ldap://your_ldap_server"
# Créez un mock pour représenter l'objet LDAP
ldap_mock = Mock()
ldap_mock.initialize.return_value = None
ldap_instance_mock = Mock()
ldap_instance_mock.simple_bind_s.return_value = None
ldap_mock.return_value = ldap_instance_mock
# Create an instance of Treatments
treatments = Treatments(
configuration=config_mock, ldap_pool=ldap_mock, save_replay=Mock()
)
# Mock __extract_uid method
treatments._Treatments__extract_uid = Mock(return_value="testuser")
# Call the set_account_ldap method
treatments.set_account_ldap(dn="test_dn", newStatus="active")
# Assert that the LDAP update method was called with the correct parameters
annuaire_mock.update.assert_called_once_with(
dn="test_dn", name=treatments.attribut_account_status, value="active"
)
Voila l'erreur :
@pytest.fixture
def mock_ldap(monkeypatch):
# Mock the LDAP class
ldap_mock = Mock()
monkeypatch.setattr("mce_sync.core.ldap.LDAP", ldap_mock)
return ldap_mock
def test_set_account_ldap(mock_ldap):
# Create a mock instance of LDAP
annuaire_mock = Mock()
mock_ldap.return_value = annuaire_mock
# Créez un mock pour représenter un objet Config
config_mock = Mock()
config_mock.ldap_write.uri = "ldap://your_ldap_server"
# Créez un mock pour représenter l'objet LDAP
ldap_mock = Mock()
ldap_mock.initialize.return_value = None
ldap_instance_mock = Mock()
ldap_instance_mock.simple_bind_s.return_value = None
ldap_mock.return_value = ldap_instance_mock
# Create an instance of Treatments
treatments = Treatments(
configuration=config_mock, ldap_pool=ldap_mock, save_replay=Mock()
)
# Mock __extract_uid method
treatments._Treatments__extract_uid = Mock(return_value="testuser")
# Call the set_account_ldap method
treatments.set_account_ldap(dn="test_dn", newStatus="active")
# Assert that the LDAP update method was called with the correct parameters
annuaire_mock.update.assert_called_once_with(
dn="test_dn", name=treatments.attribut_account_status, value="active"
)
Je me demande si j'ai la bonne approche ? Est-ce que je ne devrais pas essayer de mocker un serveur ldap complet et l'utiliser dans mes tests et si oui comment m'y prendre ? J'ai vu qu'il existait cette documentation :
https://ldap3.readthedocs.io/en/latest/mocking.html
Mais j'ai du mal à comprendre si je devrais me baser dessus pour mocker mon serveur et si ça a vraiment un interet pour moi.
Thibault houdon
Mentor
Salut Flavien !
Je pense que tu as mal collé l'erreur et juste mis le code 2 fois.
Concernant ta deuxième question, je n'ai jamais utilisé ldap donc à prendre avec un grain de sel mais ça semble effectivement une approche à envisager, avec la possibilité de loader des données depuis un JSON pour avoir des données à tester (un peu comme des fixtures qu'on load dans une base de données pour faire des tests spécifiques).
OK merci j'ai donc tenté avec la nouvelle approche comme ceci :
J'ai créer un faux serveur. A priori je n'ai pas chargé de JSON donc je pense qu'il est vide à priori.
import pytest
from ldap3 import Server, Connection, MODIFY_REPLACE
from mce_sync.core.treatments import Treatments
@pytest.fixture
def fake_ldap_server():
# Créez un faux serveur LDAP local
server = Server("localhost")
# Créez une connexion au serveur
conn = Connection(
server, user="cn=admin,dc=example,dc=com", password="adminpassword"
)
conn.bind()
yield server
conn.unbind()
@pytest.fixture
def treatments(fake_ldap_server):
# Créez une instance de Treatments avec la configuration du faux serveur
config_mock = Mock()
config_mock.ldap_values.mce_mailbox_encode = "your_encoding_value"
config_mock.ldap_write.uri = "localhost"
return Treatments(
configuration=config_mock(), ldap_pool=None, save_replay=Mock()
)
def test_set_account_ldap(treatments, fake_ldap_server):
# Créez une connexion au faux serveur LDAP pour la vérification
conn = Connection(
fake_ldap_server,
user="cn=admin,dc=example,dc=com",
password="adminpassword",
)
conn.bind()
# Préparez les données pour le test
dn = "uid=testuser,ou=users,dc=example,dc=com"
new_status = "active"
# Appelez la fonction à tester
treatments.set_account_ldap(dn=dn, newStatus=new_status)
# Vérifiez que la modification a été effectuée dans le faux serveur LDAP
conn.search(dn, "(objectClass=*)", attributes=["name"])
entry = conn.entries[0]
assert entry["name"] == [new_status]
conn.unbind()
Je me retrouve avec cette erreur :
______________________________________________________________ ERROR at setup of test_set_account_ldap _______________________________________________________________
@pytest.fixture
def fake_ldap_server():
# Créez un faux serveur LDAP local
server = Server("localhost")
# Créez une connexion au serveur
conn = Connection(
server, user="cn=admin,dc=example,dc=com", password="adminpassword"
)
> conn.bind()
tests/treatments/unit/init.py/test_treatments.py:191:
<hr/>.venv/lib/python3.9/site-packages/ldap3/core/connection.py:589: in bind
self.open(read_server_info=False)
.venv/lib/python3.9/site-packages/ldap3/strategy/sync.py:57: in open
BaseStrategy.open(self, reset_usage, read_server_info)
self = <ldap3.strategy.sync.syncstrategy 0x7ffa7f223c10="" at="" object="">, reset_usage = True, read_server_info = False
def open(self, reset_usage=True, read_server_info=True):
"""
Open a socket to a server. Choose a server from the server pool if available
"""
if log_enabled(NETWORK):
log(NETWORK, 'opening connection for <%s>', self.connection)
if self.connection.lazy and not self.connection._executing_deferred:
self.connection._deferred_open = True
self.connection.closed = False
if log_enabled(NETWORK):
log(NETWORK, 'deferring open connection for <%s>', self.connection)
else:
if not self.connection.closed and not self.connection._executing_deferred: # try to close connection if still open
self.close()
self._outstanding = dict()
if self.connection.usage:
if reset_usage or not self.connection._usage.initial_connection_start_time:
self.connection._usage.start()
if self.connection.server_pool:
new_server = self.connection.server_pool.get_server(self.connection) # get a server from the server_pool if available
if self.connection.server != new_server:
self.connection.server = new_server
if self.connection.usage:
self.connection._usage.servers_from_pool += 1
exception_history = []
if not self.no_real_dsa: # tries to connect to a real server
for candidate_address in self.connection.server.candidate_addresses():
try:
if log_enabled(BASIC):
log(BASIC, 'try to open candidate address %s', candidate_address[:-2])
self._open_socket(candidate_address, self.connection.server.ssl, unix_socket=self.connection.server.ipc)
self.connection.server.current_address = candidate_address
self.connection.server.update_availability(candidate_address, True)
break
except Exception as e:
self.connection.server.update_availability(candidate_address, False)
# exception_history.append((datetime.now(), exc_type, exc_value, candidate_address[4]))
exception_history.append((type(e)(str(e)), candidate_address[4]))
if not self.connection.server.current_address and exception_history:
if len(exception_history) == 1: # only one exception, reraise
if log_enabled(ERROR):
log(ERROR, '<%s> for <%s>', str(exception_history[0][0]) + ' ' + str((exception_history[0][1])), self.connection)
> raise exception_history[0][0]
E ldap3.core.exceptions.LDAPSocketOpenError: socket connection error while opening: [Errno 111] Connection refused
.venv/lib/python3.9/site-packages/ldap3/strategy/base.py:146: LDAPSocketOpenError
</ldap3.strategy.sync.syncstrategy>
Thibault houdon
Mentor
Salut Flavien !
Dans la documentation pour le mocking de ldap, ils indiquent que tu dois indiquer MOCK_SYNC pour la stratégie :
fake_connection = Connection(
fake_server,
user='cn=my_user,ou=test,o=lab',
password='my_password',
client_strategy=MOCK_SYNC # 👈
)
Après là il s'agit d'une erreur de connexion, comme si le serveur ne tournait pas. Je te conseillerais de mettre des points de debug et lancer ton test en mode debug pour voir le cheminement de ton programme et t'assurer que ton test est bien lancé après le lancement du serveur et que la connexion ne se termine pas avant.
Ok merci, effectivement je suis passé un peu vite sur la documentation.
Je me posais des questions sur la structure de mes tests :
Actuellement j'ai une structure de dossier suivante :
tests/
config/
mon_fichier_json.json
treatments/ correspond à un module de mon projet
unit/
test_treatments.py
integration/
Dans la plupart de mes tests je vais avoir besoin de tester des interactions avec le serveur ldap. Du coup je me demandais où mettre mon faux serveur. Est que je devrais le mettre à la racine de mon dossier dans un module séparé que j'appelerai fake_server.py par exemple ou dois le mettre dans chacun de mes modules tests, ou existe-il encore une autre solution ? Quelles sont les recomandations et meilleur pratique sur le sujet ?
Thibault houdon
Mentor
Tu peux mettre ton faux serveur LDAP dans un module séparé oui, et l'importer dans les modules de test qui en ont besoin.
Tu peux le mettre à la racine de ton dossier de tests ou même dans un sous-dossier fixtures ou mocks.
Si tu commences à avoir pas mal de tests différents (unitaires, integration, etc), tu pourrais avoir une structure qui ressemble à ça :
tests/
fixtures/
__init__.py
fake_server.py
config/
mon_fichier_json.json
treatments/
unit/
__init__.py
test_treatments.py
integration/
__init__.py
...
Et comme ça on comprend bien que ton serveur de test sert de fixture pour tes tests avec le nom du dossier quand tu fais l'import :
from tests.fixtures.fake_server import fake_ldap_server
Flavien
OK merci !
Je cloture la question, j'en ouvirai une si nécéssaire.
Inscris-toi
(c'est gratuit !)
Tu dois créer un compte pour participer aux discussions.
Créer un compte