# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
+from __future__ import print_function
import sys
import time
import os
SCOPE_BASE,
Message,
FLAG_MOD_REPLACE,
- )
+)
+from samba.compat import cmp_fn
class DrsBaseTestCase(SambaToolCmdTest):
url_dc = samba.tests.env_get_var_value("DC2")
(self.ldb_dc2, self.info_dc2) = samba.tests.connect_samdb_ex(url_dc,
ldap_only=True)
+ self.test_ldb_dc = self.ldb_dc1
# cache some of RootDSE props
self.schema_dn = self.info_dc1["schemaNamingContext"][0]
def tearDown(self):
super(DrsBaseTestCase, self).tearDown()
+ def set_test_ldb_dc(self, ldb_dc):
+ """Sets which DC's LDB we perform operations on during the test"""
+ self.test_ldb_dc = ldb_dc
+
def _GUID_string(self, guid):
- return self.ldb_dc1.schema_format_value("objectGUID", guid)
+ return self.test_ldb_dc.schema_format_value("objectGUID", guid)
def _ldap_schemaUpdateNow(self, sam_db):
rec = {"dn": "",
# bin/samba-tool drs <drs_command> <cmdline_auth>
return ["drs", drs_command, cmdline_auth]
- def _net_drs_replicate(self, DC, fromDC, nc_dn=None, forced=True, local=False, full_sync=False):
+ def _net_drs_replicate(self, DC, fromDC, nc_dn=None, forced=True,
+ local=False, full_sync=False, single=False):
if nc_dn is None:
nc_dn = self.domain_dn
# make base command line
samba_tool_cmdline += ["--local"]
if full_sync:
samba_tool_cmdline += ["--full-sync"]
+ if single:
+ samba_tool_cmdline += ["--single-object"]
(result, out, err) = self.runsubcmd(*samba_tool_cmdline)
self.assertCmdSuccess(result, out, err)
- self.assertEquals(err,"","Shouldn't be any error messages")
+ self.assertEquals(err, "", "Shouldn't be any error messages")
def _enable_inbound_repl(self, DC):
# make base command line
samba_tool_cmd += [DC, "--dsa-option=-DISABLE_INBOUND_REPL"]
(result, out, err) = self.runsubcmd(*samba_tool_cmd)
self.assertCmdSuccess(result, out, err)
- self.assertEquals(err,"","Shouldn't be any error messages")
+ self.assertEquals(err, "", "Shouldn't be any error messages")
def _disable_inbound_repl(self, DC):
# make base command line
samba_tool_cmd += [DC, "--dsa-option=+DISABLE_INBOUND_REPL"]
(result, out, err) = self.runsubcmd(*samba_tool_cmd)
self.assertCmdSuccess(result, out, err)
- self.assertEquals(err,"","Shouldn't be any error messages")
+ self.assertEquals(err, "", "Shouldn't be any error messages")
def _enable_all_repl(self, DC):
self._enable_inbound_repl(DC)
samba_tool_cmd += [DC, "--dsa-option=-DISABLE_OUTBOUND_REPL"]
(result, out, err) = self.runsubcmd(*samba_tool_cmd)
self.assertCmdSuccess(result, out, err)
- self.assertEquals(err,"","Shouldn't be any error messages")
+ self.assertEquals(err, "", "Shouldn't be any error messages")
def _disable_all_repl(self, DC):
self._disable_inbound_repl(DC)
samba_tool_cmd += [DC, "--dsa-option=+DISABLE_OUTBOUND_REPL"]
(result, out, err) = self.runsubcmd(*samba_tool_cmd)
self.assertCmdSuccess(result, out, err)
- self.assertEquals(err,"","Shouldn't be any error messages")
+ self.assertEquals(err, "", "Shouldn't be any error messages")
def _get_highest_hwm_utdv(self, ldb_conn):
res = ldb_conn.search("", scope=ldb.SCOPE_BASE, attrs=["highestCommittedUSN"])
hwm = drsuapi.DsReplicaHighWaterMark()
- hwm.tmp_highest_usn = long(res[0]["highestCommittedUSN"][0])
+ hwm.tmp_highest_usn = int(res[0]["highestCommittedUSN"][0])
hwm.reserved_usn = 0
hwm.highest_usn = hwm.tmp_highest_usn
def _get_identifier(self, ldb_conn, dn):
res = ldb_conn.search(dn, scope=ldb.SCOPE_BASE,
- attrs=["objectGUID", "objectSid"])
+ attrs=["objectGUID", "objectSid"])
id = drsuapi.DsReplicaObjectIdentifier()
id.guid = ndr_unpack(misc.GUID, res[0]['objectGUID'][0])
if "objectSid" in res[0]:
return ctr6_links
+ def _get_ctr6_object_guids(self, ctr6):
+ """Returns all the object GUIDs in a GetNCChanges response"""
+ guid_list = []
+
+ obj = ctr6.first_object
+ for i in range(0, ctr6.object_count):
+ guid_list.append(str(obj.object.identifier.guid))
+ obj = obj.next_object
+
+ return guid_list
+
def _ctr6_debug(self, ctr6):
"""
Displays basic info contained in a DsGetNCChanges response.
next_object = ctr6.first_object
for i in range(0, ctr6.object_count):
- print("Obj %d: %s %s" %(i, next_object.object.identifier.dn[:22],
+ print("Obj %d: %s %s" % (i, next_object.object.identifier.dn[:25],
next_object.object.identifier.guid))
next_object = next_object.next_object
print("Linked Attributes: %d" % ctr6.linked_attributes_count)
- ctr6_links = self._get_ctr6_links(ctr6)
- for link in ctr6_links:
+ for lidx in range(0, ctr6.linked_attributes_count):
+ l = ctr6.linked_attributes[lidx]
+ try:
+ target = ndr_unpack(drsuapi.DsReplicaObjectIdentifier3,
+ l.value.blob)
+ except:
+ target = ndr_unpack(drsuapi.DsReplicaObjectIdentifier3Binary,
+ l.value.blob)
+
print("Link Tgt %s... <-- Src %s"
- %(link.targetDN[:22], link.identifier))
+ % (target.dn[:25], l.identifier.guid))
+ state = "Del"
+ if l.flags & drsuapi.DRSUAPI_DS_LINKED_ATTRIBUTE_FLAG_ACTIVE:
+ state = "Act"
+ print(" v%u %s changed %u" % (l.meta_data.version, state,
+ l.meta_data.originating_change_time))
- print("HWM: %d" %(ctr6.new_highwatermark.highest_usn))
- print("Tmp HWM: %d" %(ctr6.new_highwatermark.tmp_highest_usn))
- print("More data: %d" %(ctr6.more_data))
+ print("HWM: %d" % (ctr6.new_highwatermark.highest_usn))
+ print("Tmp HWM: %d" % (ctr6.new_highwatermark.tmp_highest_usn))
+ print("More data: %d" % (ctr6.more_data))
def _get_replication(self, replica_flags,
- drs_error=drsuapi.DRSUAPI_EXOP_ERR_NONE, drs=None, drs_handle=None,
- highwatermark=None, uptodateness_vector=None,
- more_flags=0, max_objects=133, exop=0,
- dest_dsa=drsuapi.DRSUAPI_DS_BIND_GUID_W2K3,
- source_dsa=None, invocation_id=None, nc_dn_str=None):
+ drs_error=drsuapi.DRSUAPI_EXOP_ERR_NONE, drs=None, drs_handle=None,
+ highwatermark=None, uptodateness_vector=None,
+ more_flags=0, max_objects=133, exop=0,
+ dest_dsa=drsuapi.DRSUAPI_DS_BIND_GUID_W2K3,
+ source_dsa=None, invocation_id=None, nc_dn_str=None):
"""
Builds a DsGetNCChanges request based on the information provided
and returns the response received from the DC.
"""
if source_dsa is None:
- source_dsa = self.ldb_dc1.get_ntds_GUID()
+ source_dsa = self.test_ldb_dc.get_ntds_GUID()
if invocation_id is None:
- invocation_id = self.ldb_dc1.get_invocation_id()
+ invocation_id = self.test_ldb_dc.get_invocation_id()
if nc_dn_str is None:
- nc_dn_str = self.ldb_dc1.domain_dn()
+ nc_dn_str = self.test_ldb_dc.domain_dn()
if highwatermark is None:
if self.default_hwm is None:
- (highwatermark, _) = self._get_highest_hwm_utdv(self.ldb_dc1)
+ (highwatermark, _) = self._get_highest_hwm_utdv(self.test_ldb_dc)
else:
highwatermark = self.default_hwm
if uptodateness_vector is not None:
uptodateness_vector_v1 = drsuapi.DsReplicaCursorCtrEx()
cursors = []
- for i in xrange(0, uptodateness_vector.count):
+ for i in range(0, uptodateness_vector.count):
c = uptodateness_vector.cursors[i]
c1 = drsuapi.DsReplicaCursor()
c1.source_dsa_invocation_id = c.source_dsa_invocation_id
req8.destination_dsa_guid = misc.GUID(dest_dsa) if dest_dsa else misc.GUID()
req8.source_dsa_invocation_id = misc.GUID(invocation_id)
req8.naming_context = drsuapi.DsReplicaObjectIdentifier()
- req8.naming_context.dn = unicode(nc_dn_str)
+ req8.naming_context.dn = str(nc_dn_str)
req8.highwatermark = drsuapi.DsReplicaHighWaterMark()
req8.highwatermark.tmp_highest_usn = 0
req8.highwatermark.reserved_usn = 0
req10.destination_dsa_guid = misc.GUID(dest_dsa) if dest_dsa else misc.GUID()
req10.source_dsa_invocation_id = misc.GUID(invocation_id)
req10.naming_context = drsuapi.DsReplicaObjectIdentifier()
- req10.naming_context.dn = unicode(nc_dn_str)
+ req10.naming_context.dn = str(nc_dn_str)
req10.highwatermark = drsuapi.DsReplicaHighWaterMark()
req10.highwatermark.tmp_highest_usn = 0
req10.highwatermark.reserved_usn = 0
"""See CompareLinks() in MS-DRSR section 4.1.10.5.17"""
if not isinstance(other, AbstractLink):
if verbose:
- print "AbstractLink.__internal_cmp__(%r, %r) => wrong type" % (self, other)
+ print("AbstractLink.__internal_cmp__(%r, %r) => wrong type" % (self, other))
return NotImplemented
- c = cmp(self.selfGUID_blob, other.selfGUID_blob)
+ c = cmp_fn(self.selfGUID_blob, other.selfGUID_blob)
if c != 0:
if verbose:
- print "AbstractLink.__internal_cmp__(%r, %r) => %d different identifier" % (self, other, c)
+ print("AbstractLink.__internal_cmp__(%r, %r) => %d different identifier" % (self, other, c))
return c
c = other.attid - self.attid
if c != 0:
if verbose:
- print "AbstractLink.__internal_cmp__(%r, %r) => %d different attid" % (self, other, c)
+ print("AbstractLink.__internal_cmp__(%r, %r) => %d different attid" % (self, other, c))
return c
self_active = self.flags & drsuapi.DRSUAPI_DS_LINKED_ATTRIBUTE_FLAG_ACTIVE
c = self_active - other_active
if c != 0:
if verbose:
- print "AbstractLink.__internal_cmp__(%r, %r) => %d different FLAG_ACTIVE" % (self, other, c)
+ print("AbstractLink.__internal_cmp__(%r, %r) => %d different FLAG_ACTIVE" % (self, other, c))
return c
- c = cmp(self.targetGUID_blob, other.targetGUID_blob)
+ c = cmp_fn(self.targetGUID_blob, other.targetGUID_blob)
if c != 0:
if verbose:
- print "AbstractLink.__internal_cmp__(%r, %r) => %d different target" % (self, other, c)
+ print("AbstractLink.__internal_cmp__(%r, %r) => %d different target" % (self, other, c))
return c
c = self.flags - other.flags
if c != 0:
if verbose:
- print "AbstractLink.__internal_cmp__(%r, %r) => %d different flags" % (self, other, c)
+ print("AbstractLink.__internal_cmp__(%r, %r) => %d different flags" % (self, other, c))
return c
return 0
git.samba.org / bbaumbach / samba-autobuild / .git / blobdiff