From cf9494362b7a4c11d1e669e5d95011cfb6eb4cc0 Mon Sep 17 00:00:00 2001 From: Christian Heimes Date: Fri, 5 Jan 2018 16:56:09 +0100 Subject: [PATCH] Refactor syncrepl tests to run with bytes_mode https://github.com/python-ldap/python-ldap/pull/150 Signed-off-by: Christian Heimes --- Tests/t_ldap_syncrepl.py | 118 ++++++++++++++++++--------------------- 1 file changed, 53 insertions(+), 65 deletions(-) diff --git a/Tests/t_ldap_syncrepl.py b/Tests/t_ldap_syncrepl.py index 0581e50..8c4eb34 100644 --- a/Tests/t_ldap_syncrepl.py +++ b/Tests/t_ldap_syncrepl.py @@ -7,8 +7,14 @@ import os -import unittest import shelve +import sys +import unittest + +if sys.version_info[0] <= 2: + PY2 = True +else: + PY2 = False from slapdtest import SlapdObject, SlapdTestCase @@ -125,9 +131,9 @@ class SyncreplClient(SimpleLDAPObject, SyncreplConsumer): Needs to be separate, because once an LDAP client starts a syncrepl search, it can't be used for anything else. """ - server_class = SyncreplProvider - def __init__(self, uri, dn, password, storage=None): + def __init__(self, uri, dn, password, storage=None, filterstr=None, + **kwargs): """ Set up our object by creating a search client, connecting, and binding. """ @@ -146,11 +152,11 @@ def __init__(self, uri, dn, password, storage=None): self.data['cookie'] = None self.present = [] self.refresh_done = False + self.filterstr = filterstr - SimpleLDAPObject.__init__(self, uri) + SimpleLDAPObject.__init__(self, uri, **kwargs) self.simple_bind_s(dn, password) - def unbind_s(self): """ In addition to unbinding from LDAP, we need to close the shelf. @@ -161,7 +167,6 @@ def unbind_s(self): self.dn_attrs.close() SimpleLDAPObject.unbind_s(self) - def search(self, search_base, search_mode): """ Start a syncrepl search operation, given a base DN and search mode. @@ -170,17 +175,15 @@ def search(self, search_base, search_mode): search_base, ldap.SCOPE_SUBTREE, mode=search_mode, - filterstr='(objectClass=*)' + filterstr=self.filterstr ) - def cancel(self): """ A simple wrapper to call parent class with syncrepl search ID. """ SimpleLDAPObject.cancel(self, self.search_id) - def poll(self, timeout=None, all=0): """ Take the params, add the syncrepl search ID, and call the proper poll. @@ -191,28 +194,24 @@ def poll(self, timeout=None, all=0): all=all ) - def syncrepl_get_cookie(self): """ Pull cookie from storage, if one exists. """ return self.data['cookie'] - def syncrepl_set_cookie(self, cookie): """ Update stored cookie. """ self.data['cookie'] = cookie - def syncrepl_refreshdone(self): """ Just update a variable. """ self.refresh_done = True - def syncrepl_delete(self, uuids): """ Delete the given items from both maps. @@ -221,7 +220,6 @@ def syncrepl_delete(self, uuids): del self.dn_attrs[self.uuid_dn[uuid]] del self.uuid_dn[uuid] - def syncrepl_entry(self, dn, attrs, uuid): """ Handles adds and changes (including DN changes). @@ -236,7 +234,6 @@ def syncrepl_entry(self, dn, attrs, uuid): self.uuid_dn[uuid] = dn self.dn_attrs[dn] = attrs - def syncrepl_present(self, uuids, refreshDeletes=False): """ The 'present' message from the LDAP server is the most complicated @@ -262,7 +259,7 @@ def syncrepl_present(self, uuids, refreshDeletes=False): pass -class Test00_Syncrepl(SlapdTestCase): +class BaseSyncreplTests(object): """ This is a test of all the basic Syncrepl operations. It covers starting a search (both types of search), doing the refresh part of the search, @@ -275,7 +272,7 @@ class Test00_Syncrepl(SlapdTestCase): @classmethod def setUpClass(cls): - super(Test00_Syncrepl, cls).setUpClass() + super(BaseSyncreplTests, cls).setUpClass() # insert some Foo* objects via ldapadd cls.server.ldapadd( LDIF_TEMPLATE % { @@ -287,57 +284,39 @@ def setUpClass(cls): } ) - def setUp(self): - try: - self._ldap_conn - except AttributeError: - # open local LDAP connection - self._ldap_conn = self._open_ldap_conn() - + super(BaseSyncreplTests, self).setUp() + self.tester = None + self.suffix = None def tearDown(self): self.tester.unbind_s() + super(BaseSyncreplTests, self).tearDown() + def create_client(self): + raise NotImplementedError def test_refreshOnly_search(self): ''' Test to see if we can initialize a syncrepl search. ''' - self.tester = SyncreplClient( - self.server.ldap_uri, - self.server.root_dn, - self.server.root_pw - ) self.tester.search( - self.server.suffix, + self.suffix, 'refreshOnly' ) - def test_refreshAndPersist_search(self): - self.tester = SyncreplClient( - self.server.ldap_uri, - self.server.root_dn, - self.server.root_pw - ) self.tester.search( - self.server.suffix, + self.suffix, 'refreshAndPersist' ) - def test_refreshOnly_poll_full(self): """ Test doing a full refresh cycle, and check what we got. """ - self.tester = SyncreplClient( - self.server.ldap_uri, - self.server.root_dn, - self.server.root_pw - ) self.tester.search( - self.server.suffix, + self.suffix, 'refreshOnly' ) poll_result = self.tester.poll( @@ -347,18 +326,12 @@ def test_refreshOnly_poll_full(self): self.assertFalse(poll_result) self.assertEqual(self.tester.dn_attrs, LDAP_ENTRIES) - def test_refreshAndPersist_poll_only(self): """ Test the refresh part of refresh-and-persist, and check what we got. """ - self.tester = SyncreplClient( - self.server.ldap_uri, - self.server.root_dn, - self.server.root_pw - ) self.tester.search( - self.server.suffix, + self.suffix, 'refreshAndPersist' ) @@ -372,18 +345,12 @@ def test_refreshAndPersist_poll_only(self): self.assertEqual(self.tester.dn_attrs, LDAP_ENTRIES) - def test_refreshAndPersist_timeout(self): """ Make sure refreshAndPersist can handle a search with timeouts. """ - self.tester = SyncreplClient( - self.server.ldap_uri, - self.server.root_dn, - self.server.root_pw - ) self.tester.search( - self.server.suffix, + self.suffix, 'refreshAndPersist' ) @@ -407,18 +374,12 @@ def test_refreshAndPersist_timeout(self): timeout=1 ) - def test_refreshAndPersist_cancelled(self): """ Make sure refreshAndPersist can handle cancelling a syncrepl search. """ - self.tester = SyncreplClient( - self.server.ldap_uri, - self.server.root_dn, - self.server.root_pw - ) self.tester.search( - self.server.suffix, + self.suffix, 'refreshAndPersist' ) @@ -463,5 +424,32 @@ def test_refreshAndPersist_cancelled(self): # should pick it up during the persist phase. +class TestSyncrepl(BaseSyncreplTests, SlapdTestCase): + def setUp(self): + super(TestSyncrepl, self).setUp() + self.tester = SyncreplClient( + self.server.ldap_uri, + self.server.root_dn, + self.server.root_pw, + filterstr=u'(objectClass=*)', + bytes_mode=False + ) + self.suffix = self.server.suffix + + +@unittest.skipUnless(PY2, "no bytes_mode under Py3") +class TestSyncreplBytesMode(BaseSyncreplTests, SlapdTestCase): + def setUp(self): + super(TestSyncreplBytesMode, self).setUp() + self.tester = SyncreplClient( + self.server.ldap_uri, + self.server.root_dn.encode('utf-8'), + self.server.root_pw.encode('utf-8'), + filterstr=b'(objectClass=*)', + bytes_mode=True + ) + self.suffix = self.server.suffix.encode('utf-8') + + if __name__ == '__main__': unittest.main()