Produits Entr'ouvert » Authentic 2

            op = MANAGE_MEMBERS_OP

        Permission = rbac_utils.get_permission_model()

        operation = rbac_utils.get_operation(op)

            operation=operation,

            target_ct=ContentType.objects.get_for_model(self),

            target_id=self.pk,

            op = MANAGE_MEMBERS_OP

        Permission = rbac_utils.get_permission_model()

        operation = rbac_utils.get_operation(op)

            operation=operation, target_ct=ContentType.objects.get_for_model(self), target_id=self.pk

        )

        self.permissions.through.objects.get_or_create(role=self, permission=self_perm)

        if OrganizationalUnit.objects.exists():

            return

        # Create a default OU if none exists currently

            slug='default',

            defaults={

                'default': True,

def get_view_user_perm(ou=None):

    User = get_user_model()

    Permission = rbac_utils.get_permission_model()

        operation=rbac_utils.get_operation(VIEW_OP),

        target_ct=ContentType.objects.get_for_model(ContentType),

        target_id=ContentType.objects.get_for_model(User).pk,

def get_search_ou_perm(ou=None):

    if ou:

        Permission = rbac_utils.get_permission_model()

            operation=rbac_utils.get_operation(SEARCH_OP),

            target_ct=ContentType.objects.get_for_model(ou),

            target_id=ou.pk,

    else:

        OU = rbac_utils.get_ou_model()

        Permission = rbac_utils.get_permission_model()

            operation=rbac_utils.get_operation(SEARCH_OP),

            target_ct=ContentType.objects.get_for_model(ContentType),

            target_id=ContentType.objects.get_for_model(OU).pk,

def get_manage_authorizations_user_perm(ou=None):

    User = get_user_model()

    Permission = rbac_utils.get_permission_model()

        operation=rbac_utils.get_operation(models.MANAGE_AUTHORIZATIONS_OP),

        target_ct=ContentType.objects.get_for_model(ContentType),

        target_id=ContentType.objects.get_for_model(User).pk,

        for authenticator in self.get_authenticators():

            if hasattr(authenticator, 'authenticate_credentials'):

                try:

                        username, password, request=request

                    )

                    result['result'] = 1

@GlobalCache

