misc-vdeniaud/scripts/doublons-cd91/test_authentic_fusion.py

import datetime

import pytest
from authentic2.a2_rbac.models import Role
from django.contrib.auth import get_user_model
from django.utils.timezone import now
from mellon.models import Issuer, UserSAMLIdentifier

from .utils import call_command

User = get_user_model()


@pytest.mark.freeze_time('2022-04-19 14:00')
def test_authentic_fusion(db, caplog):
    role1 = Role.objects.create(name='role1')
    role2 = Role.objects.create(name='role2')
    role3 = Role.objects.create(name='role3')

    # duplicated users, but not agents
    User.objects.create(first_name='Normal', last_name='User', email='normal.user@gmail.com')
    User.objects.create(first_name='Normal', last_name='User', email='normal.user@gmail.com')

    # duplicated users, agents but no saml link
    User.objects.create(first_name='Agent', last_name='No SAML', email='agent.no.saml@cd-essonne.fr')
    User.objects.create(first_name='Agent', last_name='No SAML', email='agent.no.saml@cd-essonne.fr')

    # agent with saml link, no duplicate
    issuer = Issuer.objects.create(entity_id='https://idp1.example.com/', slug='idp1')
    saml_user = User.objects.create(
        first_name='Agent',
        last_name='No duplicate',
        email='agent.no.dup@cd-essonne.fr',
    )
    deactivated_saml_user = User.objects.create(
        first_name='Agent',
        last_name='No duplicate (deactivated)',
        email='agent.no.dup@cd-essonne.fr',
        is_active=False,
    )
    UserSAMLIdentifier.objects.create(user=saml_user, issuer=issuer, name_id='anodup')
    UserSAMLIdentifier.objects.create(user=deactivated_saml_user, issuer=issuer, name_id='adeact')

    # duplicated users, agents with saml link
    saml_user_old_connection = User.objects.create(
        id=42,
        uuid='uuid:42/agent@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated',
        email='agent@cd-essonne.fr',
        last_login=now() - datetime.timedelta(days=10),
    )
    UserSAMLIdentifier.objects.create(user=saml_user_old_connection, issuer=issuer, name_id='aduplicated')
    saml_user_old_connection.roles.add(role1)
    saml_user_old_connection.roles.add(role2)

    saml_user_old_connection_no_roles = User.objects.create(
        id=43,
        uuid='uuid:43/agent@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated',
        email='agent@cd-Essonne.fr',
        last_login=now() - datetime.timedelta(days=5),
    )
    UserSAMLIdentifier.objects.create(
        user=saml_user_old_connection_no_roles, issuer=issuer, name_id='Aduplicated'
    )

    saml_user_recent_connection = User.objects.create(
        id=44,
        uuid='uuid:44/agent@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated',
        email='Agent@cd-essonne.fr',
        last_login=now() - datetime.timedelta(days=1),
    )
    UserSAMLIdentifier.objects.create(user=saml_user_recent_connection, issuer=issuer, name_id='ADUPLICATED')
    saml_user_old_connection.roles.add(role1)
    saml_user_old_connection.roles.add(role3)

    # again duplicated users, agents with saml link
    saml_user_recent_connection_no_roles_2 = User.objects.create(
        id=45,
        uuid='uuid:45/agent2@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated 2',
        email='agent2@cd-essonne.fr',
        last_login=now() - datetime.timedelta(days=5),
    )
    UserSAMLIdentifier.objects.create(
        user=saml_user_recent_connection_no_roles_2, issuer=issuer, name_id='Aduplicated2'
    )

    saml_user_old_connection_2 = User.objects.create(
        id=46,
        uuid='uuid:46/agent2@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated 2',
        email='agent2@cd-essonne.fr',
        last_login=now() - datetime.timedelta(days=10),
    )
    UserSAMLIdentifier.objects.create(user=saml_user_old_connection_2, issuer=issuer, name_id='aduplicated2')
    saml_user_old_connection_2.roles.add(role1)
    saml_user_old_connection_2.roles.add(role2)

    assert User.objects.count() == 11
    assert User.objects.filter(is_active=True).count() == 10

    call_command('runscript', 'tests/authentic_fusion.py')

    log_messages = caplog.messages
    assert log_messages == [
        '=== Starting fusion at 2022-04-19T14:00:00 ===',
        '* Processing user Agent Duplicated Agent@cd-essonne.fr uuid:44/agent@cd-essonne.fr /manage/users/44/',
        'Adding role role1',
        'Adding role role2',
        'Adding role role3',
        'Disabling duplicate Agent Duplicated agent@cd-Essonne.fr uuid:43/agent@cd-essonne.fr /manage/users/43/',
        'Disabling duplicate Agent Duplicated agent@cd-essonne.fr uuid:42/agent@cd-essonne.fr /manage/users/42/',
        '* Processing user Agent Duplicated 2 agent2@cd-essonne.fr uuid:45/agent2@cd-essonne.fr /manage/users/45/',
        'Adding role role1',
        'Adding role role2',
        'Disabling duplicate Agent Duplicated 2 agent2@cd-essonne.fr uuid:46/agent2@cd-essonne.fr /manage/users/46/',
        'Result {"uuid:44/agent@cd-essonne.fr": ["uuid:43/agent@cd-essonne.fr", "uuid:42/agent@cd-essonne.fr"], "uuid:45/agent2@cd-essonne.fr": ["uuid:46/agent2@cd-essonne.fr"]}',
        '=== Did nothing ===',
    ]

    # no changes in db
    assert User.objects.count() == 11
    assert User.objects.filter(is_active=True).count() == 10
    assert saml_user_recent_connection_no_roles_2.roles.count() == 0

    caplog.clear()
    call_command('runscript', 'tests/authentic_fusion.py', '--proceed=true')

    assert log_messages[:-1] == caplog.messages[:-1]
    assert caplog.messages[-1] == '=== Success ==='

    assert User.objects.count() == 11
    assert User.objects.filter(is_active=True).count() == 7

    assert User.objects.filter(email='normal.user@gmail.com', is_active=True).count() == 2
    assert User.objects.filter(email='agent.no.saml@cd-essonne.fr', is_active=True).count() == 2
    assert User.objects.filter(email='agent.no.dup@cd-essonne.fr', is_active=True).count() == 1

    saml_user_old_connection.refresh_from_db()
    assert saml_user_old_connection.is_active is False

    saml_user_old_connection_no_roles.refresh_from_db()
    assert saml_user_old_connection_no_roles.is_active is False

    saml_user_recent_connection.refresh_from_db()
    assert saml_user_recent_connection.is_active is True
    assert set(saml_user_recent_connection.roles.all()) == {role1, role2, role3}

    saml_user_old_connection_2.refresh_from_db()
    assert saml_user_old_connection_2.is_active is False

    saml_user_recent_connection_no_roles_2.refresh_from_db()
    assert saml_user_recent_connection_no_roles_2.is_active is True
    assert set(saml_user_recent_connection_no_roles_2.roles.all()) == {role1, role2}


