sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2025-02-26 21:57:41 +03:00
Code

s4-selftest/drs Allow some DRS tests to operate against an IP

Browse Source 
This is not comprehensive, but makes some manual test runs easier by
avoiding the need for DNS names to resolve.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=10635

Signed-off-by: Andrew Bartlett <abartlet@samba.org>
Reviewed-by: Stefan Metzmacher <metze@samba.org>
(cherry picked from commit a150a2dcb1fc7fc7f606838de17ad4d3e6072bda)
This commit is contained in:
Andrew Bartlett 2022-12-02 11:42:55 +13:00 committed by Jule Anger
parent a81be07598
commit 2cb965046b
4 changed files with 18 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source4/torture/drs/python/drs_base.py
View File

@ -492,8 +492,11 @@ class DrsBaseTestCase(SambaToolCmdTest):
return req10
def _ds_bind(self, server_name, creds=None):
binding_str = "ncacn_ip_tcp:%s[seal]" % server_name
def _ds_bind(self, server_name, creds=None, ip=None):
if ip is None:
binding_str = f"ncacn_ip_tcp:{server_name}[seal]"
else:
binding_str = f"ncacn_ip_tcp:{ip}[seal,target_hostname={server_name}]"
if creds is None:
creds = self.get_credentials()

20
source4/torture/drs/python/getnc_exop.py
View File

@ -85,7 +85,7 @@ class DrsReplicaSyncTestCase(drs_base.DrsBaseTestCase):
self.ldb_dc1.add({
"dn": self.ou,
"objectclass": "organizationalUnit"})
(self.drs, self.drs_handle) = self._ds_bind(self.dnsname_dc1)
(self.drs, self.drs_handle) = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
(self.default_hwm, self.default_utdv) = self._get_highest_hwm_utdv(self.ldb_dc1)
def tearDown(self):
@ -251,7 +251,7 @@ class DrsReplicaSyncTestCase(drs_base.DrsBaseTestCase):
nc_guid=misc.GUID("c2d2f745-1610-4e93-964b-d4ba73eb32f8"),
exop=drsuapi.DRSUAPI_EXOP_NONE)
(drs, drs_handle) = self._ds_bind(fsmo_owner["dns_name"])
(drs, drs_handle) = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
try:
(level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
except WERRORError as e1:
@ -269,7 +269,7 @@ class DrsReplicaSyncTestCase(drs_base.DrsBaseTestCase):
nc_guid=misc.GUID("c2d2f745-1610-4e93-964b-d4ba73eb32f8"),
exop=drsuapi.DRSUAPI_EXOP_REPL_OBJ)
(drs, drs_handle) = self._ds_bind(fsmo_owner["dns_name"])
(drs, drs_handle) = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
try:
(level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
except WERRORError as e1:
@ -702,7 +702,7 @@ class DrsReplicaPrefixMapTestCase(drs_base.DrsBaseTestCase):
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
req8 = self._exop_req8(dest_dsa=None,
invocation_id=dc_guid_1,
@ -721,7 +721,7 @@ class DrsReplicaPrefixMapTestCase(drs_base.DrsBaseTestCase):
partial_attribute_set = self.get_partial_attribute_set([99999])
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
try:
pfm = self._samdb_fetch_pfm_and_schi()
@ -754,7 +754,7 @@ class DrsReplicaPrefixMapTestCase(drs_base.DrsBaseTestCase):
partial_attribute_set = self.get_partial_attribute_set([drsuapi.DRSUAPI_ATTID_unicodePwd])
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
try:
pfm = self._samdb_fetch_pfm_and_schi()
@ -828,7 +828,7 @@ class DrsReplicaPrefixMapTestCase(drs_base.DrsBaseTestCase):
partial_attribute_set = self.get_partial_attribute_set([drsuapi.DRSUAPI_ATTID_name])
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
try:
pfm = self._samdb_fetch_pfm_and_schi()
@ -901,7 +901,7 @@ class DrsReplicaPrefixMapTestCase(drs_base.DrsBaseTestCase):
partial_attribute_set_ex = self.get_partial_attribute_set([drsuapi.DRSUAPI_ATTID_unicodePwd])
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
try:
pfm = self._samdb_fetch_pfm_and_schi()
@ -1117,7 +1117,7 @@ class DrsReplicaSyncSortTestCase(drs_base.DrsBaseTestCase):
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
req8 = self._exop_req8(dest_dsa=None,
invocation_id=dc_guid_1,
@ -1177,7 +1177,7 @@ class DrsReplicaSyncSortTestCase(drs_base.DrsBaseTestCase):
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
# Make sure the max objects count is high enough
req8 = self._exop_req8(dest_dsa=None,

11
source4/torture/drs/python/repl_move.py
View File

@ -48,13 +48,6 @@ from samba.dcerpc.drsuapi import *
class DrsMoveObjectTestCase(drs_base.DrsBaseTestCase):
def _ds_bind(self, server_name):
binding_str = "ncacn_ip_tcp:%s[print,seal]" % server_name
drs = drsuapi(binding_str, self.get_loadparm(), self.get_credentials())
(drs_handle, supported_extensions) = drs_DsBind(drs)
return (drs, drs_handle)
def setUp(self):
super(DrsMoveObjectTestCase, self).setUp()
# disable automatic replication temporary
@ -89,8 +82,8 @@ class DrsMoveObjectTestCase(drs_base.DrsBaseTestCase):
self.dc1_guid = self.ldb_dc1.get_invocation_id()
self.dc2_guid = self.ldb_dc2.get_invocation_id()
self.drs_dc1 = self._ds_bind(self.dnsname_dc1)
self.drs_dc2 = self._ds_bind(self.dnsname_dc2)
self.drs_dc1 = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
self.drs_dc2 = self._ds_bind(self.dnsname_dc2, ip=self.url_dc2)
def tearDown(self):
try:

9
source4/torture/drs/python/repl_schema.py
View File

@ -52,13 +52,6 @@ class DrsReplSchemaTestCase(drs_base.DrsBaseTestCase):
# current Class or Attribute object id
obj_id = 0
def _ds_bind(self, server_name):
binding_str = "ncacn_ip_tcp:%s[seal]" % server_name
drs = drsuapi.drsuapi(binding_str, self.get_loadparm(), self.get_credentials())
(drs_handle, supported_extensions) = drs_DsBind(drs)
return (drs, drs_handle)
def _exop_req8(self, dest_dsa, invocation_id, nc_dn_str, exop,
replica_flags=0, max_objects=0):
req8 = drsuapi.DsGetNCChangesRequest8()
@ -281,7 +274,7 @@ class DrsReplSchemaTestCase(drs_base.DrsBaseTestCase):
dc_guid_1 = self.ldb_dc1.get_invocation_id()
drs, drs_handle = self._ds_bind(self.dnsname_dc1)
drs, drs_handle = self._ds_bind(self.dnsname_dc1, ip=self.url_dc1)
req8 = self._exop_req8(dest_dsa=None,
invocation_id=dc_guid_1,