def event_type_cache(name):

    return event_type

                else:

                    yield None

            count += 1

            yield None

    class Meta:

            .annotate(count=Count('ou'))

            .order_by('-count')

        )

            if not ou_id:

                continue

            service_ou_ids.append(ou_id)

            return

        for conn in cls.get_connections(block):

            existing_groups = cls.get_groups_dns(conn, block)

                if group_dn in existing_groups:

                    continue

                log.warning('ldap: unknown group "%s" mapped to a role', group_dn)

                            else:

                                log.warning(

                                    'could not rebind after a bind failure, unable to attach the error to the'

                                )

                            user_login_failure(authz_id)

                    else:

        if create is None:

            create = block['create_group']

        if create:

            return group

        else:

            try:

                attributes.add(block[field])

        for external_id_tuple in map_text(block['external_id_tuples']):

            attributes.update(cls.attribute_name_from_external_id_tuple(external_id_tuple))

            attributes.add(to_at)

        for mapping in block['user_attributes']:

            from_ldap = mapping.get('from_ldap')

                user._changed = False

            external_id = self.build_external_id(map_text(block['external_id_tuples'][0]), attributes)

            if external_id:

                    user=user, external_id=external_id, source=force_text(block['realm'])

                )

                if block['clean_external_id_on_update']:

        while first_pass or pg_ctrl.cookie:

            first_pass = False

            msgid = conn.search_ext(*args, serverctrls=[pg_ctrl], **kwargs)

            pg_ctrl.cookie = serverctrls[0].cookie

            yield from cls.normalize_ldap_results(data)

                context = ssl.create_default_context()

                try:

                    with socket.create_connection((hostname, port), timeout=2) as sock:

                            pass

                except (socket.herror, socket.gaierror) as e:

                    return False, 'socket address error on host %s: %s' % (hostname, e)

            queries.append(models.Q(**{username_field: username}))

            if '@' not in username:

                if app_settings.REALMS:

                        queries.append(models.Q(**{username_field: upn(username, realm)}))

        else:

            queries.append(models.Q(**{username_field: upn(username, realm)}))

    def clean_password_hash(self):

        password_hash = self.cleaned_data['password_hash']

        try:

        except ValueError:

            raise ValidationError(_('Invalid password format or unknown hashing algorithm.'))

        return password_hash

                    pass

                if not header.field:

                    try:

                        header.attribute = True

                    except Attribute.DoesNotExist:

                        pass

        content_type = ContentType.objects.get_for_model(User)

        attrs = {}

            name='first_name',

            defaults={

                'kind': 'string',

                'user_visible': True,

            },

        )

            name='last_name',

            defaults={

                'kind': 'string',

        serialize = get_kind('string').get('serialize')

        for user in User.objects.all():

            for attr_name, value in attrs.items():

                    content_type=content_type,

                    object_id=user.id,

                    attribute=value,

        assert old_user.id == new_user.id

        for group in old_user.groups.all():

            new_groups.append(GroupThrough(user_id=new_user.id, group_id=group.id))

            new_permissions.append(PermissionThrough(user_id=new_user.id, group_id=group.id))

    # mass create group and permission relations

    GroupThrough.objects.bulk_create(new_groups)

        if commit and app_settings.A2_REGISTRATION_GROUPS:

            groups = []

            for name in app_settings.A2_REGISTRATION_GROUPS:

                groups.append(group)

            user.groups = groups

        return user

        return '%s$%s$%s$%s' % (self.algorithm, count, salt, h)

    def to_drupal(self, encoded):

        count = self.i64toa(math.ceil(math.log(int(count), 2)))

        return '$S$%s%s%s' % (count, salt, h)

        assert salt and '$' not in salt

        h = salt.encode()

        password = password.encode()

            h = self.digest(h + password).digest()

        return "%s$%d$%s$%s" % (self.algorithm, iterations, salt, self.b64encode(h)[:43])

    def verify(self, password, encoded):

        assert algorithm == self.algorithm

        encoded_2 = self.encode(password, salt, int(iterations))

        return constant_time_compare(encoded, encoded_2)

        return "%s$%s$%s" % (self.algorithm, salt, hash)

    def verify(self, password, encoded):

        assert algorithm == self.algorithm

        encoded_2 = self.encode(password, salt)

        return constant_time_compare(encoded, encoded_2)

        return "%s$md5$%s$%s" % (self.algorithm, salt, hash)

    def verify(self, password, encoded):

        salt = unhexlify(salt)

        if algorithm != self.algorithm:

            raise ValueError('not a joomla encoded password')

                with tenant_context(tenant):

                    self.tenant_shown = False

                    self.output = False

                        self.check_and_repair(options)

                    if self.fake and self.output:

                        self.success('Faked!')

        if targets_with_duplicates:

            self.warning('Found %d manage members permissions with duplicates', len(targets_with_duplicates))

            if self.repair:

                    qs = Permission.objects.filter(target_ct_id=target_ct_id, target_id=target_id)

                    for perm in qs:

                        linked_admin_role = Role.objects.get(

                    self.notice(' - %s: [%s]', target, '; '.join(map(str, qs)))

            if self.do_repair():

                self.notice('Deleting duplicate manage members permissions...', ending=' ')

                    qs = Permission.objects.filter(target_ct_id=target_ct_id, target_id=target_id).order_by(

                        'id'

                    )

        elif verbosity == 3:

            ldap_logger.setLevel(logging.DEBUG)

            continue

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

        super().__init__(*args, **kwargs)

        if self.request and not self.request.user.is_anonymous:

                qs = getattr(field, 'queryset', None)

                if not qs:

                    continue

            action = form.cleaned_data.get('action')

            Permission = get_permission_model()

            if action == 'add' and operation and target:

                    operation=operation,

                    ou=ou,

                    target_ct=ContentType.objects.get_for_model(target),

            qs = OIDCAuthorization.objects.filter(user=self.get_object())

            qs = qs.filter(id=auth_id.pk)

            oidc_authorization = qs.first()

            if count:

                self.request.journal.record(

                    'manager.user.sso.authorization.deletion',

def create_trigger(apps, schema_editor):

    with schema_editor.connection.cursor() as cursor:

        cursor.execute('SHOW default_text_search_config')

        cursor.execute(

            '''CREATE OR REPLACE FUNCTION authentic2_update_atv_search_vector() RETURNS TRIGGER AS $$

BEGIN

    def database_backwards(self, app_label, schema_editor, from_state, to_state):

        if not self.allowed(app_label, schema_editor, to_state):

            return

            schema_editor.execute('DROP INDEX IF EXISTS "%s_%s"' % (self.index_name, i))

    def describe(self):

        raise ServiceAccessDenied(service=self)

    def add_authorized_role(self, role):

        return authorization

    def remove_authorized_role(self, role):

def get_definition_from_oid(oid):

        if value['oid'] == oid:

            return value

def get_definition_from_alias(alias):

        if 'alias' in value and alias in value['alias']:

            return value

            is_required = ra.get(IS_REQUIRED, 'false') == 'true'

            if name_format != lasso.SAML2_ATTRIBUTE_NAME_FORMAT_URI:

                continue

            if def_name is None:

                warnings.warn(

                    'attribute %s/%s unsupported on service provider %s'

    def cleanup(self):

        for federation in self.filter(user__isnull=True):

            results = federation_delete.send_robust(sender=federation)

                if not result:

                    return

            federation.delete()

    composition = ((2, '23456789'), (6, 'ABCDEFGHJKLMNPQRSTUVWXYZ'), (1, '%$/\\#@!'))

    parts = []

    for cnt, alphabet in composition:

            parts.append(random.SystemRandom().choice(alphabet))

    random.shuffle(parts, random.SystemRandom().random)

    return ''.join(parts)

        return []

    values = []

        try:

            values.append(int(v))

        except ValueError:

    @classmethod

    def can_edit_profile(cls):

        return bool(fields) and app_settings.A2_PROFILE_CAN_EDIT_PROFILE

    @classmethod

        fragments = []

        oidc_sessions = get_oidc_sessions(request)

            if 'frontchannel_logout_uri' not in value:

                continue

            ctx = {

        if state:

            post_logout_redirect_uri = make_url(post_logout_redirect_uri, params={'state': state})

    # FIXME: do something with id_token_hint

    return a2_logout(request, next_url=post_logout_redirect_uri, do_local=False, check_referer=False)

                    q.append(Q(ou_id=int(key[3:])))

                    continue

            elif perm_or_perms & value:

                q.append(Q(pk=int(fk)))

        if q:

            return functools.reduce(Q.__or__, q)

def get_operation(operation_tpl):

    from . import models

    return operation

def test_registration_page(settings, app, franceconnect, hooks):

    assert User.objects.count() == 0

    # a new user has been created

    assert User.objects.count() == 1

def create_user(**kwargs):

    User = get_user_model()

    password = kwargs.pop('password', None) or kwargs['username']

    if password:

        user.clear_password = password

        user.set_password(password)

    response = app.post(token_url, params=params)

    assert 'id_token' in response.json

    token = response.json['id_token']

    jwt = JWT()

    # jwt deserialization implicitly checks the token signature:

    jwt.deserialize(token, key=jwk)

    response = app.post(token_url, params=params, headers=client_authentication_headers(oidc_client))

    assert 'id_token' in response.json

    token = response.json['id_token']

    jwt = JWT()

    # jwt deserialization implicitly checks the token signature:

    jwt.deserialize(token, key=jwk)

        'username': simple_user.username,

        'password': simple_user.username,

    }

        response = app.post(token_url, params=params, status=400)

        assert response.json['error'] == 'invalid_client'

        assert 'Wrong client secret' in response.json['error_description']

        'username': simple_user.username,

        'password': simple_user.username,

    }

        app.post(token_url, params=params)

    response = app.post(token_url, params=params, status=400)

    assert response.json['error'] == 'invalid_client'

def test_update_self_admin_perm_migration(migration, new_perm_exists):

    old_apps = migration.before([('a2_rbac', '0022_auto_20200402_1101')])

    Role = old_apps.get_model('a2_rbac', 'Role')

    Permission = old_apps.get_model('a2_rbac', 'Permission')

    Operation = old_apps.get_model('django_rbac', 'Operation')

    ContentType = old_apps.get_model('contenttypes', 'ContentType')

        # with cache the same value will come back

        g = GlobalCache(f, hostname_vary=False)

        values = set()

            values.add(g())

        self.assertEqual(len(values), 1)

        # with and hostname vary 10 values will come back

        g = GlobalCache(f, hostname_vary=True)

        values = set()

            values.add(g())

        self.assertEqual(len(values), 10)

        # null timeout, no cache

        'last_name': 'Doeny',

    }

    headers = basic_authorization_header(admin)

def test_api_users_create_with_email_verified(settings, app, admin):

        'email': 'john.doe@nowhere.null',

    }

    headers = basic_authorization_header(admin)

    user = User.objects.get(id=simple_user.id)

    assert not user.email_verified

def test_api_users_boolean_attribute(app, superuser):

    superuser.attributes.boolean = True

    app.authorization = ('Basic', (superuser.username, superuser.username))

    resp = app.get('/api/users/%s/' % superuser.uuid)

def test_api_users_boolean_attribute_optional(app, superuser):

    superuser.attributes.boolean = True

    app.authorization = ('Basic', (superuser.username, superuser.username))

    resp = app.get('/api/users/%s/' % superuser.uuid)

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

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

    role1 = Role.objects.create(name='role1')

    role2 = Role.objects.create(name='role2')

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

    service1.add_authorized_role(role1)

    resp = app.get('/api/users/')

    assert len(resp.json['results']) == 4

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

    service1.add_authorized_role(role1)

    for user in superuser, user1, user2, user3:

        simple_role.members.add(user)

    if not api_user.has_perm('a2_rbac.manage_members_role', role):

        status = 403

    if api_user.has_perm('a2_rbac.manage_members_role', role):

        assert len(role.members.all()) == 0

    else:

def test_user_synchronization(app, simple_user):

    headers = basic_authorization_header(simple_user)

    uuids = []

        user = User.objects.create(first_name='ben', last_name='dauve')

        uuids.append(user.uuid)

    unknown_uuids = [uuid.uuid4().hex for i in range(100)]

def test_no_opened_session_cookie_on_api(app, user, settings):

    settings.A2_OPENED_SESSION_COOKIE_DOMAIN = 'testserver.local'

    app.authorization = ('Basic', (user.username, user.username))

    assert 'A2_OPENED_SESSION' not in app.cookies

        'last_name': 'Doe',

    }

    headers = basic_authorization_header(admin)

    payload['prefered_color'] = None

    payload['date'] = None

    payload['birthdate'] = None

    payload['prefered_color'] = ''

    payload['date'] = ''

    payload['birthdate'] = ''

    payload['prefered_color'] = None

    payload['date'] = None

    payload['birthdate'] = None

    app.authorization = ('Basic', (admin.username, admin.username))

    user = User.objects.create(first_name='Jean', last_name='Dupont', ou=get_default_ou())

    params = {

        'first_name': 'Jeanne',

        'new_password': 'password2',

    }

    url = reverse('a2-api-password-change')

    user2.delete()

    assert User.objects.get(username='john.doe').check_password('password2')

def test_api_users_delete(settings, app, admin, simple_user):

    headers = basic_authorization_header(admin)

    assert not User.objects.filter(pk=simple_user.pk).exists()

    assert_event('manager.user.deletion', user=admin, api=True)

    assert login_stats in resp.json['data']

    # same goes with users

    login_stats['filters'].append(

        {

            'id': 'users_ou',

    event_type = EventType.objects.get_for_name(event_type_name)

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

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

    resp = app.get('/api/statistics/%s/?time_interval=month' % event_name, headers=headers)

    data = resp.json['data']

def test_api_statistics_no_crash_older_drf(app, admin):

    headers = basic_authorization_header(admin)

    expected_status = 200 if drf_version > '3.9' else 404

def test_find_duplicates_put(app, admin, settings):

        'last_name': 'Doe',

        'birthdate': '2017-12-31',

    }

    assert qs.get().attributes.birthdate == datetime.date(2017, 12, 31)

    qs.delete()

    default_ou = get_default_ou()

    admin_op = get_operation(ADMIN_OP)

    role1 = Role.objects.create(name='Role 1', slug='role-1', ou=default_ou)

    perm1 = Permission.objects.create(

        operation=admin_op,

    ou.email_is_unique = True

    ou.save()

    settings.A2_EMAIL_IS_UNIQUE = True

    settings.A2_USERNAME_IS_UNIQUE = True

    AttributeValue.objects.all().delete()

        def map_threads(f, l):

            threads = []

def test_idempotency():

        s = force_bytes(str(random.getrandbits(1024)))

        assert crypto.aes_base64_decrypt(key, crypto.aes_base64_encrypt(key, s)) == s

    thomas.roles.remove(role)

    content_only_key = '''email key,_role_name

tnoel@entrouvert.com,test_name'''

    assert importer.run()

    thomas.refresh_from_db()

    assert thomas in role.members.all()

        },

        ImportContext(),

    )

    assert role.ou == ou

        {'uuid': uuid, 'name': 'some-role', 'slug': 'some-role', 'ou': {'slug': 'ou-2'}, 'service': None},

        ImportContext(),

    )

    existing_role.refresh_from_db()

    assert existing_role.ou == ou2

    assert Role.objects.exclude(slug__startswith='_').count() == 1

        {'uuid': uuid, 'slug': 'some-role', 'name': 'some role changed', 'ou': None, 'service': None},

        ImportContext(),

    )

    existing_role.refresh_from_db()

    assert role == existing_role

    assert existing_role.name == 'some role changed'

        ImportContext(),

    )

    role, status = rd.deserialize()

    assert status == 'created'

    assert role.attributes.count() == 2

    assert len(created) == 2

    rd = RoleDeserializer(child_role_dict, ImportContext())

    child_role, status = rd.deserialize()

    assert status == 'created'

    assert len(created) == 1

    import_context = ImportContext()

    rd = RoleDeserializer(some_role_dict, import_context)

    rd.deserialize()

    assert len(perm_created) == 1

    perm = perm_created[0]

    assert perm.target == perm_role

    import_context = ImportContext()

    rd = RoleDeserializer(some_role_dict, import_context)

    rd.deserialize()

    assert len(perm_created) == 1

    perm = perm_created[0]

    assert perm.target.app_label == 'admin'

def test_roles_import_no_slug(db):

    roles = [{'name': 'some role', 'description': 'some role description', 'ou': None}]

    role = Role.objects.get(name='some role')

    assert role.slug == 'some-role'

        response = client.get('/idp/cas/login', {constants.SERVICE_PARAM: self.URL})

        location = response['Location']

        query = urllib.parse.parse_qs(location.split('?')[1])

        next_url_query = urllib.parse.parse_qs(next_url_query)

        response = client.get(location)

        response = client.post(

        response = client.get('/idp/cas/login', {constants.SERVICE_PARAM: self.URL})

        location = response['Location']

        query = urllib.parse.parse_qs(location.split('?')[1])

        next_url_query = urllib.parse.parse_qs(next_url_query)

        response = client.get(location)

        response = client.post(

uGnhj8v6XwvbjKZrL9kA+xf8ziazZfvvw/VGTm+IVFYB7d1x457jY5zjjXJvNyso

owIDAQAB

-----END PUBLIC KEY-----'''

def test_null_character_nonce(app, db):

    monkeypatch.setattr(authentic2.management.commands.export_site, 'export_site', dummy_export_site)

    management.call_command('export_site')

    assert json.loads(out) == dummy_export_site()

    management.call_command('import_site', json_fixture(content))

    assert "Real run" in out

    assert "1 roles created" in out

    assert "0 roles updated" in out

def test_import_site_update_roles(db, json_fixture):

    r1 = Role.objects.create(name='Role1', slug='role1')

    content = {

        'roles': [

    assert EventTypeDefinition.get_for_name('user.sso') is SSO

    with pytest.raises(AssertionError, match='already registered'):

        class SSO2(UserEventTypes):

            name = 'user.sso'

            label = 'Single Sign On'

    for name in 'abcdef':

        event_types.append(EventType.objects.create(name=name))

        events.append(

            Event(

                type=random.choice(event_types),

    journal.record('user.registration', how='fc')

    journal.record('user.logout')

        journal.record('user.login', how='fc')

        journal.record('user.logout')

    assert stats == {'series': [], 'x_labels': []}

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

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

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

    stats = event_type_definition.get_method_statistics('timestamp')

    stats['series'].sort(key=lambda x: x['label'])

def test_statistics_fill_date_gaps(db, freezer):

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

    event_type = EventType.objects.get_for_name('user.login')

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

    freezer.move_to('2021-01-02 13:00')

    event_type_definition = event_type.definition

    Event.objects.all().delete()

    freezer.move_to('2020-11-29 12:00')

    freezer.move_to('2022-02-02 13:00')

    stats = event_type_definition.get_method_statistics('month')

    assert stats == {

        'x_labels': ['2020-11', '2020-12'] + ['2021-%02d' % i for i in range(1, 13)] + ['2022-01', '2022-02'],

    Event.objects.all().delete()

    freezer.move_to('2020-11-29 12:00')

    freezer.move_to('2025-02-02 13:00')

    stats = event_type_definition.get_method_statistics('year')

    assert stats == {

        'x_labels': ['2020', '2021', '2022', '2023', '2024', '2025'],

    event_type_definition = event_type.definition

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

    stats = event_type_definition.get_service_statistics('month')

    stats['series'].sort(key=lambda x: x['label'])

    event_type = EventType.objects.get_for_name('user.login')

    event_type_definition = event_type.definition

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

    stats = event_type_definition.get_method_statistics('month', services_ou=ou_with_no_service)

def test_prefetcher(db):

    event_type = EventType.objects.get_for_name('user.login')

        user = User.objects.create()

        Event.objects.create(type=event_type, user=user, references=[user])

        Event.objects.create(type=event_type, user=user, references=[user])

    fake = faker.Faker(locale='fr_FR')

    faker.Faker.seed(0)

        User.objects.bulk_create(

            User(

                first_name=fake.first_name() + ' ' + fake.first_name(),

        'last_name': user.last_name,

    }

        resp = app.get('/api/users/find_duplicates/', params=params)

        assert len(resp.json['data']) >= 1

        'birthdate': str(user.attributes.birthdate),

    }

        resp = app.get('/api/users/find_duplicates/', params=params)

        assert len(resp.json['data']) >= 1

        }

    ]

    with utils.check_log(caplog, "account name authentication filter doesn't contain '%s'"):

            '/login/', {'login-password-submit': '1', 'username': USERNAME, 'password': PASS}, follow=True

        )

    assert len(mailoutbox) == 0

    assert User.objects.filter(ou_id=new_ou.id).count() == 0

        ou_add = login(app, superuser_or_admin, url)

        form = ou_add.form

        form.set('first_name', 'John')

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 4

            assert not checked or key == 'all'

        assert 'all' in [o[0] for o in options]

        assert 'none' in [o[0] for o in options]

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 4

            assert not checked or key == str(get_default_ou().pk)

        q = response.pyquery.remove_namespaces()

        assert len(q('table tbody tr')) == 1

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 4

            if key == 'all':

                assert checked

            else:

        options = field.options

        assert len(options) == 1

        # ou1 is selected

        assert checked

        assert key == str(ou1.pk)

        # verify table shown

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 1

        assert checked

        assert key == str(ou1.pk)

        q = response.pyquery.remove_namespaces()

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 1

        assert checked

        assert key == str(ou1.pk)

        q = response.pyquery.remove_namespaces()

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 1

        assert checked

        assert key == str(ou1.pk)

        q = response.pyquery.remove_namespaces()

        field = response.form['search-ou']

        options = field.options

        assert len(options) == 1

        assert checked

        assert key == str(ou1.pk)

        q = response.pyquery.remove_namespaces()

def test_manager_role_inheritance_list_search_permission(app, admin, simple_user, simple_role, ou1):

    visible_role = Role.objects.create(name='visible_role', ou=simple_user.ou)

    visible_role_2 = Role.objects.create(name='visible_role_2', ou=ou1)

    admin_of_simple_role = simple_role.get_admin_role()

    admin_of_simple_role.members.add(simple_user)

    assert Role.objects.filter(name=role_ou2.name, ou=get_default_ou()).exists()

    response.form.set('search-text', 'role_ou1')

    export_response = response.click('Export')

    new_export = export_response.json

    resp.form.set('username', simple_user.username)

    resp.form.set('password', simple_user.username)

    resp = app.get(reverse('account_management'))

    assert resp.pyquery('body').attr('data-service-slug') == service.slug

    users = [User(username='user%s' % i) for i in range(10)]

    User.objects.bulk_create(users)

    resp = app.get('/manage/users/?search-text=user1')

    resp = resp.click('CSV').follow()

    resp = resp.click('Download CSV')

def assert_redirects_complex(response, expected_url, **kwargs):

    assert response.status_code == 302, 'code should be 302'

    e_scheme, e_netloc, e_path, e_query, e_fragment = urllib.parse.urlsplit(expected_url)

    e_scheme = e_scheme if e_scheme else scheme

    e_netloc = e_netloc if e_netloc else netloc

    current_run_on_commit = connection.run_on_commit

    connection.run_on_commit = []

    while current_run_on_commit:

        func()