# PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN repl_move -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
#
+from __future__ import print_function
import time
import uuid
+import samba.tests
from samba.ndr import ndr_unpack
from samba.dcerpc import drsblobs
from ldb import (
SCOPE_BASE,
SCOPE_SUBTREE,
- )
+)
-import drs_base, ldb
+import drs_base
+import ldb
from samba.dcerpc.drsuapi import *
+
class DrsMoveObjectTestCase(drs_base.DrsBaseTestCase):
def _ds_bind(self, server_name):
def setUp(self):
super(DrsMoveObjectTestCase, self).setUp()
+ # disable automatic replication temporary
+ self._disable_all_repl(self.dnsname_dc1)
+ self._disable_all_repl(self.dnsname_dc2)
+
# make sure DCs are synchronized before the test
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
+
+ self.top_ou = samba.tests.create_test_ou(self.ldb_dc1,
+ "replica_move")
+
self.ou1_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU1")
- self.ou1_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou1_dn.add_base(self.top_ou)
ou1 = {}
ou1["dn"] = self.ou1_dn
ou1["objectclass"] = "organizationalUnit"
self.ldb_dc1.add(ou1)
self.ou2_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU2")
- self.ou2_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou2_dn.add_base(self.top_ou)
ou2 = {}
ou2["dn"] = self.ou2_dn
ou2["objectclass"] = "organizationalUnit"
ou2["ou"] = self.ou2_dn.get_component_value(0)
self.ldb_dc1.add(ou2)
- # disable automatic replication temporary
- self._disable_inbound_repl(self.dnsname_dc1)
- self._disable_inbound_repl(self.dnsname_dc2)
-
# trigger replication from DC1 to DC2
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
self.dc1_guid = self.ldb_dc1.get_invocation_id()
def tearDown(self):
try:
- self.ldb_dc1.delete(self.ou1_dn, ["tree_delete:1"])
- except ldb.LdbError as (enum, string):
+ self.ldb_dc1.delete(self.top_ou, ["tree_delete:1"])
+ except ldb.LdbError as e:
+ (enum, string) = e.args
if enum == ldb.ERR_NO_SUCH_OBJECT:
pass
- self.ldb_dc1.delete(self.ou2_dn, ["tree_delete:1"])
- self._enable_inbound_repl(self.dnsname_dc1)
- self._enable_inbound_repl(self.dnsname_dc2)
+ self._enable_all_repl(self.dnsname_dc1)
+ self._enable_all_repl(self.dnsname_dc2)
super(DrsMoveObjectTestCase, self).tearDown()
def _make_username(self):
return "DrsMoveU_" + time.strftime("%s", time.gmtime())
def _check_metadata(self, user_dn, sam_ldb, drs, metadata, expected):
- repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob, str(metadata[0]))
+ repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob, metadata[0])
self.assertEqual(len(repl.ctr.array), len(expected))
req8.highwatermark.reserved_usn = 0
req8.highwatermark.highest_usn = 0
req8.uptodateness_vector = None
- req8.replica_flags = 0
+ req8.replica_flags = DRSUAPI_DRS_SYNC_FORCED
req8.max_object_count = 1
req8.max_ndr_size = 402116
req8.extended_op = DRSUAPI_EXOP_REPL_OBJ
for o in drs_ctr.first_object.meta_data_ctr.meta_data:
i = 0
drs_attid = drs_ctr.first_object.object.attribute_ctr.attributes[att_idx]
- e = expected[i];
+ e = expected[i]
(attid, orig_dsa, version) = e
# Skip the RDN from the expected set, it is not sent over DRS
- if (user_dn.get_rdn_name().upper() == "CN" \
+ if (user_dn.get_rdn_name().upper() == "CN"
and attid == DRSUAPI_ATTID_cn) \
- or (user_dn.get_rdn_name().upper() == "OU" \
+ or (user_dn.get_rdn_name().upper() == "OU"
and attid == DRSUAPI_ATTID_ou):
i = i + 1
- e = expected[i];
+ e = expected[i]
(attid, orig_dsa, version) = e
self.assertEquals(attid, drs_attid.attid,
return user_cur
-
def test_ReplicateMoveObject1(self):
"""Verifies how a moved container with a user inside is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
initial_metadata = [
(DRSUAPI_ATTID_objectClass, self.dc1_guid, 1),
(DRSUAPI_ATTID_cn, self.dc1_guid, 1),
is_deleted=False,
expected_metadata=modified_metadata)
-
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
expected_metadata=deleted_modified_metadata_dc1)
self.assertFalse("description" in user_cur)
-
def test_ReplicateMoveObject2(self):
"""Verifies how a moved container with a user inside is not
replicated between two DCs as no replication is triggered
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
initial_metadata = [
(DRSUAPI_ATTID_objectClass, self.dc1_guid, 1),
(DRSUAPI_ATTID_cn, self.dc1_guid, 1),
is_deleted=True,
expected_metadata=deleted_metadata_dc1)
-
def test_ReplicateMoveObject3(self):
"""Verifies how a moved container with a user inside is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
initial_metadata = [
(DRSUAPI_ATTID_objectClass, self.dc1_guid, 1),
(DRSUAPI_ATTID_cn, self.dc1_guid, 1),
obj_orig=user_orig, is_deleted=False,
expected_metadata=initial_metadata)
-
new_dn = ldb.Dn(self.ldb_dc1, "CN=%s" % username)
new_dn.add_base(self.ou2_dn)
self.ldb_dc1.rename(user_dn, new_dn)
is_deleted=True,
expected_metadata=deleted_metadata_dc2)
-
def test_ReplicateMoveObject3b(self):
"""Verifies how a moved container with a user inside is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
initial_metadata = [
(DRSUAPI_ATTID_objectClass, self.dc1_guid, 1),
(DRSUAPI_ATTID_cn, self.dc1_guid, 1),
obj_orig=user_orig, is_deleted=False,
expected_metadata=initial_metadata)
-
new_dn = ldb.Dn(self.ldb_dc1, "CN=%s" % username)
new_dn.add_base(self.ou2_dn)
self.ldb_dc1.rename(user_dn, new_dn)
is_deleted=True,
expected_metadata=deleted_metadata_dc2)
-
def test_ReplicateMoveObject4(self):
"""Verifies how a moved container with a user inside is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
initial_metadata = [
(DRSUAPI_ATTID_objectClass, self.dc1_guid, 1),
(DRSUAPI_ATTID_cn, self.dc1_guid, 1),
(DRSUAPI_ATTID_objectCategory, self.dc1_guid, 2),
(DRSUAPI_ATTID_isRecycled, self.dc1_guid, 1)]
-
# check user info on DC1 - should be deleted user
user_cur = self._check_obj(sam_ldb=self.ldb_dc1, drs=self.drs_dc1,
obj_orig=user_moved_orig,
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_cur = self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
new_dn = ldb.Dn(self.ldb_dc1, "CN=%s" % username)
user_cur = self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_moved_orig, is_deleted=True)
self.assertFalse("description" in user_cur)
-
def test_ReplicateMoveObject6(self):
"""Verifies how a moved container is replicated between two DCs.
This test should verify that:
ou_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=False)
# trigger replication from DC1 to DC2
ou_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=ou_moved_orig, is_deleted=True)
self.assertFalse("description" in ou_cur)
-
def test_ReplicateMoveObject7(self):
"""Verifies how a moved container is replicated between two DCs.
This test should verify that:
ou_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=False)
# trigger replication from DC1 to DC2
ou_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=ou_moved_orig, is_deleted=True)
self.assertFalse("description" in ou_cur)
-
def test_ReplicateMoveObject8(self):
"""Verifies how a moved container is replicated between two DCs.
This test should verify that:
ou_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=False)
# trigger replication from DC1 to DC2
ou_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=ou_moved_orig, is_deleted=True)
self.assertFalse("description" in ou_cur)
-
def test_ReplicateMoveObject9(self):
"""Verifies how a moved container is replicated between two DCs.
This test should verify that:
ou_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=False)
# trigger replication from DC1 to DC2
ou_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=ou_moved_orig, is_deleted=True)
self.assertFalse("description" in ou_cur)
-
def test_ReplicateMoveObject10(self):
"""Verifies how a moved container is replicated between two DCs.
This test should verify that:
ou_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=False)
# trigger replication from DC1 to DC2
ou_cur = self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=True)
self.assertFalse("description" in ou_cur)
-
def test_ReplicateMoveObject11(self):
"""Verifies how a moved container is replicated between two DCs.
This test should verify that:
ou_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (self.ou1_dn, self._GUID_string(ou_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=ou_orig, is_deleted=False)
# trigger replication from DC1 to DC2
self.assertFalse("description" in ou_cur)
-
class DrsMoveBetweenTreeOfObjectTestCase(drs_base.DrsBaseTestCase):
def setUp(self):
super(DrsMoveBetweenTreeOfObjectTestCase, self).setUp()
+ # disable automatic replication temporary
+ self._disable_all_repl(self.dnsname_dc1)
+ self._disable_all_repl(self.dnsname_dc2)
+
+ self.top_ou = samba.tests.create_test_ou(self.ldb_dc1,
+ "replica_move")
+
# make sure DCs are synchronized before the test
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
+
self.ou1_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU1")
- self.ou1_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou1_dn.add_base(self.top_ou)
self.ou1 = {}
self.ou1["dn"] = self.ou1_dn
self.ou1["objectclass"] = "organizationalUnit"
self.ou1["ou"] = self.ou1_dn.get_component_value(0)
self.ou2_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU2,OU=DrsOU1")
- self.ou2_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou2_dn.add_base(self.top_ou)
self.ou2 = {}
self.ou2["dn"] = self.ou2_dn
self.ou2["objectclass"] = "organizationalUnit"
self.ou2["ou"] = self.ou2_dn.get_component_value(0)
self.ou2b_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU2B,OU=DrsOU1")
- self.ou2b_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou2b_dn.add_base(self.top_ou)
self.ou2b = {}
self.ou2b["dn"] = self.ou2b_dn
self.ou2b["objectclass"] = "organizationalUnit"
self.ou2b["ou"] = self.ou2b_dn.get_component_value(0)
self.ou2c_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU2C,OU=DrsOU1")
- self.ou2c_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou2c_dn.add_base(self.top_ou)
self.ou3_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU3,OU=DrsOU2,OU=DrsOU1")
- self.ou3_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou3_dn.add_base(self.top_ou)
self.ou3 = {}
self.ou3["dn"] = self.ou3_dn
self.ou3["objectclass"] = "organizationalUnit"
self.ou3["ou"] = self.ou3_dn.get_component_value(0)
self.ou4_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU4,OU=DrsOU3,OU=DrsOU2,OU=DrsOU1")
- self.ou4_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou4_dn.add_base(self.top_ou)
self.ou4 = {}
self.ou4["dn"] = self.ou4_dn
self.ou4["objectclass"] = "organizationalUnit"
self.ou4["ou"] = self.ou4_dn.get_component_value(0)
self.ou5_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU5,OU=DrsOU4,OU=DrsOU3,OU=DrsOU2,OU=DrsOU1")
- self.ou5_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou5_dn.add_base(self.top_ou)
self.ou5 = {}
self.ou5["dn"] = self.ou5_dn
self.ou5["objectclass"] = "organizationalUnit"
self.ou5["ou"] = self.ou5_dn.get_component_value(0)
self.ou6_dn = ldb.Dn(self.ldb_dc1, "OU=DrsOU6,OU=DrsOU5,OU=DrsOU4,OU=DrsOU3,OU=DrsOU2,OU=DrsOU1")
- self.ou6_dn.add_base(self.ldb_dc1.get_default_basedn())
+ self.ou6_dn.add_base(self.top_ou)
self.ou6 = {}
self.ou6["dn"] = self.ou6_dn
self.ou6["objectclass"] = "organizationalUnit"
self.ou6["ou"] = self.ou6_dn.get_component_value(0)
- # disable automatic replication temporary
- self._disable_inbound_repl(self.dnsname_dc1)
- self._disable_inbound_repl(self.dnsname_dc2)
-
-
def tearDown(self):
- self.ldb_dc1.delete(self.ou1_dn, ["tree_delete:1"])
- self._enable_inbound_repl(self.dnsname_dc1)
- self._enable_inbound_repl(self.dnsname_dc2)
+ self.ldb_dc1.delete(self.top_ou, ["tree_delete:1"])
+ self._enable_all_repl(self.dnsname_dc1)
+ self._enable_all_repl(self.dnsname_dc2)
super(DrsMoveBetweenTreeOfObjectTestCase, self).tearDown()
def _make_username(self):
return user_cur
-
def test_ReplicateMoveInTree1(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
self.ldb_dc1.add(self.ou2)
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
-
def test_ReplicateMoveInTree2(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
self.ldb_dc1.add(self.ou2)
user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_moved_orig, is_deleted=True)
self.assertFalse("description" in user_cur)
-
def test_ReplicateMoveInTree3(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
self.ldb_dc1.add(self.ou2)
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
-
def test_ReplicateMoveInTree3b(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
self.ldb_dc1.add(self.ou2)
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
-
def test_ReplicateMoveInTree4(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
user_dn = ldb_res[0]["dn"]
# check user info on DC1
- print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+ print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
self.ldb_dc1.add(self.ou2)
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
-
def test_ReplicateAddInOU(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
-
def test_ReplicateAddInMovedOU(self):
"""Verifies how an object is replicated between two DCs.
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
# trigger replication from DC1 to DC2, for cleanup
self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
-
def test_ReplicateAddInConflictOU_time(self):
"""Verifies how an object is replicated between two DCs, when created in an ambigious location
This test should verify that:
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
# create user on DC1
self.ldb_dc1.newuser(username=username,
- userou="ou=%s" % self.ou1_dn.get_component_value(0),
+ userou="ou=%s,ou=%s"
+ % (self.ou1_dn.get_component_value(0),
+ self.top_ou.get_component_value(0)),
password=None, setpassword=False)
ldb_res = self.ldb_dc1.search(base=self.ou1_dn,
scope=SCOPE_SUBTREE,
git.samba.org / nivanova / samba-autobuild / .git / blobdiff