Produits Entr'ouvert » Authentic 2

from django.core.exceptions import PermissionDenied, MultipleObjectsReturned

from django.db import IntegrityError

            try:

                account = models.FcAccount.objects.select_related().get(sub=sub)

            except MultipleObjectsReturned:

                account = models.FcAccount.objects.select_related().get(sub=sub, order=0)

        else:

            user = account.user

            logger.debug(u'found user %s with sub %s', user, sub)

            if not user.is_active:

                logger.info(u'user %s login refused, it is inactive', user)

                raise PermissionDenied

                try:

                    models.FcAccount.objects.create(

                        user=user,

                        sub=sub,

                        order=0,

                        token=json.dumps(kwargs['token']))

                except IntegrityError:

                    # uniqueness check failed, as the user is new, it can only means that the sub is not unique

                    # let's try again

                    user.delete()

                    return self.authenticate(sub, **kwargs)

                else:

                    logger.debug(u'user creation enabled with fc_account (sub : %s - token : %s)',

                                 sub, json.dumps(kwargs['token']))

                    hooks.call_hooks('event', name='fc-create', user=user, sub=sub)

# -*- coding: utf-8 -*-

# Generated by Django 1.11.20 on 2019-06-12 21:27

from __future__ import unicode_literals

from collections import Counter

from django.db import migrations, models

def set_fcaccount_order(apps, schema_editor):

    FcAccount = apps.get_model('authentic2_auth_fc', 'FcAccount')

    c = Counter()

    FcAccount.objects.update(order=0)

    user_ids = (

        FcAccount.objects

        .values('user_id')

        .annotate(total=models.Count('id'))

        .filter(total__gt=1)

        .values_list('user_id', flat=True)

    )

    subs = (

        FcAccount.objects

        .values_list('sub')

        .annotate(total=models.Count('id'))

        .filter(total__gt=1)

        .values_list('sub', flat=True)

    )

    for account in FcAccount.objects.filter(models.Q(user_id__in=user_ids) | models.Q(sub__in=subs)):

        order = max(c[account.user_id], c[account.sub])

        if account.order != order:

            account.order = order

            account.save()

        c[account.user_id] = order + 1

        c[account.sub] = order + 1

    assert FcAccount.objects.filter(order__isnull=True).count() == 0

def noop(apps, schema_editor):

    pass

class Migration(migrations.Migration):

    # Disembiguate sub and user_id columns using an integer order column, then

    # force sub/order and user_id/order pairs unique to be unique in order

    # to move to a model where user and sub are unique without breaking on existing data

    dependencies = [

        ('authentic2_auth_fc', '0001_initial'),

    ]

    operations = [

        migrations.AddField(

            model_name='fcaccount',

            name='order',

            field=models.IntegerField(null=True, verbose_name='order'),

        ),

        migrations.RunPython(set_fcaccount_order, noop),

    ]

# -*- coding: utf-8 -*-

# Generated by Django 1.11.20 on 2019-06-13 09:44

from __future__ import unicode_literals

from django.conf import settings

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        migrations.swappable_dependency(settings.AUTH_USER_MODEL),

        ('authentic2_auth_fc', '0002_fcaccount_order1'),

    ]

    operations = [

        migrations.AlterField(

            model_name='fcaccount',

            name='order',

            field=models.PositiveIntegerField(default=0, verbose_name='order'),

        ),

        migrations.AlterUniqueTogether(

            name='fcaccount',

            unique_together=set([('sub', 'order'), ('user', 'order')]),

        ),

    ]

    order = models.PositiveIntegerField(

        verbose_name=_('order'),

        default=0)

    class Meta:

        unique_together = [

            ('sub', 'order'),

            ('user', 'order'),

        ]

