Produits Entr'ouvert » Authentic 2

from django.contrib.postgres.fields.jsonb import KeyTextTransform

from django.db.models import QuerySet, Q, F, Value, Count

from django.db.models.functions import Trunc

    @classmethod

    def get_statistics(cls, group_by_time, group_by_field, group_by_references=False, start_datetime=None, end_datetime=None):

        if group_by_time not in ('timestamp', 'day', 'week', 'month', 'year'):

            raise ValueError('Usupported value for group_by_time: %s' % time_group_by)

        event_type = EventType.objects.get_for_name(cls.name)

        qs = Event.objects.filter(type=event_type)

        if start_datetime:

            qs = qs.filter(timestamp__gte=start_datetime)

        if end_datetime:

            qs = qs.filter(timestamp__lte=end_datetime)

        value_fields = []

        if group_by_time != 'timestamp':

            qs = qs.annotate(**{group_by_time: Trunc('timestamp', kind=group_by_time)})

        if not group_by_field.startswith('user') and not group_by_field.startswith('session'):

            # get field from JSONField

            qs = qs.annotate(**{group_by_field: KeyTextTransform(group_by_field, 'data')})

        values = [group_by_time, group_by_field]

        if group_by_references:

            values.append('reference_ids')

        qs = qs.values(*values)

        qs = qs.annotate(count=Count('id'))

        return qs.order_by(group_by_time)

from django.contrib.contenttypes.models import ContentType

from authentic2.apps.journal.models import EventTypeDefinition, n_2_pairing_rev

from .models import Service

        (service,) = event.get_typed_references(Service)

class EventTypeWithMethod(EventTypeWithService):

    @classmethod

    def record(cls, user, session, service, how):

        super().record(user=user, session=session, service=service, data={'how': how})

    @classmethod

    def get_method_statistics(cls, group_by_time='timestamp', **kwargs):

        stats = cls.get_statistics(group_by_time=group_by_time, group_by_field='how', **kwargs)

        x_labels = list(stats.distinct().values_list(group_by_time, flat=True))

        x_labels_indexes = {label: i for i, label in enumerate(x_labels)}

        empty_data = [None] * len(x_labels)

        y_data = {}

        for stat in stats:

            data = y_data.setdefault(stat['how'], empty_data.copy())

            x_label = stat[group_by_time]

            data[x_labels_indexes[x_label]] = stat['count']

        series = []

        for how, data in y_data.items():

            label = _(login_method_label(how)) if how else _('none')

            series.append({'label': label, 'data': data})

        return {

            'x_labels': [label.isoformat() for label in x_labels],

            'series': series,

        }

    @classmethod

    def _get_method_statistics_with_reference(cls, group_by_time, get_reference_label, **kwargs):

        stats = cls.get_statistics(group_by_time=group_by_time, group_by_field='how', group_by_references=True, **kwargs)

        x_labels = list(stats.distinct().values_list(group_by_time, flat=True))

        x_labels_indexes = {label: i for i, label in enumerate(x_labels)}

        empty_data = [0] * len(x_labels)

        loop_data = {}

        service_ct_id = ContentType.objects.get_for_model(Service).pk

        labels_cache = {}

        for stat in stats:

            for reference_id in stat['reference_ids'] or []:

                content_type_id, instance_pk = n_2_pairing_rev(reference_id)

                if content_type_id == service_ct_id:

                    label = labels_cache.get(instance_pk)

                    if not label:

                        label = get_reference_label(instance_pk)

                        labels_cache[instance_pk] = label

                    break

            else:

                label = _('None')

            y_data = loop_data.setdefault(label, {})

            data = y_data.setdefault(stat['how'], empty_data.copy())

            x_label = stat[group_by_time]

            data[x_labels_indexes[x_label]] += stat['count']

        series = {}

        for loop_label, y_data in loop_data.items():

            for how, data in y_data.items():

                y_label = _(login_method_label(how)) if how else _('none')

                data = [count or None for count in data]

                series[loop_label] = {'label': y_label, 'data': data}

        return {

            'x_labels': [label.isoformat() for label in x_labels],

            'series': series,

        }

    @classmethod

    def get_service_statistics(cls, group_by_time='timestamp', **kwargs):

        stats = cls.get_statistics(group_by_time=group_by_time, group_by_field='how', group_by_references=True, **kwargs)

        def get_reference_label(service_pk):

            return str(Service.objects.get(pk=service_pk))

        return cls._get_method_statistics_with_reference(group_by_time, get_reference_label, **kwargs)

    @classmethod

    def get_service_ou_statistics(cls, group_by_time='timestamp', **kwargs):

        def get_reference_label(service_pk):

            service = Service.objects.get(pk=service_pk)

            return str(service.ou)

        return cls._get_method_statistics_with_reference(group_by_time, get_reference_label, **kwargs)

