expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
if hasattr(self, 'disallowDNFilter') and \
hasattr(self, 'IDX') and \
- (hasattr(self, 'IDXGUID') or \
+ (hasattr(self, 'IDXGUID') or
((hasattr(self, 'IDXONE') == False and hasattr(self, 'IDX')))):
self.assertEqual(len(res11), 0)
else:
check_duplicates=True):
'''handle a orphaned backlink value'''
if check_duplicates is True and self.has_duplicate_links(target_dn, forward_attr, forward_syntax):
- self.report("WARNING: Keep orphaned backlink attribute " + \
+ self.report("WARNING: Keep orphaned backlink attribute " +
"'%s' in '%s' for link '%s' in '%s'" % (
backlink_attr, obj_dn, forward_attr, target_dn))
return
return (missing_forward_links, error_count)
if "sortedLinks" in self.compatibleFeatures:
- self.report("Not checking for missing forward links because the db " + \
+ self.report("Not checking for missing forward links because the db " +
"has the sortedLinks feature")
return (missing_forward_links, error_count)
fn = getattr(traffic_packets, fn_name)
except AttributeError as e:
- print("Conversation(%s) Missing handler %s" % \
+ print("Conversation(%s) Missing handler %s" %
(conversation.conversation_id, fn_name),
file=sys.stderr)
return
def ch_minPwdAge(self, val):
old_val = self.ldb.get_minPwdAge()
- self.logger.info('KDC Minimum Password age was changed from %s to %s' \
+ self.logger.info('KDC Minimum Password age was changed from %s to %s'
% (old_val, val))
self.gp_db.store(str(self), self.attribute, str(old_val))
self.ldb.set_minPwdAge(val)
def ch_maxPwdAge(self, val):
old_val = self.ldb.get_maxPwdAge()
- self.logger.info('KDC Maximum Password age was changed from %s to %s' \
+ self.logger.info('KDC Maximum Password age was changed from %s to %s'
% (old_val, val))
self.gp_db.store(str(self), self.attribute, str(old_val))
self.ldb.set_maxPwdAge(val)
def ch_minPwdLength(self, val):
old_val = self.ldb.get_minPwdLength()
self.logger.info(
- 'KDC Minimum Password length was changed from %s to %s' \
+ 'KDC Minimum Password length was changed from %s to %s'
% (old_val, val))
self.gp_db.store(str(self), self.attribute, str(old_val))
self.ldb.set_minPwdLength(val)
def ch_pwdProperties(self, val):
old_val = self.ldb.get_pwdProperties()
- self.logger.info('KDC Password Properties were changed from %s to %s' \
+ self.logger.info('KDC Password Properties were changed from %s to %s'
% (old_val, val))
self.gp_db.store(str(self), self.attribute, str(old_val))
self.ldb.set_pwdProperties(val)
try:
check_refresh_gpo_list(dc_hostname, lp, creds, gpos)
except:
- logger.error('Failed downloading gpt cache from \'%s\' using SMB' \
+ logger.error('Failed downloading gpt cache from \'%s\' using SMB'
% dc_hostname)
return
try:
ext.parse(ext.list(path), test_ldb, gp_db, lp)
except Exception as e:
- logger.error('Failed to parse gpo %s for extension %s' % \
+ logger.error('Failed to parse gpo %s for extension %s' %
(guid, str(ext)))
logger.error('Message was: ' + str(e))
store.cancel()
cname_target = "%s.%s" % (name, zone)
IPs = samba.interface_ips(ctx.lp, ctx.force_all_ips)
- ctx.logger.info("Adding %d remote DNS records for %s.%s" % \
+ ctx.logger.info("Adding %d remote DNS records for %s.%s" %
(len(IPs), name, zone))
binding_options = "sign"
# This is verbose, but it is the safe, escape-proof way
# to add a base and add an arbitrary RDN.
if dn.add_base(samdb.get_config_basedn()) == False:
- raise DemoteException("Failed constructing DN %s by adding base %s" \
+ raise DemoteException("Failed constructing DN %s by adding base %s"
% (dn, samdb.get_config_basedn()))
if dn.add_child("CN=X") == False:
- raise DemoteException("Failed constructing DN %s by adding child CN=X"\
+ raise DemoteException("Failed constructing DN %s by adding child CN=X"
% (dn))
dn.set_component(0, "CN", dc_name)
try:
# to add a base and add an arbitrary RDN.
dn = ldb.Dn(samdb, s)
if dn.add_base(samdb.get_default_basedn()) == False:
- raise DemoteException("Failed constructing DN %s by adding base" % \
+ raise DemoteException("Failed constructing DN %s by adding base" %
(dn, samdb.get_default_basedn()))
if dn.add_child("CN=X") == False:
raise DemoteException("Failed constructing DN %s by adding child "
a_recs = [r for r in a_recs if not a_rec_to_remove(r)]
if len(a_recs) != orig_num_recs:
- logger.info("updating %s keeping %d values, removing %s values" % \
+ logger.info("updating %s keeping %d values, removing %s values" %
(a_name, len(a_recs), orig_num_recs - len(a_recs)))
samdb.dns_replace(a_name, a_recs)
for v in orig_values if not to_remove(v)]
if len(values) != len(orig_values):
- logger.info("updating %s keeping %d values, removing %s values" \
+ logger.info("updating %s keeping %d values, removing %s values"
% (record.dn, len(values),
len(orig_values) - len(values)))
dns_admin = str(ndr_unpack(security.dom_sid, res[0]['objectSid'][0]))
- packed_sd = descriptor.sddl2binary("O:SYG:BA" \
- "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" \
- "(A;;CC;;;AU)" \
- "(A;;RPLCLORC;;;WD)" \
- "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)" \
+ packed_sd = descriptor.sddl2binary("O:SYG:BA"
+ "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)"
+ "(A;;CC;;;AU)"
+ "(A;;RPLCLORC;;;WD)"
+ "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)"
"(A;CI;RPWPCRCCDCLCRCWOWDSDDTSW;;;ED)",
domain_sid, {"DnsAdmins": dns_admin})
expected_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_sd))
dns_admin = str(ndr_unpack(security.dom_sid, res[0]['objectSid'][0]))
- packed_sd = descriptor.sddl2binary("O:DAG:DA" \
- "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" \
- "(A;;CC;;;AU)" \
- "(A;;RPLCLORC;;;WD)" \
- "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)" \
+ packed_sd = descriptor.sddl2binary("O:DAG:DA"
+ "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)"
+ "(A;;CC;;;AU)"
+ "(A;;RPLCLORC;;;WD)"
+ "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)"
"(A;CI;RPWPCRCCDCLCRCWOWDSDDTSW;;;ED)",
domain_sid, {"DnsAdmins": dns_admin})
expected_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_sd))
dns_admin = str(ndr_unpack(security.dom_sid, res[0]['objectSid'][0]))
- packed_sd = descriptor.sddl2binary("O:DAG:DA" \
- "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" \
- "(A;;CC;;;AU)" \
- "(A;;RPLCLORC;;;WD)" \
- "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)" \
+ packed_sd = descriptor.sddl2binary("O:DAG:DA"
+ "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)"
+ "(A;;CC;;;AU)"
+ "(A;;RPLCLORC;;;WD)"
+ "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)"
"(A;CI;RPWPCRCCDCLCRCWOWDSDDTSW;;;ED)",
domain_sid, {"DnsAdmins": dns_admin})
expected_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_sd))
after = 'addom.samba.example.com/Policies/' \
'{31B2F340-016D-11D2-945F-00C04FB984F9}/GPT.INI'
result = check_safe_path(before)
- self.assertEquals(result, after, 'check_safe_path() didn\'t' \
+ self.assertEquals(result, after, 'check_safe_path() didn\'t'
' correctly convert \\ to /')
def test_gpt_ext_register(self):
self.creds_string)
try:
self.assertCmdSuccess(result, out, err,
- "Failed to query for a record" \
+ "Failed to query for a record"
"which had DNS_RANK_NONE.")
self.assertTrue("testrecord" in out and record_str in out,
- "Query for a record which had DNS_RANK_NONE" \
+ "Query for a record which had DNS_RANK_NONE"
"succeeded but produced no resulting records.")
except AssertionError as e:
# Windows produces no resulting records
self.zone, "testrecord", record_type_str,
record_str, self.creds_string)
try:
- self.assertCmdFail(result, "Successfully added duplicate record" \
+ self.assertCmdFail(result, "Successfully added duplicate record"
"of one which had DNS_RANK_NONE.")
except AssertionError as e:
errors.append(e)
self.zone, "testrecord", record_type_str,
record_str, self.creds_string)
try:
- self.assertCmdSuccess(result, out, err, "Failed to delete record" \
+ self.assertCmdSuccess(result, out, err, "Failed to delete record"
"which had DNS_RANK_NONE.")
except AssertionError as e:
errors.append(e)
self.zone, "testrecord",
record_type_str, self.creds_string)
try:
- self.assertCmdFail(result, "Successfully queried for deleted record" \
+ self.assertCmdFail(result, "Successfully queried for deleted record"
"which had DNS_RANK_NONE.")
except AssertionError as e:
errors.append(e)
self.zone, "testrecord",
dnstype, record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add" \
+ self.assertCmdSuccess(result, out, err, "Failed to add"
"record %s with type %s."
% (record, dnstype))
self.zone, "testrecord",
dnstype,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to query" \
+ self.assertCmdSuccess(result, out, err, "Failed to query"
"record %s with qualifier %s."
% (record, dnstype))
self.zone, "testrecord",
dnstype, record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to remove" \
+ self.assertCmdSuccess(result, out, err, "Failed to remove"
"record %s with type %s."
% (record, dnstype))
except AssertionError as e:
if num_failures > 0:
for msg in failure_msgs:
print(msg)
- self.fail("Failed to accept valid commands. %d total failures." \
+ self.fail("Failed to accept valid commands. %d total failures."
"Errors above." % num_failures)
def test_reject_invalid_commands(self):
self.zone, "testrecord",
dnstype, record,
self.creds_string)
- self.assertCmdFail(result, "Successfully added invalid" \
+ self.assertCmdFail(result, "Successfully added invalid"
"record '%s' of type '%s'."
% (record, dnstype))
except AssertionError as e:
self.zone, "testrecord",
dnstype, record,
self.creds_string)
- self.assertCmdFail(result, "Successfully deleted invalid" \
- "record '%s' of type '%s' which" \
+ self.assertCmdFail(result, "Successfully deleted invalid"
+ "record '%s' of type '%s' which"
"shouldn't exist." % (record, dnstype))
except AssertionError as e:
num_failures = num_failures + 1
self.zone, "testrecord",
dnstype, good_record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add " \
+ self.assertCmdSuccess(result, out, err, "Failed to add "
"record '%s' with type %s."
% (record, dnstype))
dnstype, good_record,
bad_record,
self.creds_string)
- self.assertCmdFail(result, "Successfully updated valid " \
- "record '%s' of type '%s' to invalid " \
+ self.assertCmdFail(result, "Successfully updated valid "
+ "record '%s' of type '%s' to invalid "
"record '%s' of the same type."
% (good_record, dnstype, bad_record))
self.zone, "testrecord",
dnstype, good_record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Could not delete " \
+ self.assertCmdSuccess(result, out, err, "Could not delete "
"valid record '%s' of type '%s'."
% (good_record, dnstype))
except AssertionError as e:
if num_failures > 0:
for msg in failure_msgs:
print(msg)
- self.fail("Failed to reject invalid commands. %d total failures. " \
+ self.fail("Failed to reject invalid commands. %d total failures. "
"Errors above." % num_failures)
def test_update_invalid_type(self):
self.zone, "testrecord",
dnstype1, record1,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add " \
+ self.assertCmdSuccess(result, out, err, "Failed to add "
"record %s with type %s."
% (record1, dnstype1))
self.zone, "testrecord",
dnstype1, record1,
record2, self.creds_string)
- self.assertCmdFail(result, "Successfully updated record '%s' " \
- "to '%s', even though the latter is of " \
+ self.assertCmdFail(result, "Successfully updated record '%s' "
+ "to '%s', even though the latter is of "
"type '%s' where '%s' was expected."
% (record1, record2, dnstype2, dnstype1))
self.zone, "testrecord",
dnstype2, record1, record2,
self.creds_string)
- self.assertCmdFail(result, "Successfully updated record " \
- "'%s' to '%s', even though the former " \
+ self.assertCmdFail(result, "Successfully updated record "
+ "'%s' to '%s', even though the former "
"is of type '%s' where '%s' was expected."
% (record1, record2, dnstype1, dnstype2))
self.zone, "testrecord",
dnstype, record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add " \
+ self.assertCmdSuccess(result, out, err, "Failed to add "
"record %s with type %s."
% (record, dnstype))
self.zone, "testrecord",
dnstype, record, record,
self.creds_string)
- self.assertCmdFail(result, "Successfully updated record " \
+ self.assertCmdFail(result, "Successfully updated record "
"'%s' to be exactly the same." % record)
result, out, err = self.runsubcmd("dns", "delete",
self.zone, "testrecord",
dnstype, record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Could not delete " \
+ self.assertCmdSuccess(result, out, err, "Could not delete "
"valid record '%s' of type '%s'."
% (record, dnstype))
self.zone, "testrecord",
"SRV", record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add " \
+ self.assertCmdSuccess(result, out, err, "Failed to add "
"record %s with type 'SRV'." % record)
split = record.split(' ')
self.zone, "testrecord",
"SRV", record,
new_record, self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to update record " \
+ self.assertCmdSuccess(result, out, err, "Failed to update record "
"'%s' of type '%s' to '%s'."
% (record, "SRV", new_record))
os.environ["SERVER"],
self.zone, "testrecord",
"SRV", self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to query for " \
+ self.assertCmdSuccess(result, out, err, "Failed to query for "
"record '%s' of type '%s'."
% (new_record, "SRV"))
self.zone, "testrecord",
"SRV", new_record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Could not delete " \
+ self.assertCmdSuccess(result, out, err, "Could not delete "
"valid record '%s' of type '%s'."
% (new_record, "SRV"))
self.zone, "testrecord",
dnstype, used_record,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add record %s " \
+ self.assertCmdSuccess(result, out, err, "Failed to add record %s "
"with type %s." % (used_record, dnstype))
result, out, err = self.runsubcmd("dns", "update",
dnstype, unused_record,
new_record,
self.creds_string)
- self.assertCmdFail(result, "Successfully updated record '%s' " \
- "from '%s' to '%s', even though the given " \
+ self.assertCmdFail(result, "Successfully updated record '%s' "
+ "from '%s' to '%s', even though the given "
"source record is incorrect."
% (used_record, unused_record, new_record))
self.zone, "testrecord",
"SOA", "test",
self.creds_string)
- self.assertCmdFail(result, "Successfully added record of type SOA, " \
+ self.assertCmdFail(result, "Successfully added record of type SOA, "
"when this type should not be available.")
self.assertTrue("type SOA is not supported" in err,
- "Invalid error message '%s' when attempting to " \
+ "Invalid error message '%s' when attempting to "
"add record of type SOA." % err)
def test_add_overlapping_different_type(self):
self.zone, "testrecord",
dnstype1, record1,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add record " \
+ self.assertCmdSuccess(result, out, err, "Failed to add record "
"'%s' of type '%s'." % (record1, dnstype1))
result, out, err = self.runsubcmd("dns", "add",
self.zone, "testrecord",
dnstype2, record2,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to add record " \
- "'%s' of type '%s' when a record '%s' " \
+ self.assertCmdSuccess(result, out, err, "Failed to add record "
+ "'%s' of type '%s' when a record '%s' "
"of type '%s' with the same name exists."
% (record1, dnstype1, record2, dnstype2))
os.environ["SERVER"],
self.zone, "testrecord",
dnstype1, self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to query for " \
- "record '%s' of type '%s' when a new " \
- "record '%s' of type '%s' with the same " \
+ self.assertCmdSuccess(result, out, err, "Failed to query for "
+ "record '%s' of type '%s' when a new "
+ "record '%s' of type '%s' with the same "
"name was added."
% (record1, dnstype1, record2, dnstype2))
os.environ["SERVER"],
self.zone, "testrecord",
dnstype2, self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to query " \
- "record '%s' of type '%s' which should " \
- "have been added with the same name as " \
+ self.assertCmdSuccess(result, out, err, "Failed to query "
+ "record '%s' of type '%s' which should "
+ "have been added with the same name as "
"record '%s' of type '%s'."
% (record2, dnstype2, record1, dnstype1))
self.zone, "testrecord",
dnstype1, record1,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to delete " \
+ self.assertCmdSuccess(result, out, err, "Failed to delete "
"record '%s' of type '%s'."
% (record1, dnstype1))
self.zone, "testrecord",
dnstype2, record2,
self.creds_string)
- self.assertCmdSuccess(result, out, err, "Failed to delete " \
+ self.assertCmdSuccess(result, out, err, "Failed to delete "
"record '%s' of type '%s'."
% (record2, dnstype2))
parser.set_defaults(server_helper="squid-2.5-ntlmssp")
parser.set_defaults(config_file="/etc/samba/smb.conf")
- parser.add_option("--client-username", dest="client_username",\
+ parser.add_option("--client-username", dest="client_username",
help="User name for the client. [default: foo]")
- parser.add_option("--client-password", dest="client_password",\
+ parser.add_option("--client-password", dest="client_password",
help="Password the client will send. [default: secret]")
- parser.add_option("--client-domain", dest="client_domain",\
+ parser.add_option("--client-domain", dest="client_domain",
help="Domain the client authenticates for. [default: FOO]")
- parser.add_option("--client-helper", dest="client_helper",\
+ parser.add_option("--client-helper", dest="client_helper",
help="Helper mode for the ntlm_auth client. [default: ntlmssp-client-1]")
- parser.add_option("--client-use-cached-creds", dest="client_use_cached_creds",\
+ parser.add_option("--client-use-cached-creds", dest="client_use_cached_creds",
help="Use winbindd credentials cache (rather than default username/pw)", action="store_true")
- parser.add_option("--target-hostname", dest="target_hostname",\
+ parser.add_option("--target-hostname", dest="target_hostname",
help="Target hostname for kerberos")
- parser.add_option("--target-service", dest="target_service",\
+ parser.add_option("--target-service", dest="target_service",
help="Target service for kerberos")
- parser.add_option("--server-username", dest="server_username",\
+ parser.add_option("--server-username", dest="server_username",
help="User name server uses for local auth. [default: foo]")
- parser.add_option("--server-password", dest="server_password",\
+ parser.add_option("--server-password", dest="server_password",
help="Password server uses for local auth. [default: secret]")
- parser.add_option("--server-domain", dest="server_domain",\
+ parser.add_option("--server-domain", dest="server_domain",
help="Domain server uses for local auth. [default: FOO]")
- parser.add_option("--server-helper", dest="server_helper",\
+ parser.add_option("--server-helper", dest="server_helper",
help="Helper mode for the ntlm_auth server. [default: squid-2.5-server]")
- parser.add_option("--server-use-winbindd", dest="server_use_winbindd",\
+ parser.add_option("--server-use-winbindd", dest="server_use_winbindd",
help="Use winbindd to check the password (rather than default username/pw)", action="store_true")
- parser.add_option("--require-membership-of", dest="sid",\
+ parser.add_option("--require-membership-of", dest="sid",
help="Require that the user is a member of this group to authenticate.")
- parser.add_option("-s", "--configfile", dest="config_file",\
+ parser.add_option("-s", "--configfile", dest="config_file",
help="Path to smb.conf file. [default:/etc/samba/smb.conf")
(opts, args) = parser.parse_args()
self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
# Modify on attribute you have rights for
self.ldb_user.modify_ldif(ldif)
- res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
% self.get_user_dn(self.user_with_wp), attrs=["adminDescription"])
self.assertEqual(res[0]["adminDescription"][0], "blah blah blah")
mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
self.ldb_user2.modify_ldif(ldif)
- res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
% ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_sm))
# but not other users
add: Member
Member: """ + self.get_user_dn(self.user_with_wp)
self.ldb_user.modify_ldif(ldif)
- res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
% ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_wp))
ldif = """
add: Member
Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
self.ldb_user.modify_ldif(ldif)
- res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
% ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
self.assertEqual(res[0]["Member"][0], "CN=test_modify_user2,CN=Users," + self.base_dn)
def create_clean_ou(self, object_dn):
""" Base repeating setup for unittests to follow """
- res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE, \
+ res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE,
expression="distinguishedName=%s" % object_dn)
# Make sure top testing OU has been deleted before starting the test
self.assertEqual(len(res), 0)
self.ldb_admin.create_ou("OU=test_rename_ou1," + self.base_dn)
self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
try:
- self.ldb_user.rename("CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn), \
+ self.ldb_user.rename("CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn),
"CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
except LdbError as e21:
(num, _) = e21.args
self.deleted_objects_domain_dn \
= self.ldb.get_wellknown_dn(self.ldb.get_default_basedn(),
dsdb.DS_GUID_DELETED_OBJECTS_CONTAINER)
- sites_obj = self.search_dn("cn=sites,%s" \
+ sites_obj = self.search_dn("cn=sites,%s"
% self.ldb.get_config_basedn())
self.sites_dn = sites_obj.dn
self.sites_guid \
delete_force(self.ldb_admin, "OU=test_domain_ou1," + self.base_dn)
# SCHEMA
# CONFIGURATION
- delete_force(self.ldb_admin, "CN=test-specifier1,CN=test-container1,CN=DisplaySpecifiers," \
+ delete_force(self.ldb_admin, "CN=test-specifier1,CN=test-container1,CN=DisplaySpecifiers,"
+ self.configuration_dn)
delete_force(self.ldb_admin, "CN=test-container1,CN=DisplaySpecifiers," + self.configuration_dn)
def create_clean_ou(self, object_dn):
""" Base repeating setup for unittests to follow """
- res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE, \
+ res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE,
expression="distinguishedName=%s" % object_dn)
# Make sure top testing OU has been deleted before starting the test
self.assertEqual(len(res), 0)
attrs=["sDRightsEffective"])
# user whould have DACL_SECURITY_INFORMATION, OWNER_SECURITY_INFORMATION, GROUP_SECURITY_INFORMATION
self.assertEquals(len(res), 1)
- self.assertEquals(res[0]["sDRightsEffective"][0], \
+ self.assertEquals(res[0]["sDRightsEffective"][0],
("%d") % (SECINFO_DACL | SECINFO_GROUP | SECINFO_OWNER | SECINFO_SACL))
def test_allowedChildClassesEffective(self):
f.close()
store_hdr = store_blob[0:28]
- (store_version, \
- store_dn_len, store_dn_ofs, \
- store_hwm_len, store_hwm_ofs, \
+ (store_version,
+ store_dn_len, store_dn_ofs,
+ store_hwm_len, store_hwm_ofs,
store_utdv_len, store_utdv_ofs) = \
struct.unpack("<LLLLLLL", store_hdr)
dn_ofs = store_hdr_len
hwm_ofs = dn_ofs + len(dn)
utdv_ofs = hwm_ofs + len(store_hwm_blob)
- store_blob = struct.pack("<LLLLLLL", 1, \
+ store_blob = struct.pack("<LLLLLLL", 1,
len(dn), dn_ofs,
- len(store_hwm_blob), hwm_ofs, \
+ len(store_hwm_blob), hwm_ofs,
len(store_utdv_blob), utdv_ofs) + \
dn + store_hwm_blob + store_utdv_blob
(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];
"objectClass": ["top", "classSchema"],
"cn": class_name,
"lDAPDisplayName": class_ldn,
- "governsId": "1.3.6.1.4.1.7165.4.6.2.5." \
+ "governsId": "1.3.6.1.4.1.7165.4.6.2.5."
+ str((100000 * base_int) + random.randint(1, 100000)) + ".1.5.13",
"instanceType": "4",
"objectClassCategory": "%d" % oc_cat,
"objectClass": ["top", "attributeSchema"],
"cn": attr_name,
"lDAPDisplayName": attr_ldn,
- "attributeId": "1.3.6.1.4.1.7165.4.6.1.5." \
+ "attributeId": "1.3.6.1.4.1.7165.4.6.1.5."
+ str((100000 * base_int) + random.randint(1, 100000)) + ".1.5.13",
"attributeSyntax": "2.5.5.12",
"omSyntax": "64",
self.parser.add_option("--use-ntvfs", action='store_true', default=False, help='use NTVFS for the fileserver')
self.parser.add_option("--dns-backend", type="choice",
choices=["SAMBA_INTERNAL", "BIND9_FLATFILE", "BIND9_DLZ", "NONE"],
- help="The DNS server backend. SAMBA_INTERNAL is the builtin name server (default), " \
- "BIND9_FLATFILE uses bind9 text database to store zone information, " \
- "BIND9_DLZ uses samba4 AD to store zone information, " \
+ help="The DNS server backend. SAMBA_INTERNAL is the builtin name server (default), "
+ "BIND9_FLATFILE uses bind9 text database to store zone information, "
+ "BIND9_DLZ uses samba4 AD to store zone information, "
"NONE skips the DNS setup entirely (not recommended)",
default="SAMBA_INTERNAL")
git.samba.org / metze / samba / wip.git / commitdiff