s4:tests/dirsync: add tests for dirsync with extended_dn

[metze/samba/wip.git] / source4 / dsdb / tests / python / dirsync.py

diff --git a/source4/dsdb/tests/python/dirsync.py b/source4/dsdb/tests/python/dirsync.py
index 0f49fd7e6304de6208cde39056f1e3568f3e99a5..405980455b7c150c63e31d089301c921dca4d0d6 100755 (executable)
--- a/source4/dsdb/tests/python/dirsync.py
+++ b/source4/dsdb/tests/python/dirsync.py
@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 #
 # Unit tests for dirsync control
 # Copyright (C) Matthieu Patou <mat@matws.net> 2011
@@ -31,7 +31,7 @@ import base64
 from ldb import LdbError, SCOPE_BASE
 from ldb import Message, MessageElement, Dn
 from ldb import FLAG_MOD_ADD, FLAG_MOD_DELETE
-from samba.dcerpc import security, misc, drsblobs, security
+from samba.dcerpc import security, misc, drsblobs
 from samba.ndr import ndr_unpack, ndr_pack
 
 from samba.auth import system_session
@@ -58,7 +58,7 @@ if len(args) < 1:
     sys.exit(1)
 
 host = args.pop()
-if not "://" in host:
+if "://" not in host:
     ldaphost = "ldap://%s" % host
     ldapshost = "ldaps://%s" % host
 else:
@@ -73,6 +73,7 @@ creds = credopts.get_credentials(lp)
 # Tests start here
 #
 
+
 class DirsyncBaseTests(samba.tests.TestCase):
 
     def setUp(self):
@@ -228,19 +229,19 @@ class SimpleDirsyncTests(DirsyncBaseTests):
                                     expression="samaccountname=*",
                                     controls=["dirsync:1:0:1"])
         # Check that nTSecurityDescriptor is returned as it's the case when doing dirsync
-        self.assertTrue(res.msgs[0].get("ntsecuritydescriptor") != None)
+        self.assertTrue(res.msgs[0].get("ntsecuritydescriptor") is not None)
         # Check that non replicated attributes are not returned
-        self.assertTrue(res.msgs[0].get("badPwdCount") == None)
+        self.assertTrue(res.msgs[0].get("badPwdCount") is None)
         # Check that non forward link are not returned
-        self.assertTrue(res.msgs[0].get("memberof") == None)
+        self.assertTrue(res.msgs[0].get("memberof") is None)
 
         # Asking for instanceType will return also objectGUID
         res = self.ldb_admin.search(self.base_dn,
                                     expression="samaccountname=Administrator",
                                     attrs=["instanceType"],
                                     controls=["dirsync:1:0:1"])
-        self.assertTrue(res.msgs[0].get("objectGUID") != None)
-        self.assertTrue(res.msgs[0].get("instanceType") != None)
+        self.assertTrue(res.msgs[0].get("objectGUID") is not None)
+        self.assertTrue(res.msgs[0].get("instanceType") is not None)
 
         # We don't return an entry if asked for objectGUID
         res = self.ldb_admin.search(self.base_dn,
@@ -254,10 +255,10 @@ class SimpleDirsyncTests(DirsyncBaseTests):
                                     expression="(distinguishedName=%s)" % str(self.base_dn),
                                     attrs=["name"],
                                     controls=["dirsync:1:0:1"])
-        self.assertTrue(res.msgs[0].get("objectGUID") != None)
-        self.assertTrue(res.msgs[0].get("name") != None)
-        self.assertTrue(res.msgs[0].get("parentGUID") == None)
-        self.assertTrue(res.msgs[0].get("instanceType") != None)
+        self.assertTrue(res.msgs[0].get("objectGUID") is not None)
+        self.assertTrue(res.msgs[0].get("name") is not None)
+        self.assertTrue(res.msgs[0].get("parentGUID") is None)
+        self.assertTrue(res.msgs[0].get("instanceType") is not None)
 
         # Asking for name will return also objectGUID and parentGUID
         # and instanceType and of course name
@@ -265,10 +266,10 @@ class SimpleDirsyncTests(DirsyncBaseTests):
                                     expression="samaccountname=Administrator",
                                     attrs=["name"],
                                     controls=["dirsync:1:0:1"])
-        self.assertTrue(res.msgs[0].get("objectGUID") != None)
-        self.assertTrue(res.msgs[0].get("name") != None)
-        self.assertTrue(res.msgs[0].get("parentGUID") != None)
-        self.assertTrue(res.msgs[0].get("instanceType") != None)
+        self.assertTrue(res.msgs[0].get("objectGUID") is not None)
+        self.assertTrue(res.msgs[0].get("name") is not None)
+        self.assertTrue(res.msgs[0].get("parentGUID") is not None)
+        self.assertTrue(res.msgs[0].get("instanceType") is not None)
 
         # Asking for dn will not return not only DN but more like if attrs=*
         # parentGUID should be returned
@@ -314,14 +315,14 @@ class SimpleDirsyncTests(DirsyncBaseTests):
         delete_force(self.ldb_admin, ouname)
 
     def test_dirsync_with_controls(self):
-        """Check that dirsync return correct informations when dealing with the NC"""
+        """Check that dirsync return correct information when dealing with the NC"""
         res = self.ldb_admin.search(self.base_dn,
                                     expression="(distinguishedName=%s)" % str(self.base_dn),
                                     attrs=["name"],
                                     controls=["dirsync:1:0:10000", "extended_dn:1", "show_deleted:1"])
 
     def test_dirsync_basenc(self):