@pytest.mark.freeze_time('2022-04-19 14:00')
def test_authentic_fusion_local_account(db, caplog):
    role1 = Role.objects.create(name='role1')
    role2 = Role.objects.create(name='role2')
    role3 = Role.objects.create(name='role3')

    # duplicated users, but not agents, should ignore
    User.objects.create(first_name='Normal', last_name='User', email='normal.user@gmail.com')
    User.objects.create(first_name='Normal', last_name='User', email='normal.user@gmail.com')

    # two local duplicates, should ignore
    User.objects.create(first_name='Agent', last_name='No SAML', email='agent.no.saml@cd-essonne.fr')
    User.objects.create(first_name='Agent', last_name='No SAML', email='agent.no.saml@cd-essonne.fr')

    # three local duplicates, should ignore
    User.objects.create(first_name='Agent', last_name='3 dups', email='Agent3dups@cd-essonne.fr')
    User.objects.create(first_name='Agent', last_name='3 dups', email='aGent3dups@cd-essonne.fr')
    User.objects.create(first_name='Agent', last_name='3 dups', email='agEnt3dups@cd-essonne.fr')

    # agent with saml link, no duplicate, should ignore
    issuer = Issuer.objects.create(entity_id='https://idp1.example.com/', slug='idp1')
    saml_user = User.objects.create(
        first_name='Agent',
        last_name='No duplicate',
        email='agent.no.dup@cd-essonne.fr',
    )
    deactivated_saml_user = User.objects.create(
        first_name='Agent',
        last_name='No duplicate (deactivated)',
        email='agent.no.dup@cd-essonne.fr',
        is_active=False,
    )
    UserSAMLIdentifier.objects.create(user=saml_user, issuer=issuer, name_id='anodup')
    UserSAMLIdentifier.objects.create(user=deactivated_saml_user, issuer=issuer, name_id='adeact')

    # two saml duplicates, should ignore
    saml_user_old_connection = User.objects.create(
        id=42,
        uuid='uuid:42/agent@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated',
        email='agent@cd-essonne.fr',
        last_login=now() - datetime.timedelta(days=10),
    )
    UserSAMLIdentifier.objects.create(user=saml_user_old_connection, issuer=issuer, name_id='aduplicated')
    saml_user_old_connection_no_roles = User.objects.create(
        id=43,
        uuid='uuid:43/agent@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated',
        email='agent@cd-Essonne.fr',
        last_login=now() - datetime.timedelta(days=5),
    )
    UserSAMLIdentifier.objects.create(
        user=saml_user_old_connection_no_roles, issuer=issuer, name_id='Aduplicated'
    )

    # local and saml duplicates, should process
    saml_user = User.objects.create(
        id=45,
        uuid='uuid:45/agent2@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated 2',
        email='agent2@cd-essonne.fr',
    )
    UserSAMLIdentifier.objects.create(user=saml_user, issuer=issuer, name_id='Aduplicated2')

    local_user = User.objects.create(
        id=46,
        uuid='uuid:46/agent2@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated 2',
        email='agent2@cd-essonne.fr',
    )
    local_user.roles.add(role2)

    # again
    local_user2 = User.objects.create(
        id=47,
        uuid='uuid:47/agent3@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated 3',
        email='agent3@cd-essonne.fr',
    )
    local_user.roles.add(role1)

    saml_user2 = User.objects.create(
        id=48,
        uuid='uuid:48/agent3@cd-essonne.fr',
        first_name='Agent',
        last_name='Duplicated 3',
        email='agent3@cd-essonne.fr',
    )
    UserSAMLIdentifier.objects.create(user=saml_user2, issuer=issuer, name_id='Aduplicated3')
    saml_user2.roles.add(role2)

    assert User.objects.count() == 15
    assert User.objects.filter(is_active=True).count() == 14

    call_command('runscript', 'tests/authentic_fusion_local.py')

    log_messages = caplog.messages
    assert log_messages == [
        '=== Starting fusion at 2022-04-19T14:00:00 ===',
        '* SKIPPING USER Agent Duplicated agent@cd-essonne.fr '
        'uuid:42/agent@cd-essonne.fr /manage/users/42/ (duplicates are both saml '
        'accounts)',
        '* SKIPPING USER Agent No SAML agent.no.saml@cd-essonne.fr '
        '06231a68f4bc483293faad5fb3105ddd /manage/users/544/ (duplicates are both '
        'local accounts)',
        '* SKIPPING USER Agent 3 dups Agent3dups@cd-essonne.fr '
        '653c422dd6d94654a0c1bd6cd708a93c /manage/users/546/ (more than 2 duplicates)',
        '* Processing user Agent Duplicated 2 agent2@cd-essonne.fr '
        'uuid:45/agent2@cd-essonne.fr /manage/users/45/',
        'Adding role role1',
        'Adding role role2',
        'Disabling duplicate Agent Duplicated 2 agent2@cd-essonne.fr '
        'uuid:46/agent2@cd-essonne.fr /manage/users/46/',
        '* Processing user Agent Duplicated 3 agent3@cd-essonne.fr '
        'uuid:48/agent3@cd-essonne.fr /manage/users/48/',
        'Adding role role2',
        'Disabling duplicate Agent Duplicated 3 agent3@cd-essonne.fr '
        'uuid:47/agent3@cd-essonne.fr /manage/users/47/',
        'Result {"uuid:45/agent2@cd-essonne.fr": ["uuid:46/agent2@cd-essonne.fr"], '
        '"uuid:48/agent3@cd-essonne.fr": ["uuid:47/agent3@cd-essonne.fr"]}',
        '=== Did nothing ===',
    ]

    # no changes in db
    assert User.objects.count() == 15
    assert User.objects.filter(is_active=True).count() == 14
    assert saml_user_recent_connection_no_roles_2.roles.count() == 0

    caplog.clear()
    call_command('runscript', 'tests/authentic_fusion_local.py', '--proceed=true')

    assert log_messages[:-1] == caplog.messages[:-1]
    assert caplog.messages[-1] == '=== Success ==='

    assert User.objects.count() == 11
    assert User.objects.filter(is_active=True).count() == 7

    assert User.objects.filter(email='normal.user@gmail.com', is_active=True).count() == 2
    assert User.objects.filter(email='agent.no.saml@cd-essonne.fr', is_active=True).count() == 2
    assert User.objects.filter(email='agent.no.dup@cd-essonne.fr', is_active=True).count() == 1

    saml_user_old_connection.refresh_from_db()
    assert saml_user_old_connection.is_active is False

    saml_user_old_connection_no_roles.refresh_from_db()
    assert saml_user_old_connection_no_roles.is_active is False

    saml_user_recent_connection.refresh_from_db()
    assert saml_user_recent_connection.is_active is True
    assert set(saml_user_recent_connection.roles.all()) == {role1, role2, role3}

    saml_user_old_connection_2.refresh_from_db()
    assert saml_user_old_connection_2.is_active is False

    saml_user_recent_connection_no_roles_2.refresh_from_db()
    assert saml_user_recent_connection_no_roles_2.is_active is True
    assert set(saml_user_recent_connection_no_roles_2.roles.all()) == {role1, role2}