From ddc77c8e348b0f67f2777d3a49bd4a4af1602a2b Mon Sep 17 00:00:00 2001 From: Marcio Mazza Date: Fri, 30 Jan 2015 19:12:15 -0200 Subject: [PATCH] Add initial tests for sync_ldap command --- .../management/commands/sync_ldap.py | 24 ++-- .../management/commands/test_sync_ldap.py | 130 ++++++++++++++++++ 2 files changed, 144 insertions(+), 10 deletions(-) create mode 100644 sigi/apps/servidores/management/commands/test_sync_ldap.py diff --git a/sigi/apps/servidores/management/commands/sync_ldap.py b/sigi/apps/servidores/management/commands/sync_ldap.py index 8a11f63..6c2c9c3 100644 --- a/sigi/apps/servidores/management/commands/sync_ldap.py +++ b/sigi/apps/servidores/management/commands/sync_ldap.py @@ -1,9 +1,10 @@ # coding: utf-8 import ldap -from django.core.management.base import BaseCommand, CommandError from django.contrib.auth.models import User, Group -from sigi.settings import * +from django.core.management.base import BaseCommand + from sigi.apps.servidores.models import Servidor +from sigi.settings import * class Command(BaseCommand): @@ -56,7 +57,7 @@ class Command(BaseCommand): def get_ldap_property(ldap_user, property_name, default_value=None): value = ldap_user[1].get(property_name, None) - return value[0] if value else default_value + return value[0].decode('utf8') if value else default_value for ldap_user in ldap_users: username = get_ldap_property(ldap_user, 'sAMAccountName') @@ -69,27 +70,30 @@ class Command(BaseCommand): except User.DoesNotExist: try: user = User.objects.get(email=email) + old_username = user.username user.username = username + print "User with email '%s' had his/her username updated from [%s] to [%s]." % ( + email, old_username, username) except User.DoesNotExist: user = User.objects.create_user( username=username, - email=email + first_name=first_name, + last_name=last_name, + email=email, ) print "User '%s' created." % username - user.first_name = first_name - user.last_name = last_name - if not user.email == email: - user.email = email - print "User '%s' email updated." % username if not user.first_name == first_name: user.first_name = first_name print "User '%s' first name updated." % username if not user.last_name == last_name: user.last_name = last_name print "User '%s' last name updated." % username + if not user.email == email: + user.email = email + print "User '%s' email updated." % username - nome_completo = get_ldap_property(ldap_user, 'cn', '').decode('utf8') + nome_completo = get_ldap_property(ldap_user, 'cn', '') try: servidor = user.servidor except Servidor.DoesNotExist: diff --git a/sigi/apps/servidores/management/commands/test_sync_ldap.py b/sigi/apps/servidores/management/commands/test_sync_ldap.py new file mode 100644 index 0000000..8b74825 --- /dev/null +++ b/sigi/apps/servidores/management/commands/test_sync_ldap.py @@ -0,0 +1,130 @@ +# -*- coding: utf-8 -*- +import pytest +from django.contrib.auth.models import User +from django_dynamic_fixture import G + +from sigi.apps.servidores.management.commands.sync_ldap import Command + + +pytestmark = pytest.mark.django_db + + +class StubCommand(Command): + + def __init__(self, users): + super(StubCommand, self).__init__() + self.users = users + + def get_ldap_users(self): + return self.users + + +def create_stub_user(username, nome_completo, first_name, last_name, email): + user = G(User, username=username, first_name=first_name, last_name=last_name, email=email) + user.servidor.nome_completo = nome_completo + return user + +ALEX_LDAP, BRUNO_LDAP, RITA_LDAP = [ + ('...', + {'cn': ['Alex Lima'], + 'givenName': ['Alex'], + 'sAMAccountName': ['alexlima'], + 'sn': ['Lima'], + 'userPrincipalName': ['alexlima@interlegis.leg.br']}), + + ('...', + {'cn': ['Bruno Almeida Prado'], + 'givenName': ['Bruno'], + 'sAMAccountName': ['bruno'], + 'sn': ['Almeida Prado'], + 'userPrincipalName': ['bruno@interlegis.leg.br']}), + + ('...', + {'cn': ['Cl\xc3\xa1udia de C\xc3\xa1ssia'], + 'givenName': ['Cl\xc3\xa1udia'], + 'sAMAccountName': ['claudia'], + 'sn': ['de C\xc3\xa1ssia'], + 'userPrincipalName': ['claudia@interlegis.leg.br']}), +] + + +@pytest.mark.parametrize("before, ldap_users, after, messages", [ + # on an empy base a new user is created + ([], + [ALEX_LDAP], + [(u'alexlima', u'Alex Lima', u'Alex', u'Lima', u'alexlima@interlegis.leg.br')], + u''' +User 'alexlima' created. +Users are synchronized. + '''), + + # nothing changes + ([(u'alexlima', u'Alex Lima', u'Alex', u'Lima', u'alexlima@interlegis.leg.br')], + [ALEX_LDAP], + [(u'alexlima', u'Alex Lima', u'Alex', u'Lima', u'alexlima@interlegis.leg.br')], + u''' +Users are synchronized. + '''), + + # encoding from LDAP data works well + ([('claudia', u'Cláudia de Cássia', u'Cláudia', u'de Cássia', 'claudia@interlegis.leg.br', )], + [RITA_LDAP], + [(u'claudia', u'Cláudia de Cássia', u'Cláudia', u'de Cássia', u'claudia@interlegis.leg.br', )], + u''' +Users are synchronized. + '''), + + # update: first name, last name, email + ([('alexlima', '___', '___', '___', '___', ), + ('bruno', 'Bruno Almeida Prado', '___', 'Almeida Prado', '___', ), + ('claudia', '___', u'Cláudia', '___', 'claudia@interlegis.leg.br', )], + [ALEX_LDAP, BRUNO_LDAP, RITA_LDAP], + [(u'alexlima', u'Alex Lima', u'Alex', u'Lima', u'alexlima@interlegis.leg.br', ), + (u'bruno', u'Bruno Almeida Prado', u'Bruno', u'Almeida Prado', u'bruno@interlegis.leg.br', ), + (u'claudia', u'Cláudia de Cássia', u'Cláudia', u'de Cássia', u'claudia@interlegis.leg.br', )], + u''' +User 'alexlima' first name updated. +User 'alexlima' last name updated. +User 'alexlima' email updated. +Full name of Servidor 'Alex Lima' updated. +User 'bruno' first name updated. +User 'bruno' email updated. +Full name of Servidor 'Bruno Almeida Prado' updated. +User 'claudia' last name updated. +Full name of Servidor 'Cláudia de Cássia' updated. +Users are synchronized. + '''), + + # update username (username from LDAP not in base, so match user by email and update username) + + # create servidor + + # connect servidor with nome_completo to user + + # create new servidor with nome_completo and connect to user + + + # delete user not present in ldap ???? + # TODO not implemented. really implement?? what would be the side effects? + + +]) +def test_sync_users(before, ldap_users, after, messages, capsys): + + # setup + for user_setup in before: + if type(user_setup) == tuple: + create_stub_user(*user_setup) + assert User.objects.count() == len(before) + + command = StubCommand(ldap_users) + command.sync_users() + users = User.objects.all().order_by('username') + for user, expected in zip(users, after): + real = user.username, user.servidor.nome_completo, user.first_name, user.last_name, user.email + assert real == expected + + # feedbak messages + out, err = capsys.readouterr() + assert out.strip() == messages.strip() + assert err == ''