Produits Entr'ouvert » Mandaye

# Test App Settings

class Test(AppSettings):

    SITE_LOGIN_PATH = '/'

    SITE_LOCATORS = [

        {

            'id': '#login',

            'label': 'Login',

            'name': 'login',

            'kind': 'string',

        },

        {

            'id': '#password',

            'label': 'Password',

            'name': 'password',

            'kind': 'password'

        }

    ]

    SITE_AUTH_CHECKER = 'js/test/auth.checker.js'

    SITE_AUTH_COOKIE_KEYS = [ 'test']

    SITE_FORCE_REDIRECT_URL = '/whatever'

from __future__ import absolute_import

import os

import csv

import json

import ldif

import logging

from optparse import make_option

from django.core.management.base import BaseCommand, CommandError

from django.db import IntegrityError

from django.contrib.auth.models import User

from mandayejs.mandaye.models import UserCredentials  

logger = logging.getLogger(__name__)

class Command(BaseCommand):

    args = '<filename>'

    help = 'Migrate users from ldif file or csv file'

    option_list = BaseCommand.option_list +(

        make_option(

            '--ldap',

            action='store_true',

            default=False,

            help='Migrate users from a ldap dump file'

        ),

        make_option(

            '--csv',

            action='store_true',

            default=False,

            help='Migrate users from a csv file'

        )

    )

    def handle(self, *args, **kwargs):

        if len(args) < 1:

            self.stdout.write('An input file is required')

            return False

        if kwargs.get('csv'):

            for filename in args:

                data = self.get_csv_data(filename)

                self.migrate(data)

        else:

            for filename in args:

                data = self.get_ldif_data(filename)

                data = [ d[1] for d in data]

                data = [ 

                    { k : ''.join(v) for k,v in d.items()} for d in data

                ]

                self.migrate(data)

    def get_ldif_data(self, filename):

        with open(filename, 'rb') as fd:

            return ldif.ParseLDIF(fd)

    def get_csv_data(self, filename):

        with open(filename, 'rb') as fd:

            fieldnames = ['idpUniqueID', 'spPostValues']

            reader = csv.DictReader(fd, delimiter=';', quotechar='|', fieldnames=fieldnames)

            return list(reader)

    def migrate(self, parsed_data):

        for data in parsed_data:

            try:

                user, created = User.objects.get_or_create(username=data.get('idpUniqueID'), 

                        last_name=data.get('spLogin',''))

                user.save()

                uc = UserCredentials(user=user, locators=json.loads(data.get('spPostValues')))

                uc.decrypt()

                uc.save()

                logger.debug('{idpUniqueID} imported'.format(**data))

            except (IntegrityError,) as e:

                logger.debug(e)

                continue

_288CE3839831B75D8ADD6524329236B7;{"username": "csv1", "password": "cIrWyg=="}

_8C36D00EE18E5702A51813413097028F;{"username": "csv2", "password": "cIrWyQ=="}

_45EB3616CEE069A42AFD23D7039E6FBD;{"username": "csv3", "password": "cIrWyA=="}

_45EB3616CEJ069A42AFD23D70S9E6FBD;{"username": "csv4", "password": "cIrWzw=="}

spPostValues: {"password": "Y4HL6cbGxRsoHQU97VzXgkqEdA==", "login": "ldap_user1"}

spPostValues: {"password": "Y4HL6cbGxRsoHQU97VzXgkqEdw==", "login": "ldap_user2"}

spPostValues: {"password": "Y4HL6cbGxRsoHQU97VzXgkqEdg==", "login": "ldap_user3"}

settings.SITE_APP = 'mandayejs.applications.Test'

settings.SECRET_KEY = 'od5cei4aeveel8dui4lei2ou9ahsei2A'

# ENCRYPTION/DECRYPTION

# MIGRATION COMMAND

def cmd(*args, **kwargs):

    cmd = type(

        'Cmd',

        (object,),

        {

            'name': 'migrate-users',

            'args': args,

            'opts': kwargs

        }

    )

    return cmd 

@pytest.fixture

def command_ldap():

    return cmd(

            'tests/ldap_users.txt',

            ldap= True

        )

@pytest.fixture

def command_csv():

    return cmd(

        'tests/csv_users.csv',

         csv= True

    )

@pytest.fixture(params=['command_ldap', 'command_csv'])

def command(request, command_ldap, command_csv):

    return locals().get(request.param)

def test_command_migrate_users(command):

    call_command(command.name, *command.args, **command.opts)

    if command.opts.get('ldap'):

        credentials = UserCredentials.objects.filter(user__last_name__in=[

                'ldap_user1',

                'ldap_user2',

                'ldap_user3'

            ])

        assert len(credentials) == 3

        for cred in credentials:

            assert cred.to_login_info(decrypt=True)['#password'] == 'password_{}'.format(cred.user.last_name)

    else:

        credentials = UserCredentials.objects.all().exclude(user__last_name__in=[

        assert len(credentials) == 4

        for cred in credentials:

            assert cred.to_login_info(decrypt=True)['#password'] == cred.to_login_info()['#username']