Produits Entr'ouvert » Passerelle

# Generated by Django 3.2.14 on 2022-08-02 14:37

from django.db import migrations, models

import passerelle.utils.models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='Resource',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                ('ldap_url', passerelle.utils.models.LDAPURLField(max_length=512, verbose_name='Server URL')),

                ('ldap_base_dn', models.CharField(max_length=256, verbose_name='Base DN of the directory')),

                (

                    'ldap_bind_dn',

                    models.CharField(blank=True, max_length=256, null=True, verbose_name='Bind DN'),

                ),

                (

                    'ldap_bind_password',

                    models.CharField(blank=True, max_length=128, null=True, verbose_name='Bind password'),

                ),

                (

                    'ldap_tls_cert',

                    passerelle.utils.models.BinaryFileField(

                        blank=True, max_length=32768, null=True, verbose_name='TLS client certificate'

                    ),

                ),

                (

                    'ldap_tls_key',

                    passerelle.utils.models.BinaryFileField(

                        blank=True, max_length=32768, null=True, verbose_name='TLS client key'

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_ldap_resource_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'LDAP',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2022  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import base64

import contextlib

import tempfile

import ldap

import ldap.filter

from django.core.exceptions import ValidationError

from django.db import models

from django.utils.html import format_html

from django.utils.translation import gettext_lazy as _

from OpenSSL import crypto

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

from passerelle.utils.models import BinaryFileField, LDAPURLField

from passerelle.utils.templates import render_to_string

def validate_certificate(value):

    try:

        crypto.load_certificate(crypto.FILETYPE_PEM, bytes(value))

    except Exception:

        raise ValidationError(_('Invalid private key.'))

def validate_private_key(value):

    try:

        crypto.load_privatekey(crypto.FILETYPE_PEM, bytes(value))

    except Exception:

        raise ValidationError(_('Invalid private key.'))

class Resource(BaseResource):

    ldap_url = LDAPURLField(verbose_name=_('Server URL'), max_length=512)

    ldap_base_dn = models.CharField(verbose_name=_('Base DN of the directory'), max_length=256)

    ldap_bind_dn = models.CharField(verbose_name=_('Bind DN'), max_length=256, null=True, blank=True)

    ldap_bind_password = models.CharField(

        verbose_name=_('Bind password'), max_length=128, null=True, blank=True

    )

    ldap_tls_cert = BinaryFileField(

        verbose_name=_('TLS client certificate'),

        max_length=1024 * 32,

        null=True,

        blank=True,

        validators=[validate_certificate],

    )

    ldap_tls_key = BinaryFileField(

        verbose_name=_('TLS client key'),

        max_length=1024 * 32,

        null=True,

        blank=True,

        validators=[validate_private_key],

    )

    category = _('Misc')

    class Meta:

        verbose_name = _('LDAP')

    def tls_cert(self, value):

        try:

            cert = crypto.load_certificate(crypto.FILETYPE_PEM, bytes(value))

            name = ','.join(

                '%s=%s' % (a.decode(), b.decode()) for a, b in cert.get_subject().get_components()

            )

        except Exception:

            name = ('%s bytes') % len(value)

        return format_html(

            '<a href="data:application/octet-string;base64,{}" target="_blank" download="tls.crt">{}<a/>',

            base64.b64encode(value).decode(),

            name,

        )

    def clean(self):

        if bool(self.ldap_bind_dn) != bool(self.ldap_bind_password):

            raise ValidationError('Bind DN and password must be set together.')

        if bool(self.ldap_tls_cert) != bool(self.ldap_tls_key):

            raise ValidationError('Client certificate and key must be set together.')

    def get_description_fields(self):

        fields = super().get_description_fields()

        fields = [

            (field, self.tls_cert(value) if field.name == 'ldap_tls_cert' and value else value)

            for field, value in fields

        ]

        return fields

    def check_status(self):

        with self.get_connection() as conn:

            conn.whoami_s()

    @contextlib.contextmanager

    def get_connection(self):

        with contextlib.ExitStack() as stack:

            conn = ldap.initialize(self.ldap_url)

            conn.set_option(ldap.OPT_DEBUG_LEVEL, 255)

            conn.set_option(ldap.OPT_X_TLS_REQUIRE_SAN, ldap.OPT_X_TLS_NEVER)

            conn.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)

            if self.ldap_tls_cert and self.ldap_tls_key:

                cert_tempfile = stack.enter_context(tempfile.NamedTemporaryFile())

                cert_tempfile.write(self.ldap_tls_cert)

                cert_tempfile.flush()

                key_tempfile = stack.enter_context(tempfile.NamedTemporaryFile())

                key_tempfile.write(self.ldap_tls_key)

                key_tempfile.flush()

                conn.set_option(ldap.OPT_X_TLS_CACERTFILE, cert_tempfile.name)

                conn.set_option(ldap.OPT_X_TLS_CERTFILE, cert_tempfile.name)

                conn.set_option(ldap.OPT_X_TLS_KEYFILE, key_tempfile.name)

            conn.set_option(ldap.OPT_X_TLS_NEWCTX, 0)

            if self.ldap_bind_dn:

                conn.simple_bind_s(self.ldap_bind_dn, self.ldap_bind_password or '')

            else:

                conn.simple_bind_s()

            yield conn

            conn.unbind()

    def ldap_search(self, base_dn, scope, ldap_filter, ldap_attributes, sizelimit=-1, timeout=5):

        with self.get_connection() as conn:

            message_id = conn.search_ext(

                base_dn, scope, ldap_filter, ldap_attributes, timeout=timeout, sizelimit=sizelimit

            )

            while True:

                try:

                    result_type, entries = conn.result(message_id, all=0)

                except ldap.SIZELIMIT_EXCEEDED:

                    break

                if not entries:

                    break

                for dn, attributes in entries:

                    if dn:

                        decoded_attributes = cidict()

                        # decode values to unicode, if possible, and keep only the first value

                        for k, values in attributes.items():

                            decoded_values = []

                            for value in values:

                                try:

                                    decoded_values.append(value.decode())

                                except UnicodeDecodeError:

                                    pass

                            if decoded_values:

                                if len(decoded_values) == 1:

                                    decoded_attributes[k] = decoded_values[0]

                                else:

                                    decoded_attributes[k] = decoded_values

                        yield dn, decoded_attributes

    @endpoint(

        description=_('Search'),

        perm='can_access',

        parameters={

            'ldap_base_dn': {

                'description': _('Base DN for the LDAP search'),

                'example_value': 'dc=company,dc=com',

            },

            'search_attribute': {

                'description': _('Attribute to search for the substring search'),

                'example_value': 'cn',

            },

            'id_attribute': {

                'description': _('Attribute used as a unique identifier'),

                'example_value': 'uid',

            },

            'text_template': {

                'description': _(

                    'Optional template string based on LDAP attributes '

                    'to create a text value, if none given the search_attribute is used'

                ),

                'example_value': '{{ givenName }} {{ surname }}',

            },

            'ldap_attributes': {

                'description': _('Space separated list of LDAP attributes to retrieve'),

                'example_value': 'l sn givenName locality',

            },

            'id': {

                'description': _('Identifier for exacte retrieval, using the id_attribute'),

                'example_value': 'johndoe',

            },

            'q': {

                'description': _('Substring to search in the search_attribute'),

                'example_value': 'John Doe',

            },

            'sizelimit': {

                'description': _('Maximum number of entries to retrieve, between 1 and 200, default is 30.')

            },

            'scope': {

                'description': _('Scope of the LDAP search, subtree or onelevel, default is subtree.'),

            },

            'filter': {

                'description': _('Extra LDAP filter.'),

                'example_value': 'objectClass=*',

            },

        },

    )

    def search(

        self,

        request,

        ldap_base_dn,

        search_attribute,

        id_attribute,

        text_template=None,

        ldap_attributes=None,

        id=None,

        q=None,

        sizelimit=None,

        scope=None,

        filter=None,

    ):

        search_attribute = search_attribute.lower()

        id_attribute = id_attribute.lower()

        if not search_attribute.isascii():

            raise APIError('search_attribute contains non ASCII characters')

        if not id_attribute.isascii():

            raise APIError('id_attribute contains non ASCII characters')

        ldap_attributes = set(ldap_attributes.split()) if ldap_attributes else set()

        ldap_attributes.update([search_attribute, id_attribute])

        if not all(attribute.isascii() for attribute in ldap_attributes):

            raise APIError('ldap_attributes contains non ASCII characters')

        try:

            sizelimit = int(sizelimit)

        except (ValueError, TypeError):

            pass

        sizelimit = max(1, min(sizelimit or 30, 200))

        if not q and not id:

            raise APIError('q or id are mandatory parameters', http_status=400)

        if id:

            ldap_filter = '(%s=%s)' % (id_attribute, ldap.filter.escape_filter_chars(id))

        elif q:

            ldap_filter = '(%s=*%s*)' % (search_attribute, ldap.filter.escape_filter_chars(q))

        if filter:

            if not filter.startswith('('):

                filter = '(%s)' % filter

            ldap_filter = '(&%s%s)' % (ldap_filter, filter)

        print('ldap_filter', ldap_filter)

        scopes = {

            'subtree': ldap.SCOPE_SUBTREE,

            'onelevel': ldap.SCOPE_ONELEVEL,

        }

        scope = scopes.get(scope, ldap.SCOPE_SUBTREE)

        try:

            entries = list(

                self.ldap_search(ldap_base_dn, scope, ldap_filter, ldap_attributes, sizelimit=sizelimit)

            )

        except ldap.LDAPError as e:

            return {

                'err': 1,

                'data': [

                    {

                        'id': '',

                        'text': _('Directory server is unavailable'),

                        'disabled': True,

                    }

                ],

                'err_clss': 'directory-server-unavailable',

                'err_desc': str(e),

            }

        data = []

        for dn, attributes in entries:

            entry_id = attributes.get(id_attribute)

            if not entry_id:

                continue

            if text_template:

                entry_text = render_to_string(text_template, attributes)

            else:

                entry_text = attributes.get(search_attribute)

            data.append(

                {

                    'id': entry_id,

                    'text': entry_text,

                    'dn': dn,

                    'attributes': attributes,

                }

            )

        data.sort(key=lambda x: (x['text'], x['id']))

        return {'data': data}

# use a case-insensitive dictionnary to handle map of attribute to values.

class cidict(dict):

    '''Case insensitive dictionnary'''

    def __setitem__(self, key, value):

        super().__setitem__(key.lower(), value)

    def __getitem__(self, key):

        return super().__getitem__(key.lower())

    def __contains__(self, key):

        return super().__contains__(key.lower())

    def get(self, key, default=None, /):

        return super().get(key.lower(), default)

    'passerelle.apps.ldap',

from django.core import validators

class LDAPURLField(forms.URLField):

    default_validators = [validators.URLValidator(schemes=['ldap', 'ldaps'])]

from django.core import validators

class LDAPURLField(models.URLField):

    default_validators = [validators.URLValidator(schemes=['ldap', 'ldaps'])]

    def formfield(self, **kwargs):

        from .forms import LDAPURLField

        return super().formfield(

            **{

                'form_class': LDAPURLField,

                **kwargs,

            }

        )

            raise

        'python-ldap',

        'pyOpenSSL',

-----BEGIN CERTIFICATE-----

MIIDBjCCAe6gAwIBAgIUTKopT76CFlsVcI7FAilaYLILz0owDQYJKoZIhvcNAQEL

BQAwIzEhMB8GA1UEAwwYbG9jYWxob3N0LmVudHJvdXZlcnQub3JnMB4XDTE4MTIw

NTE2NTkyNFoXDTI4MTIwMjE2NTkyNFowIzEhMB8GA1UEAwwYbG9jYWxob3N0LmVu

dHJvdXZlcnQub3JnMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4xsL

n25yittbjk5bcKvY2I8zPivL6YWn2MJimaQQSNzCw/8POmVPLmMIb3lcZjydFRad

+RTxZfnuvCCJrnGrG7hOsJNenTLLU0ugN/yQ1869cM07a9tjSzL7NCz9H1NIK1+Q

cBsTExc77dOWpwWI9TjqYYRL+zex3ml8cdqcQ7BQUQxAvA4UU63DM2G+5O3dE7l8

uvyBUU3kW/shHyhfweWNXO8IXXIjvDfPYkOsjc6en2kFMr+sENSUKgfDKjz/Uzqy

S7LBb4tkJALZM8QP56VeQAG1JZF2J2/y1RqBfIGRIEkYoaHcj6UATZa1xcZjMubL

z3otRNYcRXKJMYWGbQIDAQABozIwMDAJBgNVHRMEAjAAMCMGA1UdEQQcMBqCGGxv

Y2FsaG9zdC5lbnRyb3V2ZXJ0Lm9yZzANBgkqhkiG9w0BAQsFAAOCAQEAFVPavBah

mIjgnTjq6ZbFxXTNJW0TrqN8olbKJ6SfwWVk0I8px7POekFaXd+egsFJlWYyH9q4

HkKotddRYYrWoXcPiodNfUa+bRnh2WYl2rEGMW5dbBf/MYCDts68c3SoA7JIYJ8w

0QZGAkijKNtVML0/FrLuJWbfFBAWH8JB46BcAg/8flbMHAULzV3F1g/v0A3FG3Y/

9fVr+lN5qs+NB9NXIMdf5wXrmJQYRjotyOjUO6yTFqDFvqE7DEpKQD5hnvqJoXCz

zYQS1DjH1qSRc5vC8I7YlJowCfnI9MsEICSrsk75DhT091aJC2XX93o4zhfNxmO5

Kj28hP87GHgNIg==

-----END CERTIFICATE-----

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2022  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import contextlib

import socket

import pytest

from ldaptools.slapd import Slapd, has_slapd

pytestmark = pytest.mark.skipif(not has_slapd(), reason='slapd is not installed')

def find_free_tcp_port():

    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:

        s.bind(('', 0))

        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        return s.getsockname()[1]

@pytest.fixture

def ldap_params():

    return {

        'ldap_url': 'ldap://localhost.entrouvert.org:%s' % find_free_tcp_port(),

    }

@pytest.fixture

def ldap_object(ldap_params):

    with Slapd(**ldap_params) as slapd:

        yield slapd

@pytest.fixture

def ldap_configure():

    pass

@pytest.fixture

def ldap_server(ldap_object, ldap_configure):

    return ldap_object

@pytest.fixture

def resource_class(db):

    from passerelle.apps.ldap.models import Resource

    return Resource

@pytest.fixture

def resource_params(ldap_params):

    return {

        'title': 'resource',

        'slug': 'resource',

        'description': 'resource',

        'ldap_url': ldap_params['ldap_url'],

        'ldap_base_dn': 'o=orga',

    }

@pytest.fixture

def resource(resource_class, resource_params):

    return resource_class(**resource_params)

-----BEGIN PRIVATE KEY-----

MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDjGwufbnKK21uO

Tltwq9jYjzM+K8vphafYwmKZpBBI3MLD/w86ZU8uYwhveVxmPJ0VFp35FPFl+e68

IImucasbuE6wk16dMstTS6A3/JDXzr1wzTtr22NLMvs0LP0fU0grX5BwGxMTFzvt

05anBYj1OOphhEv7N7HeaXxx2pxDsFBRDEC8DhRTrcMzYb7k7d0TuXy6/IFRTeRb

+yEfKF/B5Y1c7whdciO8N89iQ6yNzp6faQUyv6wQ1JQqB8MqPP9TOrJLssFvi2Qk

AtkzxA/npV5AAbUlkXYnb/LVGoF8gZEgSRihodyPpQBNlrXFxmMy5svPei1E1hxF

cokxhYZtAgMBAAECggEAOUZI2BxyprJLlMgOJ4wvU+5JbhR9iJc8jV34n+bQdI+4

TtW0cXW7UmeHaRWiR+Zhd0AM9xRhDObLXoaWMnhYPtVsgvunkN2OiaM49OWtYb+x

5xDbO4hIsl5ZG/98lrnaKZYgRyWM2fOyGXiTNewfbji8Y3uJ7gFNylmwGMaZQjhr

YNaqNEV7Vs2n7oERxqzKG9947oBAx2hpmoaW6eMyXcWl2ov7iHpJSKUBKho+5PWc

J731no0OsGuS+3jHa/0nZXrT8nKmemyDMdSfWmtTv659L/guFInZpHPfFVLf56vc

J6zb/IzEJV+Nh7CfBsMbHlTYBeUFlRWsy9t70+OZwQKBgQD3J4aVN4vJhXX1zDgL

dVAczwLGKXY38BoBjOeRtVhXHs5p/eNIqeZ2YbYwBBy3nL414Un7gqb9fDtg0i3n

5mQIOWhvpIYUxtwIPgYwzumxxp/n7XdU4BPDbxejZxkuC7AR5bB34pwJAJvWRGEf

0X1TxJlqULhiZ6g18O3S0oiJtwKBgQDrO9ROaj6kkxjYHmljBZIXXhdKDcn0AqPi

w20Aaafx0oxNQAoq8Gtu22Z1QHwRdBeUJwqCbmHVCCwbMf/568zFAANuT9bKMe6X

J0p0nTDiyn8w9MfduFuG4cUMn4oK6dIuYlscguoPQCvQdciwG+djqwTrHib5TEbm

jeKEkY2A+wKBgQDvXt+wy2BeqBzMF6M8Lb2OeUwVgniVyrxVPhPVgk5x6ks+SoAD

k1G62/3o2UK67lsmsfDGYA69uMGFj2qYjAHcGUW1wyF9I/BdJz01rmCWJmoe5VXK

5U8e3AyH3MV9XCKF4vCb2+UFrwo/ZnCusWVxaRqw5kb+P6ihvZuIsRE+VwKBgQC7

2duBg3bjFlUQwbiHSzuPTaRrjvdn1XPq8wVo/vcPNoS0bB+yiqxAqxT3LbfmeD8c

INFTt7KI3S3byeIRQy0TZR9YSInOjnFqZAYheiY/9lX8Un4Jod/1pvYlToJ+lJs0

T3dTHXitFSHoJydM+/ucrEYRPNMC4tb75vKty06lYQKBgQCx49+g5kQaaRZ+4psw

+eolMpAwKDkpK5gYen6OsrT8m4hpxTmtiteMsH5Avb/fxqoJWLOjhN4EnEZTMJzr

LyGoKsTv7rhZwhRznE15rOzxmldWrcCkl7DGuM2GcKgguhCYF7U7KA+vUCeqCE0H

LA2grkY+TxFpg1pwYdF1hekmTw==

-----END PRIVATE KEY-----

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2022  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os.path

import ldap

import pytest

base_dir = os.path.dirname(__file__)

cert_file = os.path.join(base_dir, 'cert.pem')

key_file = os.path.join(base_dir, 'key.pem')

def test_get_connection(resource):

    resource.get_connection()

class TestCheckStatus:

    def test_nok(self, resource):

        with pytest.raises(ldap.LDAPError):

            resource.check_status()

    def test_ok(self, resource, ldap_server):

        resource.check_status()

class TestTLSAuthentication:

    @pytest.fixture

    def ldap_params(self, ldap_params):

        ldap_params['ldap_url'] = ldap_params['ldap_url'].replace('ldap:', 'ldaps:')

        return {**ldap_params, 'tls': (key_file, cert_file)}

    @pytest.fixture

    def ldap_configure(self, ldap_object):

        conn = ldap_object.get_connection_admin()

        conn.modify_s(

            'cn=config',

            [

                (ldap.MOD_ADD, 'olcTLSCACertificateFile', cert_file.encode()),

                (ldap.MOD_ADD, 'olcTLSVerifyClient', b'demand'),

            ],

        )

    @pytest.fixture

    def resource_params(self, resource_params):

        with open(cert_file, 'rb') as cert_file_fd, open(key_file, 'rb') as key_file_fd:

            return {

                **resource_params,

                'ldap_tls_cert': cert_file_fd.read(),

                'ldap_tls_key': key_file_fd.read(),

            }

    def test_ok(self, resource, ldap_server):

        resource.check_status()

class TestLdapSearch:

    def test_nok(self, resource):

        with pytest.raises(ldap.LDAPError):

            list(resource.ldap_search('o=orga', ldap.SCOPE_SUBTREE, 'objectClass=*', ['*']))

    def test_ok(self, resource, ldap_server):

        entries = list(resource.ldap_search('o=orga', ldap.SCOPE_SUBTREE, 'objectClass=*', ['*']))

        assert entries == [('o=orga', {'o': 'orga', 'objectclass': 'organization'})]

class TestSearch:

    @pytest.fixture

    def ldap_configure(self, ldap_object):

        ldap_object.add_ldif(

            '''

dn: uid=johndoe,o=orga

objectClass: inetOrgPerson

uid: johndoe

cn: John Doe

sn: Doe

gn: John

dn: uid=janedoe,o=orga

objectClass: inetOrgPerson

uid: janedoe

cn: Jane Doe

sn: Doe

gn: Jane

dn: uid=janefoo,o=orga

objectClass: inetOrgPerson

uid: janefoo

cn: Jane Foo

sn: Foo

gn: Jane

'''

        )

    def test_server_unavailaible(self, resource):

        assert resource.search(

            request=None, q='Doe', ldap_base_dn='o=orga', search_attribute='cn', id_attribute='uid'

        ) == {

            'data': [{'disabled': True, 'id': '', 'text': 'Directory server is unavailable'}],

            'err': 1,

            'err_clss': 'directory-server-unavailable',

            'err_desc': '{\'result\': -1, \'desc\': "Can\'t contact LDAP server", '

            "'errno': 107, 'ctrls': [], 'info': 'Transport endpoint is not "

            "connected'}",

        }

    def test_q(self, resource, ldap_server):

        result = resource.search(

            request=None, q='Doe', ldap_base_dn='o=orga', search_attribute='cn', id_attribute='uid'

        )

        assert result == {

            'data': [

                {'attributes': {'cn': 'Jane Doe', 'uid': 'janedoe'}, 'id': 'janedoe', 'text': 'Jane Doe'},

                {'attributes': {'cn': 'John Doe', 'uid': 'johndoe'}, 'id': 'johndoe', 'text': 'John Doe'},

            ]

        }

    def test_id(self, resource, ldap_server):

        result = resource.search(

            request=None, id='janedoe', ldap_base_dn='o=orga', search_attribute='cn', id_attribute='uid'

        )

        assert result == {

            'data': [

                {

                    'attributes': {

                        'cn': 'Jane Doe',

                        'uid': 'janedoe',

                    },

                    'id': 'janedoe',

                    'text': 'Jane Doe',

                }

            ]

        }

    def test_q_sizelimit(self, resource, ldap_server):

        result = resource.search(

            request=None,

            q='Doe',

            ldap_base_dn='o=orga',

            search_attribute='cn',

            id_attribute='uid',

            sizelimit='1',

        )

        assert result == {

            'data': [

                {

                    'attributes': {

                        'cn': 'Jane Doe',

                        'uid': 'janedoe',

                    },

                    'id': 'janedoe',

                    'text': 'Jane Doe',

                }

            ]

        }

    def test_q_text_template(self, resource, ldap_server):

        result = resource.search(

            request=None,

            q='Doe',

            ldap_base_dn='o=orga',

            search_attribute='cn',

            id_attribute='uid',

            sizelimit='1',

            text_template='{{ sN }} {{ giVenName }} ({{ uId }})',

            ldap_attributes='givenname sn',

        )

        assert result == {

            'data': [

                {

                    'attributes': {'cn': 'Jane Doe', 'givenname': 'Jane', 'sn': 'Doe', 'uid': 'janedoe'},

                    'id': 'janedoe',

                    'text': 'Doe Jane (janedoe)',

                }

            ]

        }

  ldaptools