-        """Check that dirsync return correct informations when dealing with the NC"""
+        """Check that dirsync return correct information when dealing with the NC"""
         res = self.ldb_admin.search(self.base_dn,
                                     expression="(distinguishedName=%s)" % str(self.base_dn),
                                     attrs=["name"],
@@ -365,14 +366,13 @@ class SimpleDirsyncTests(DirsyncBaseTests):
                                     controls=["dirsync:1:0:0"])
         self.assertEqual(len(res.msgs), nb - 1)
         if nb > 1:
-            self.assertTrue(res.msgs[0].get("objectGUID") != None)
+            self.assertTrue(res.msgs[0].get("objectGUID") is not None)
         else:
             res = self.ldb_admin.search(self.base_dn,
                                         expression="(objectclass=configuration)",
                                         attrs=["objectGUID"],
                                         controls=["dirsync:1:0:0"])
 
-
     def test_dirsync_send_delta(self):
         """Check that dirsync return correct delta when sending the last cookie"""
         res = self.ldb_admin.search(self.base_dn,
@@ -454,8 +454,8 @@ class SimpleDirsyncTests(DirsyncBaseTests):
                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
                                     controls=[control3])
 
-        self.assertTrue(res[0].get("parentGUID") != None)
-        self.assertTrue(res[0].get("name") != None)
+        self.assertTrue(res[0].get("parentGUID") is not None)
+        self.assertTrue(res[0].get("name") is not None)
         delete_force(self.ldb_admin, ouname)
 
     def test_dirsync_linkedattributes(self):
@@ -540,8 +540,6 @@ class SimpleDirsyncTests(DirsyncBaseTests):
         self.ldb_admin.deletegroup("testgroup")
         self.assertEqual(len(res[0].get("member")), 0)
 
-
-
     def test_dirsync_deleted_items(self):
         """Check that dirsync returnd deleted objects too"""
         # Let's create an OU
@@ -573,7 +571,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
         guid2 = str(ndr_unpack(misc.GUID, res[0].get("objectGUID")[0]))
         self.assertEqual(guid2, guid)
         self.assertTrue(res[0].get("isDeleted"))
-        self.assertTrue(res[0].get("name") != None)
+        self.assertTrue(res[0].get("name") is not None)
 
     def test_cookie_from_others(self):
         res = self.ldb_admin.search(self.base_dn,
@@ -598,7 +596,7 @@ class ExtendedDirsyncTests(SimpleDirsyncTests):
                                     expression="(name=Administrators)",
                                     controls=["dirsync:1:%d:1" % flag_incr_linked])
 
-        self.assertTrue(res[0].get("member;range=1-1") != None)
+        self.assertTrue(res[0].get("member;range=1-1") is not None)
         self.assertTrue(len(res[0].get("member;range=1-1")) > 0)
         size = len(res[0].get("member;range=1-1"))
 
@@ -612,7 +610,6 @@ class ExtendedDirsyncTests(SimpleDirsyncTests):
         self.ldb_admin.add_remove_group_members("Administrators", [self.dirsync_user],
                                                 add_members_operation=True)
 
-
         res = self.ldb_admin.search(self.base_dn,
                                     expression="(name=Administrators)",
                                     controls=[control1])
@@ -658,6 +655,37 @@ class ExtendedDirsyncTests(SimpleDirsyncTests):
         self.assertEqual(res[0].get("member;range=1-1"), None)
         self.assertEqual(len(res[0].get("member;range=0-0")), 2)
 
+    def test_dirsync_extended_dn(self):
+        """Check that dirsync works together with the extended_dn control"""
+        # Let's search for members
+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
+        res = self.ldb_simple.search(self.base_dn,
+                                     expression="(name=Administrators)",
+                                     controls=["dirsync:1:1:1"])
+
+        self.assertTrue(len(res[0].get("member")) > 0)
+        size = len(res[0].get("member"))
+
+        resEX1 = self.ldb_simple.search(self.base_dn,
+                                        expression="(name=Administrators)",
+                                        controls=["dirsync:1:1:1","extended_dn:1:1"])
+        self.assertTrue(len(resEX1[0].get("member")) > 0)
+        sizeEX1 = len(resEX1[0].get("member"))
+        self.assertEqual(sizeEX1, size)
+        self.assertIn(res[0]["member"][0], resEX1[0]["member"][0])
+        self.assertIn(b"<GUID=", resEX1[0]["member"][0])
+        self.assertIn(b">;<SID=S-1-5-21-", resEX1[0]["member"][0])
+
+        resEX0 = self.ldb_simple.search(self.base_dn,
+                                        expression="(name=Administrators)",
+                                        controls=["dirsync:1:1:1","extended_dn:1:0"])
+        self.assertTrue(len(resEX0[0].get("member")) > 0)
+        sizeEX0 = len(resEX0[0].get("member"))
+        self.assertEqual(sizeEX0, size)
+        self.assertIn(res[0]["member"][0], resEX0[0]["member"][0])
+        self.assertIn(b"<GUID=", resEX0[0]["member"][0])
+        self.assertIn(b">;<SID=010500000000000515", resEX0[0]["member"][0])
+
     def test_dirsync_deleted_items(self):
         """Check that dirsync returnd deleted objects too"""
         # Let's create an OU
@@ -676,7 +704,7 @@ class ExtendedDirsyncTests(SimpleDirsyncTests):
             if str(e["name"]) == "testou3":
                 guid = str(ndr_unpack(misc.GUID, e.get("objectGUID")[0]))
 
-        self.assertTrue(guid != None)
+        self.assertTrue(guid is not None)
         ctl = str(res.controls[0]).split(":")
         ctl[1] = "1"
         ctl[2] = "1"