lc_ldap/test.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import psycopg2
import traceback
import random
import string
import lc_ldap
import config
from affich_tools import anim, OK, cprint, ERREUR

show_traceback = False

def randomMAC():
    mac = [ 0x00, 0x16, 0x3e,
        random.randint(0x00, 0x7f),
        random.randint(0x00, 0xff),
        random.randint(0x00, 0xff) ]
    return ':'.join(map(lambda x: "%02x" % x, mac))

def randomStr(n=10):
    return ''.join( random.choice(string.lowercase + string.digits) for i in range(n))

adherent_ldif = {
 'prenom' : ['toto'],
 'nom' : ['passoir'],
 'chbre' : ['EXT'],
 'tel' : ['0000000000'],
 'mail' : ['nobody@test.org'],
 'etudes' : ['ENS', '1', 'A0'],
}

machine_ldif = {
 'macAddress' : [randomMAC()],
 'host' : ["autotest-%s.crans.org" % randomStr() ]
}

club_ldif = {
 'nom' : [ 'autotest-club' ],
 'chbre' : [ 'EXT' ],
}

facture_ldif = {}


def keys_of_list_of_dict(l, type=''):
    """Récupère la liste de tous les attributs existant.
    Teste l'écriture des objets si on est en mode rw"""
    keys = set()
    if l:
        anime=anim("Test des objets %r" % type, len(l))
        anime.reinit()
    ok=True
    for item in l:
        keys = keys.union(item.attrs.keys())
        try:
            if item.mode in  ['w', 'rw']:
                item.save()
        except Exception as error:
            anime.reinit()
            print ERREUR
            if show_traceback: print traceback.format_exc()
            else: print "\t%r" % error
            ok = False
        anime.cycle()
    if l:
        anime.reinit()
        print OK if ok else ERREUR
    keys = list(keys)
    keys.sort()
    return keys

def test_list_of_dict(keys, list):
    """Test l'instanciation des attributs"""
    for key in keys:
        anim("\tTest de l'attribut %s" % key)
        ok = True
        for item in list:
            try: item.get(key, [])
            except psycopg2.OperationalError as error:
                print ERREUR
                if show_traceback: print traceback.format_exc()
                else: print "\t%r" % error
                return
            except Exception as error:
                if ok: print ERREUR
                if show_traceback: print traceback.format_exc()
                else: print "\t%r" % error
                ok = False
        if ok: print OK

def tests_machines(parent_dn, realm_list, ipsec=False):
    """Test la création et la suppression de machines dans la realm_list"""
    for realm in realm_list:
        anim("Creation de machines %s" % realm)
        try: 
            machine = conn.newMachine(parent_dn, realm,  machine_ldif)
            if ipsec: machine['ipsec'] = 'auto'
            machine.create()
        except Exception as error:
            print ERREUR
            if show_traceback: print traceback.format_exc()
            else: print "\t%r" % error
        else:
            print OK
            anim("Suppression d'une machines %s" % realm)
            try:
                machine = conn.search('mid=%s' % machine['mid'][0], mode='rw')[0]
                machine.delete()
            except Exception as error:
                print ERREUR
                if show_traceback: print traceback.format_exc()
                else: print "\t%r" % error
            else: print OK



print "Test de la librairie lc_ldap"

print "Connection"
conn= lc_ldap.lc_ldap_test()

print u"Tests effectués avec les droits %s " % ', '.join(conn.droits)

###############################################
# Instanciation de tous les attributs de tous #
# les adhérents et de toutes les machines     #
###############################################

anim("Appel de allMachinesAdherents en rw")
try:
    machines, adherents = conn.allMachinesAdherents(mode='rw')
except EnvironmentError as error:
    print ERREUR
    if show_traceback: print traceback.format_exc()
    else: print "\t%s" % error
    exit(1)
except Exception as error:
    print ERREUR
    if show_traceback: print traceback.format_exc()
    else: print "\t%r" % error
    anim("Fallback en ro")
    machines, adherents = conn.allMachinesAdherents()
    print OK
else:
    print OK
machines_attrs_keys = keys_of_list_of_dict(machines, 'machines')
print "Test des attributs des machines"
test_list_of_dict(machines_attrs_keys, machines)

adherents_attrs_keys = keys_of_list_of_dict(adherents, 'adherents')
print "Test des attributs des adhérents"
test_list_of_dict(adherents_attrs_keys, adherents)


print "Test de création d'objets"
print "Instanciation"


##############################################
# Création  d'un Adhérent et de ses machines #
##############################################

anim("Creation d'un adherent")
try:
    adherent = conn.newAdherent(adherent_ldif)
    adherent.create()
except:
    print ERREUR
    print traceback.format_exc()
    adherent = None
else:
    print OK

    tests_machines(adherent.dn, ["adherents", "fil-adherents", "fil-v6", "personnel-ens"])
    tests_machines(adherent.dn, ["wifi", "wifi-v6"], ipsec=True)




#############################################
# Création et suppression de machines Crans #
#############################################

tests_machines(lc_ldap.base_dn, ["adm", "serveurs", "serveurs-v6", "adm-v6"])
tests_machines(lc_ldap.base_dn, ["bornes"])


######################
# Création d'un club #
######################

anim("Creation d'un club")
try:
    club = conn.newClub(club_ldif)
    club['responsable'] = str(adherent['aid'][0])
    club.create()
except:
    print ERREUR
    print traceback.format_exc()
else:
    print OK

    tests_machines(club.dn, ["adherents"])

    anim("Suppression d'un club")
    try:
        club = conn.search('cid=%s' % club['cid'][0], mode='rw')[0] 
        club.delete()
    except: 
        print ERREUR
        print traceback.format_exc()
    else: print OK





############################
# Supression d'un adhérent #
############################

if adherent:
    anim("Suppression d'un adherent")
    try:
        adherent = conn.search('aid=%s' % adherent['aid'][0], mode='rw')[0] 
        adherent.delete()
    except: 
        print ERREUR
        print traceback.format_exc()
    else: print OK


#######################################
# Création et suppression de factures #
#######################################

anim("Creation d'une facture")
try:
    facture = conn.newFacture(facture_ldif)
except Exception as error:
    print ERREUR
    if show_traceback: print traceback.format_exc()
    else: print "\t%r" % error
else:
    print OK
    try:
        facture = conn.search('fid=%s' % facture['fid'][0], mode='rw')[0]
        facture.delete()
    except:
        print ERREUR
        print traceback.format_exc()
    else: print OK