class UserLogin(EventTypeWithMethod):

class UserRegistration(EventTypeWithMethod):

class UserServiceSSO(EventTypeWithMethod):

@pytest.mark.parametrize('event_type_name', ['user.login', 'user.registration'])

def test_statistics(db, event_type_name, freezer):

    from authentic2.a2_rbac.utils import get_default_ou

    from authentic2.a2_rbac.models import OrganizationalUnit as OU

    user = User.objects.create(username='john.doe', email='john.doe@example.com')

    user2 = User.objects.create(username='jane.doe', email='jane.doe@example.com')

    ou = OU.objects.create(name='Second OU')

    portal = Service.objects.create(name='portal', slug='portal', ou=ou)

    agendas = Service.objects.create(name='agendas', slug='agendas', ou=get_default_ou())

    forms = Service.objects.create(name='forms', slug='forms', ou=get_default_ou())

    method = {'how': 'password-on-https'}

    method2 = {'how': 'fc'}

    event_type = EventType.objects.get_for_name(event_type_name)

    freezer.move_to('2020-02-03 12:00')

    event = Event.objects.create(type=event_type, references=[user, portal], user=user, data=method)

    event = Event.objects.create(type=event_type, references=[user2, portal], user=user2, data=method)

    freezer.move_to('2020-02-03 13:00')

    event = Event.objects.create(type=event_type, references=[user, portal], user=user, data=method2)

    event = Event.objects.create(type=event_type, references=[user2, portal], user=user2, data=method2)

    freezer.move_to('2020-03-03 12:00')

    event = Event.objects.create(type=event_type, references=[user, portal], user=user, data=method)

    event = Event.objects.create(type=event_type, references=[user, agendas], user=user, data=method)

    event = Event.objects.create(type=event_type, references=[user, forms], user=user, data=method)

    event = Event.objects.create(type=event_type, user=user)

    event_type_definition = event_type.definition

    stats = event_type_definition.get_method_statistics()

    assert stats == {

        'x_labels': ['2020-02-03T12:00:00+00:00', '2020-02-03T13:00:00+00:00', '2020-03-03T12:00:00+00:00'],

        'series': [

            {'label': 'password', 'data': [2, None, 3]},

            {'label': 'FranceConnect', 'data': [None, 2, None]},

            {'label': 'none', 'data': [None, None, 1]},

        ],

    }

    stats = event_type_definition.get_method_statistics('month')

    assert stats == {

        'x_labels': ['2020-02-01T00:00:00+01:00', '2020-03-01T00:00:00+01:00'],

        'series': [

            {'label': 'FranceConnect', 'data': [2, None]},

            {'label': 'password', 'data': [2, 3]},

            {'label': 'none', 'data': [None, 1]},

        ],

    }

    stats = event_type_definition.get_method_statistics('year')

    assert stats == {

        'x_labels': ['2020-01-01T00:00:00+01:00'],

        'series': [

            {'label': 'FranceConnect', 'data': [2]},

            {'label': 'password', 'data': [5]},

            {'label': 'none', 'data': [1]},

        ],

    }

    stats = event_type_definition.get_service_statistics('month')

    assert stats == {

        'x_labels': ['2020-02-01T00:00:00+01:00', '2020-03-01T00:00:00+01:00'],

        'series': {

            'portal': {'label': 'password', 'data': [2, 1]},

            'agendas': {'label': 'password', 'data': [None, 1]},

            'forms': {'label': 'password', 'data': [None, 1]},

            'None': {'label': 'none', 'data': [None, 1]},

        },

    }

    stats = event_type_definition.get_service_ou_statistics('month')

    assert stats == {

        'x_labels': ['2020-02-01T00:00:00+01:00', '2020-03-01T00:00:00+01:00'],

        'series': {

            'Second OU': {'label': 'password', 'data': [2, 1]},

            'Default organizational unit': {'label': 'password', 'data': [None, 2]},

            'None': {'label': 'none', 'data': [None, 1]},

        },

    }