from django.db import IntegrityError

    user_info = None

    @property

    def fc_display_name(self):

        '''Human representation of the current FC account'''

        if not self.user_info:

            return u''

        return u'{0} {1}'.format(self.user_info['family_name'], self.user_info['given_name'])

        self.fc_account.token = json.dumps(self.token)

        self.fc_account.save(update_fields=['token', 'user_info'])

    def uniqueness_check_failed(self, request):

        if models.FcAccount.objects.filter(user=request.user, order=0).count():

            messages.error(request,

                           _('Your account is already linked to a FranceConnect account'))

        else:

            messages.error(request,

                           _('The FranceConnect account {} is already'

                             ' linked with another account.').format(self.fc_display_name))

        return self.redirect(request)

                self.fc_account, created = models.FcAccount.objects.get_or_create(

                    sub=self.sub, user=request.user, order=0,

                    defaults={'token': json.dumps(self.token)})

            except IntegrityError:

                # unique index check failed, find why.

                return self.uniqueness_check_failed(request)

                messages.info(request,

                              _('Your FranceConnect account {} has been linked.').format(self.fc_display_name))

                hooks.call_hooks('event', name='fc-link', user=request.user, sub=self.sub, request=request)

            self.update_user_info()

        if user:

            self.fc_account = user.fc_accounts.get(order=0)

                        try:

                            self.fc_account, created = models.FcAccount.objects.get_or_create(

                                defaults={'token': json.dumps(self.token)},

                                sub=self.sub, user=user, order=0)

                        except IntegrityError:

                            # unique index check failed, find why.

                            return self.uniqueness_check_failed(request)

def test_migration_0002(migration):

    migrate_from = [('authentic2_auth_fc', '0001_initial')]

    # it's a two-parts migration, as it contains data and schema changes.

    migrate_to = [('authentic2_auth_fc', '0003_fcaccount_order2')]

    old_apps = migration.before(migrate_from)

    User = old_apps.get_model('custom_user', 'User')

    FcAccount = old_apps.get_model('authentic2_auth_fc', 'FcAccount')

    user1 = User.objects.create(username='user1')

    user2 = User.objects.create(username='user2')

    user3 = User.objects.create(username='user3')

    FcAccount.objects.create(user=user1, sub='sub1')

    FcAccount.objects.create(user=user1, sub='sub2')

    FcAccount.objects.create(user=user2, sub='sub2')

    FcAccount.objects.create(user=user2, sub='sub3')

    FcAccount.objects.create(user=user3, sub='sub3')

    assert len(set(FcAccount.objects.values_list('user_id', flat=True))) == 3

    assert len(set(FcAccount.objects.values_list('sub', flat=True))) == 3

    assert FcAccount.objects.count() == 5

    # execute migration

    new_apps = migration.apply(migrate_to)

    FcAccount = new_apps.get_model('authentic2_auth_fc', 'FcAccount')

    assert len(set(FcAccount.objects.values_list('user_id', 'order'))) == 5

    assert len(set(FcAccount.objects.values_list('sub', 'order'))) == 5

import django

from django.db.migrations.executor import MigrationExecutor

from django.db import connection

from django.core.management import call_command

@pytest.fixture()

def migration(request, transactional_db):

    # see https://gist.github.com/asfaltboy/b3e6f9b5d95af8ba2cc46f2ba6eae5e2

    if django.VERSION < (1, 9):

        pytest.skip('migration fixture only works with Django 1.9')

    """

    This fixture returns a helper object to test Django data migrations.

    The fixture returns an object with two methods;

     - `before` to initialize db to the state before the migration under test

     - `after` to execute the migration and bring db to the state after the migration

    The methods return `old_apps` and `new_apps` respectively; these can

    be used to initiate the ORM models as in the migrations themselves.

    For example:

        def test_foo_set_to_bar(migration):

            old_apps = migration.before([('my_app', '0001_inital')])

            Foo = old_apps.get_model('my_app', 'foo')

            Foo.objects.create(bar=False)

            assert Foo.objects.count() == 1

            assert Foo.objects.filter(bar=False).count() == Foo.objects.count()

            # executing migration

            new_apps = migration.apply([('my_app', '0002_set_foo_bar')])

            Foo = new_apps.get_model('my_app', 'foo')

            assert Foo.objects.filter(bar=False).count() == 0

            assert Foo.objects.filter(bar=True).count() == Foo.objects.count()

    Based on: https://gist.github.com/blueyed/4fb0a807104551f103e6

    """

    class Migrator(object):

        def before(self, targets):

            """ Specify app and starting migration names as in:

                before([('app', '0001_before')]) => app/migrations/0001_before.py

            """

            self.executor = MigrationExecutor(connection)

            # prepare state of db to before the migration ("migrate_from" state)

            self._old_apps = self.executor.migrate(targets).apps

            return self._old_apps

        def apply(self, targets):

            """ Migrate forwards to the "targets" migration """

            self.executor.loader.build_graph()  # reload.

            self._new_apps = self.executor.migrate(targets).apps

            return self._new_apps

    # ensure to migrate forward migrated apps all the way after test

    def migrate_to_end():

        call_command('migrate', verbosity=0)

    request.addfinalizer(migrate_to_end)

    return Migrator()