# -*- coding: utf-8 -*-
"""
Automatic tests for python-ldap's module ldap.ldapobject

See https://www.python-ldap.org/ for details.
"""

from __future__ import unicode_literals

import sys

if sys.version_info[0] <= 2:
    PY2 = True
    text_type = unicode
else:
    PY2 = False
    text_type = str

import errno
import contextlib
import linecache
import os
import unittest
import warnings
import pickle

# Switch off processing .ldaprc or ldap.conf before importing _ldap
os.environ['LDAPNOINIT'] = '1'

import ldap
from ldap.ldapobject import SimpleLDAPObject, ReconnectLDAPObject

from slapdtest import SlapdTestCase
from slapdtest import requires_ldapi, requires_sasl, requires_tls


LDIF_TEMPLATE = """dn: %(suffix)s
objectClass: dcObject
objectClass: organization
dc: %(dc)s
o: %(dc)s

dn: %(rootdn)s
objectClass: applicationProcess
objectClass: simpleSecurityObject
cn: %(rootcn)s
userPassword: %(rootpw)s

dn: cn=user1,%(suffix)s
objectClass: applicationProcess
objectClass: simpleSecurityObject
cn: user1
userPassword: user1_pw

dn: cn=Foo1,%(suffix)s
objectClass: organizationalRole
cn: Foo1

dn: cn=Foo2,%(suffix)s
objectClass: organizationalRole
cn: Foo2

dn: cn=Foo3,%(suffix)s
objectClass: organizationalRole
cn: Foo3

dn: ou=Container,%(suffix)s
objectClass: organizationalUnit
ou: Container

dn: cn=Foo4,ou=Container,%(suffix)s
objectClass: organizationalRole
cn: Foo4

"""


class Test00_SimpleLDAPObject(SlapdTestCase):
    """
    test LDAP search operations
    """

    ldap_object_class = SimpleLDAPObject

    @classmethod
    def setUpClass(cls):
        super(Test00_SimpleLDAPObject, cls).setUpClass()
        # insert some Foo* objects via ldapadd
        cls.server.ldapadd(
            LDIF_TEMPLATE % {
                'suffix':cls.server.suffix,
                'rootdn':cls.server.root_dn,
                'rootcn':cls.server.root_cn,
                'rootpw':cls.server.root_pw,
                'dc': cls.server.suffix.split(',')[0][3:],
            }
        )

    def setUp(self):
        try:
            self._ldap_conn
        except AttributeError:
            # open local LDAP connection
            self._ldap_conn = self._open_ldap_conn(bytes_mode=False)

    def test_reject_bytes_base(self):
        base = self.server.suffix
        l = self._ldap_conn

        with self.assertRaises(TypeError) as e:
            l.search_s(
                base.encode('utf-8'), ldap.SCOPE_SUBTREE, '(cn=Foo*)', ['*']
            )
        if PY2:
            self.assertIn(
                u"got type 'str' for 'base'", text_type(e.exception)
            )
        elif sys.version_info >= (3, 5, 0):
            # Python 3.4.x does not include 'search_ext()' in message
            self.assertEqual(
                "search_ext() argument 1 must be str, not bytes",
                text_type(e.exception)
            )

        with self.assertRaises(TypeError) as e:
            l.search_s(
                base, ldap.SCOPE_SUBTREE, b'(cn=Foo*)', ['*']
            )
        if PY2:
            self.assertIn(
                u"got type 'str' for 'filterstr'", text_type(e.exception)
            )
        elif sys.version_info >= (3, 5, 0):
            self.assertEqual(
                "search_ext() argument 3 must be str, not bytes",
                text_type(e.exception)
            )

        with self.assertRaises(TypeError) as e:
            l.search_s(
                base, ldap.SCOPE_SUBTREE, '(cn=Foo*)', [b'*']
            )
        if PY2:
            self.assertIn(
                u"got type 'str' for 'attrlist'", text_type(e.exception)
            )
        elif sys.version_info >= (3, 5, 0):
            self.assertEqual(
                ('attrs_from_List(): expected string in list', b'*'),
                e.exception.args
            )

    def test_search_keys_are_text(self):
        base = self.server.suffix
        l = self._ldap_conn
        result = l.search_s(base, ldap.SCOPE_SUBTREE, '(cn=Foo*)', ['*'])
        result.sort()
        dn, fields = result[0]
        self.assertEqual(dn, 'cn=Foo1,%s' % base)
        self.assertEqual(type(dn), text_type)
        for key, values in fields.items():
            self.assertEqual(type(key), text_type)
            for value in values:
                self.assertEqual(type(value), bytes)

    def _get_bytes_ldapobject(self, explicit=True, **kwargs):
        if explicit:
            kwargs.setdefault('bytes_mode', True)
        else:
            kwargs = {}
        return self._open_ldap_conn(
            who=self.server.root_dn.encode('utf-8'),
            cred=self.server.root_pw.encode('utf-8'),
            **kwargs
        )

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_bytesmode_search_requires_bytes(self):
        l = self._get_bytes_ldapobject()
        base = self.server.suffix

        with self.assertRaises(TypeError):
            l.search_s(base.encode('utf-8'), ldap.SCOPE_SUBTREE, '(cn=Foo*)', [b'*'])
        with self.assertRaises(TypeError):
            l.search_s(base.encode('utf-8'), ldap.SCOPE_SUBTREE, b'(cn=Foo*)', ['*'])
        with self.assertRaises(TypeError):
            l.search_s(base, ldap.SCOPE_SUBTREE, b'(cn=Foo*)', [b'*'])

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_bytesmode_search_results_have_bytes(self):
        l = self._get_bytes_ldapobject()
        base = self.server.suffix
        result = l.search_s(base.encode('utf-8'), ldap.SCOPE_SUBTREE, b'(cn=Foo*)', [b'*'])
        result.sort()
        dn, fields = result[0]
        self.assertEqual(dn, b'cn=Foo1,%s' % base)
        self.assertEqual(type(dn), bytes)
        for key, values in fields.items():
            self.assertEqual(type(key), bytes)
            for value in values:
                self.assertEqual(type(value), bytes)

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_bytesmode_search_defaults(self):
        l = self._get_bytes_ldapobject()
        base = 'cn=Foo1,' + self.server.suffix
        kwargs = dict(
            base=base.encode('utf-8'),
            scope=ldap.SCOPE_SUBTREE,
            # filterstr=b'(objectClass=*)'
        )
        expected = [
            (
                base,
                {'cn': [b'Foo1'], 'objectClass': [b'organizationalRole']}
            ),
        ]

        result = l.search_s(**kwargs)
        self.assertEqual(result, expected)
        result = l.search_st(**kwargs)
        self.assertEqual(result, expected)
        result = l.search_ext_s(**kwargs)
        self.assertEqual(result, expected)

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_unset_bytesmode_search_warns_bytes(self):
        l = self._get_bytes_ldapobject(explicit=False)
        base = self.server.suffix

        l.search_s(base.encode('utf-8'), ldap.SCOPE_SUBTREE, '(cn=Foo*)', [b'*'])
        l.search_s(base.encode('utf-8'), ldap.SCOPE_SUBTREE, b'(cn=Foo*)', ['*'])
        l.search_s(base, ldap.SCOPE_SUBTREE, b'(cn=Foo*)', [b'*'])

    def _search_wrong_type(self, bytes_mode, strictness):
        if bytes_mode:
            l = self._get_bytes_ldapobject(bytes_strictness=strictness)
        else:
            l = self._open_ldap_conn(bytes_mode=False,
                                     bytes_strictness=strictness)
        base = 'cn=Foo1,' + self.server.suffix
        if not bytes_mode:
            base = base.encode('utf-8')
        result = l.search_s(base, scope=ldap.SCOPE_SUBTREE)
        return result[0][-1]['cn']

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_bytesmode_silent(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            self._search_wrong_type(bytes_mode=True, strictness='silent')
        self.assertEqual(w, [])

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_bytesmode_warn(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            self._search_wrong_type(bytes_mode=True, strictness='warn')
        self.assertEqual(len(w), 1)

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_bytesmode_error(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            with self.assertRaises(TypeError):
                self._search_wrong_type(bytes_mode=True, strictness='error')
        self.assertEqual(w, [])

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_textmode_silent(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            self._search_wrong_type(bytes_mode=True, strictness='silent')
        self.assertEqual(w, [])

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_textmode_warn(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            self._search_wrong_type(bytes_mode=True, strictness='warn')
        self.assertEqual(len(w), 1)

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_textmode_error(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            with self.assertRaises(TypeError):
                self._search_wrong_type(bytes_mode=True, strictness='error')
        self.assertEqual(w, [])

    def test_search_accepts_unicode_dn(self):
        base = self.server.suffix
        l = self._ldap_conn

        with self.assertRaises(ldap.NO_SUCH_OBJECT):
            result = l.search_s("CN=abc\U0001f498def", ldap.SCOPE_SUBTREE)

    def test_filterstr_accepts_unicode(self):
        l = self._ldap_conn
        base = self.server.suffix
        result = l.search_s(base, ldap.SCOPE_SUBTREE, '(cn=abc\U0001f498def)', ['*'])
        self.assertEqual(result, [])

    def test_attrlist_accepts_unicode(self):
        base = self.server.suffix
        result = self._ldap_conn.search_s(
            base, ldap.SCOPE_SUBTREE,
            '(cn=Foo*)', ['abc', 'abc\U0001f498def'])
        result.sort()

        for dn, attrs in result:
            self.assertIsInstance(dn, text_type)
            self.assertEqual(attrs, {})

    def test001_search_subtree(self):
        result = self._ldap_conn.search_s(
            self.server.suffix,
            ldap.SCOPE_SUBTREE,
            '(cn=Foo*)',
            attrlist=['*'],
        )
        result.sort()
        self.assertEqual(
            result,
            [
                (
                    'cn=Foo1,'+self.server.suffix,
                    {'cn': [b'Foo1'], 'objectClass': [b'organizationalRole']}
                ),
                (
                    'cn=Foo2,'+self.server.suffix,
                    {'cn': [b'Foo2'], 'objectClass': [b'organizationalRole']}
                ),
                (
                    'cn=Foo3,'+self.server.suffix,
                    {'cn': [b'Foo3'], 'objectClass': [b'organizationalRole']}
                ),
                (
                    'cn=Foo4,ou=Container,'+self.server.suffix,
                    {'cn': [b'Foo4'], 'objectClass': [b'organizationalRole']}
                ),
            ]
        )

    def test002_search_onelevel(self):
        result = self._ldap_conn.search_s(
            self.server.suffix,
            ldap.SCOPE_ONELEVEL,
            '(cn=Foo*)',
            ['*'],
        )
        result.sort()
        self.assertEqual(
            result,
            [
                (
                    'cn=Foo1,'+self.server.suffix,
                    {'cn': [b'Foo1'], 'objectClass': [b'organizationalRole']}
                ),
                (
                    'cn=Foo2,'+self.server.suffix,
                    {'cn': [b'Foo2'], 'objectClass': [b'organizationalRole']}
                ),
                (
                    'cn=Foo3,'+self.server.suffix,
                    {'cn': [b'Foo3'], 'objectClass': [b'organizationalRole']}
                ),
            ]
        )

    def test003_search_oneattr(self):
        result = self._ldap_conn.search_s(
            self.server.suffix,
            ldap.SCOPE_SUBTREE,
            '(cn=Foo4)',
            ['cn'],
        )
        result.sort()
        self.assertEqual(
            result,
            [('cn=Foo4,ou=Container,'+self.server.suffix, {'cn': [b'Foo4']})]
        )

    def test_find_unique_entry(self):
        result = self._ldap_conn.find_unique_entry(
            self.server.suffix,
            ldap.SCOPE_SUBTREE,
            '(cn=Foo4)',
            ['cn'],
        )
        self.assertEqual(
            result,
            ('cn=Foo4,ou=Container,'+self.server.suffix, {'cn': [b'Foo4']})
        )
        with self.assertRaises(ldap.SIZELIMIT_EXCEEDED):
            # > 2 entries returned
            self._ldap_conn.find_unique_entry(
                self.server.suffix,
                ldap.SCOPE_ONELEVEL,
                '(cn=Foo*)',
                ['*'],
            )
        with self.assertRaises(ldap.NO_UNIQUE_ENTRY):
            # 0 entries returned
            self._ldap_conn.find_unique_entry(
                self.server.suffix,
                ldap.SCOPE_ONELEVEL,
                '(cn=Bar*)',
                ['*'],
            )

    def test_search_subschema(self):
        l = self._ldap_conn
        dn = l.search_subschemasubentry_s()
        self.assertIsInstance(dn, text_type)
        self.assertEqual(dn, "cn=Subschema")
        subschema = l.read_subschemasubentry_s(dn)
        self.assertIsInstance(subschema, dict)
        self.assertEqual(
            sorted(subschema),
            [
                u'attributeTypes',
                u'ldapSyntaxes',
                u'matchingRuleUse',
                u'matchingRules',
                u'objectClasses'
            ]
        )

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_search_subschema_have_bytes(self):
        l = self._get_bytes_ldapobject()
        dn = l.search_subschemasubentry_s()
        self.assertIsInstance(dn, bytes)
        self.assertEqual(dn, b"cn=Subschema")
        subschema = l.read_subschemasubentry_s(dn)
        self.assertIsInstance(subschema, dict)
        self.assertEqual(
            sorted(subschema),
            [
                b'attributeTypes',
                b'ldapSyntaxes',
                b'matchingRuleUse',
                b'matchingRules',
                b'objectClasses'
            ]
        )

    def test004_enotconn(self):
        l = self.ldap_object_class('ldap://127.0.0.1:42')
        try:
            m = l.simple_bind_s("", "")
            r = l.result4(m, ldap.MSG_ALL, self.timeout)
        except ldap.SERVER_DOWN as ldap_err:
            errno_val = ldap_err.args[0]['errno']
            if errno_val != errno.ENOTCONN:
                self.fail("expected errno=%d, got %d"
                          % (errno.ENOTCONN, errno_val))
            info = ldap_err.args[0]['info']
            expected_info = os.strerror(errno.ENOTCONN)
            if info != expected_info:
                self.fail("expected info=%r, got %d" % (expected_info, info))
        else:
            self.fail("expected SERVER_DOWN, got %r" % r)

    def test005_invalid_credentials(self):
        l = self.ldap_object_class(self.server.ldap_uri)
        # search with invalid filter
        try:
            m = l.simple_bind(self.server.root_dn, self.server.root_pw+'wrong')
            r = l.result4(m, ldap.MSG_ALL)
        except ldap.INVALID_CREDENTIALS:
            pass
        else:
            self.fail("expected INVALID_CREDENTIALS, got %r" % r)

    @requires_sasl()
    @requires_ldapi()
    def test006_sasl_extenal_bind_s(self):
        l = self.ldap_object_class(self.server.ldapi_uri)
        l.sasl_external_bind_s()
        self.assertEqual(l.whoami_s(), 'dn:'+self.server.root_dn.lower())
        authz_id = 'dn:cn=Foo2,%s' % (self.server.suffix)
        l = self.ldap_object_class(self.server.ldapi_uri)
        l.sasl_external_bind_s(authz_id=authz_id)
        self.assertEqual(l.whoami_s(), authz_id.lower())

    def test007_timeout(self):
        l = self.ldap_object_class(self.server.ldap_uri)
        m = l.search_ext(self.server.suffix, ldap.SCOPE_SUBTREE, '(objectClass=*)')
        l.abandon(m)
        with self.assertRaises(ldap.TIMEOUT):
            result = l.result(m, timeout=0.001)

    def assertIsSubclass(self, cls, other):
        self.assertTrue(
            issubclass(cls, other),
            cls.__mro__
        )

    def test_simple_bind_noarg(self):
        l = self.ldap_object_class(self.server.ldap_uri)
        l.simple_bind_s()
        self.assertEqual(l.whoami_s(), u'')
        l = self.ldap_object_class(self.server.ldap_uri)
        l.simple_bind_s(None, None)
        self.assertEqual(l.whoami_s(), u'')

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_ldapbyteswarning(self):
        self.assertIsSubclass(ldap.LDAPBytesWarning, BytesWarning)
        self.assertIsSubclass(ldap.LDAPBytesWarning, Warning)
        self.assertIsInstance(self.server.suffix, text_type)
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            conn = self._get_bytes_ldapobject(explicit=False)
            result = conn.search_s(
                self.server.suffix,
                ldap.SCOPE_SUBTREE,
                b'(cn=Foo*)',
                attrlist=[b'*'],
            )
            self.assertEqual(len(result), 4)

        # ReconnectLDAP only emits one warning
        self.assertGreaterEqual(len(w), 1, w)
        msg = w[-1]
        self.assertIs(msg.category, ldap.LDAPBytesWarning)
        self.assertEqual(
            text_type(msg.message),
            "Received non-bytes value for 'base' in bytes "
            "mode; please choose an explicit option for bytes_mode on your "
            "LDAP connection"
        )

    @contextlib.contextmanager
    def catch_byteswarnings(self, *args, **kwargs):
        with warnings.catch_warnings(record=True) as w:
            conn = self._get_bytes_ldapobject(*args, **kwargs)
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            yield conn, w

    def _check_byteswarning(self, warning, expected_message):
        self.assertIs(warning.category, ldap.LDAPBytesWarning)
        self.assertIn(expected_message, text_type(warning.message))

        def _normalize(filename):
            # Python 2 likes to report the ".pyc" file in warnings,
            # tracebacks or __file__.
            # Use the corresponding ".py" in that case.
            if filename.endswith('.pyc'):
                return filename[:-1]
            return filename

        # Assert warning points to a line marked CORRECT LINE in this file
        self.assertEquals(_normalize(warning.filename), _normalize(__file__))
        self.assertIn(
            'CORRECT LINE',
            linecache.getline(warning.filename, warning.lineno)
        )

    def _test_byteswarning_level_search(self, methodname):
        with self.catch_byteswarnings(explicit=False) as (conn, w):
            method = getattr(conn, methodname)
            result = method(
                self.server.suffix.encode('utf-8'),
                ldap.SCOPE_SUBTREE,
                '(cn=Foo*)',
                attrlist=['*'],  # CORRECT LINE
            )
            self.assertEqual(len(result), 4)

        self.assertEqual(len(w), 2, w)

        self._check_byteswarning(
            w[0], u"Received non-bytes value for 'filterstr'")

        self._check_byteswarning(
            w[1], u"Received non-bytes value for 'attrlist'")

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_byteswarning_level_search(self):
        self._test_byteswarning_level_search('search_s')
        self._test_byteswarning_level_search('search_st')
        self._test_byteswarning_level_search('search_ext_s')

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_byteswarning_initialize(self):
        with warnings.catch_warnings(record=True) as w:
            warnings.resetwarnings()
            warnings.simplefilter('always', ldap.LDAPBytesWarning)
            bytes_uri = self.server.ldap_uri.decode('utf-8')
            self.ldap_object_class(bytes_uri)  # CORRECT LINE

        self.assertEqual(len(w), 1, w)

        self._check_byteswarning(
            w[0], u"Under Python 2, python-ldap uses bytes by default.")

    @requires_tls()
    def test_multiple_starttls(self):
        # Test for openldap does not re-register nss shutdown callbacks
        # after nss_Shutdown is called
        # https://github.com/python-ldap/python-ldap/issues/60
        # https://bugzilla.redhat.com/show_bug.cgi?id=1520990
        for _ in range(10):
            l = self.ldap_object_class(self.server.ldap_uri)
            l.set_option(ldap.OPT_X_TLS_CACERTFILE, self.server.cafile)
            l.set_option(ldap.OPT_X_TLS_NEWCTX, 0)
            l.start_tls_s()
            l.simple_bind_s(self.server.root_dn, self.server.root_pw)
            self.assertEqual(l.whoami_s(), 'dn:' + self.server.root_dn)

    def test_dse(self):
        dse = self._ldap_conn.read_rootdse_s()
        self.assertIsInstance(dse, dict)
        self.assertEqual(dse[u'supportedLDAPVersion'], [b'3'])
        keys = set(dse)
        # SASL info may be missing in restricted build environments
        keys.discard(u'supportedSASLMechanisms')
        self.assertEqual(
            keys,
            {u'configContext', u'entryDN', u'namingContexts', u'objectClass',
             u'structuralObjectClass', u'subschemaSubentry',
             u'supportedControl', u'supportedExtension', u'supportedFeatures',
             u'supportedLDAPVersion'}
        )
        self.assertEqual(
            self._ldap_conn.get_naming_contexts(),
            [self.server.suffix.encode('utf-8')]
        )

    @unittest.skipUnless(PY2, "no bytes_mode under Py3")
    def test_dse_bytes(self):
        l = self._get_bytes_ldapobject()
        dse = l.read_rootdse_s()
        self.assertIsInstance(dse, dict)
        self.assertEqual(dse[u'supportedLDAPVersion'], [b'3'])
        self.assertEqual(
            l.get_naming_contexts(),
            [self.server.suffix.encode('utf-8')]
        )

    def test_compare_s_true(self):
        base = self.server.suffix
        l = self._ldap_conn
        result = l.compare_s('cn=Foo1,%s' % base, 'cn', b'Foo1')
        self.assertIs(result, True)

    def test_compare_s_false(self):
        base = self.server.suffix
        l = self._ldap_conn
        result = l.compare_s('cn=Foo1,%s' % base, 'cn', b'Foo2')
        self.assertIs(result, False)


class Test01_ReconnectLDAPObject(Test00_SimpleLDAPObject):
    """
    test ReconnectLDAPObject by restarting slapd
    """

    ldap_object_class = ReconnectLDAPObject

    @requires_sasl()
    @requires_ldapi()
    def test101_reconnect_sasl_external(self):
        l = self.ldap_object_class(self.server.ldapi_uri)
        l.sasl_external_bind_s()
        authz_id = l.whoami_s()
        self.assertEqual(authz_id, 'dn:'+self.server.root_dn.lower())
        self.server.restart()
        self.assertEqual(l.whoami_s(), authz_id)

    def test102_reconnect_simple_bind(self):
        l = self.ldap_object_class(self.server.ldap_uri)
        bind_dn = 'cn=user1,'+self.server.suffix
        l.simple_bind_s(bind_dn, 'user1_pw')
        self.assertEqual(l.whoami_s(), 'dn:'+bind_dn)
        self.server.restart()
        self.assertEqual(l.whoami_s(), 'dn:'+bind_dn)

    def test103_reconnect_get_state(self):
        l1 = self.ldap_object_class(self.server.ldap_uri)
        bind_dn = 'cn=user1,'+self.server.suffix
        l1.simple_bind_s(bind_dn, 'user1_pw')
        self.assertEqual(l1.whoami_s(), 'dn:'+bind_dn)
        self.assertEqual(
            l1.__getstate__(),
            {
                str('_last_bind'): (
                    'simple_bind_s',
                    (bind_dn, 'user1_pw'),
                    {}
                ),
                str('_options'): [(17, 3)],
                str('_reconnects_done'): 0,
                str('_retry_delay'): 60.0,
                str('_retry_max'): 1,
                str('_start_tls'): 0,
                str('_trace_level'): ldap._trace_level,
                str('_trace_stack_limit'): 5,
                str('_uri'): self.server.ldap_uri,
                str('bytes_mode'): l1.bytes_mode,
                str('bytes_strictness'): l1.bytes_strictness,
                str('timeout'): -1,
            },
        )

    def test104_reconnect_restore(self):
        l1 = self.ldap_object_class(self.server.ldap_uri)
        bind_dn = 'cn=user1,'+self.server.suffix
        l1.simple_bind_s(bind_dn, 'user1_pw')
        self.assertEqual(l1.whoami_s(), 'dn:'+bind_dn)
        l1_state = pickle.dumps(l1)
        del l1
        l2 = pickle.loads(l1_state)
        self.assertEqual(l2.whoami_s(), 'dn:'+bind_dn)


if __name__ == '__main__':
    unittest.main()