sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2025-08-02 00:22:11 +03:00
Code Activity

s4-python: Various formatting fixes.

Browse Source
This commit is contained in:
Jelmer Vernooij
2012-02-18 23:59:48 +01:00
parent c29e0258e0
commit a977de9fdf
9 changed files with 42 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
source4/scripting/python/samba/__init__.py
View File

@ -28,15 +28,17 @@ import os
import sys
import samba.param
def source_tree_topdir():
"""Return the top level source directory."""
paths = [ "../../..", "../../../.." ]
paths = ["../../..", "../../../.."]
for p in paths:
topdir = os.path.normpath(os.path.join(os.path.dirname(__file__), p))
if os.path.exists(os.path.join(topdir, 'source4')):
return topdir
raise RuntimeError("unable to find top level source directory")
def in_source_tree():
"""Return True if we are running from within the samba source tree"""
try:
@ -49,6 +51,7 @@ def in_source_tree():
import ldb
from samba._ldb import Ldb as _Ldb
class Ldb(_Ldb):
"""Simple Samba-specific LDB subclass that takes care
of setting up the modules dir, credentials pointers, etc.
@ -166,7 +169,8 @@ class Ldb(_Ldb):
# Try to delete user/computer accounts to allow deletion of groups
self.erase_users_computers(basedn)
# Delete the 'visible' records, and the invisble 'deleted' records (if this DB supports it)
# Delete the 'visible' records, and the invisble 'deleted' records (if
# this DB supports it)
for msg in self.search(basedn, ldb.SCOPE_SUBTREE,
"(&(|(objectclass=*)(distinguishedName=*))(!(distinguishedName=@BASEINFO)))",
[], controls=["show_deleted:0", "show_recycled:0"]):
@ -178,7 +182,8 @@ class Ldb(_Ldb):
raise
res = self.search(basedn, ldb.SCOPE_SUBTREE,
"(&(|(objectclass=*)(distinguishedName=*))(!(distinguishedName=@BASEINFO)))", [], controls=["show_deleted:0", "show_recycled:0"])
"(&(|(objectclass=*)(distinguishedName=*))(!(distinguishedName=@BASEINFO)))",
[], controls=["show_deleted:0", "show_recycled:0"])
assert len(res) == 0
# delete the specials

7
source4/scripting/python/samba/common.py
View File

@ -18,7 +18,10 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import ldb, dsdb
import ldb
import dsdb
def confirm(msg, forced=False, allow_all=False):
"""confirm an action with the user
@ -73,7 +76,7 @@ class dsdb_Dn(object):
syntax_oid = dsdb.DSDB_SYNTAX_STRING_DN
else:
syntax_oid = dsdb.DSDB_SYNTAX_OR_NAME
if syntax_oid in [ dsdb.DSDB_SYNTAX_BINARY_DN, dsdb.DSDB_SYNTAX_STRING_DN ]:
if syntax_oid in [dsdb.DSDB_SYNTAX_BINARY_DN, dsdb.DSDB_SYNTAX_STRING_DN]:
# it is a binary DN
colons = dnstring.split(':')
if len(colons) < 4:

59
source4/scripting/python/samba/dbchecker.py
View File

@ -27,11 +27,12 @@ from samba.ndr import ndr_unpack
from samba.dcerpc import drsblobs
from samba.common import dsdb_Dn
class dbcheck(object):
"""check a SAM database for errors"""
def __init__(self, samdb, samdb_schema=None, verbose=False, fix=False, yes=False,
quiet=False, in_transaction=False):
def __init__(self, samdb, samdb_schema=None, verbose=False, fix=False,
yes=False, quiet=False, in_transaction=False):
self.samdb = samdb
self.dict_oid_name = None
self.samdb_schema = (samdb_schema or samdb)
@ -67,20 +68,14 @@ class dbcheck(object):
if error_count != 0 and not self.fix:
self.report("Please use --fix to fix these errors")
self.report('Checked %u objects (%u errors)' % (len(res), error_count))
return error_count
def report(self, msg):
'''print a message unless quiet is set'''
if not self.quiet:
print(msg)
################################################################
# a local confirm function that obeys the --fix and --yes options
def confirm(self, msg, allow_all=False, forced=False):
'''confirm a change'''
if not self.fix:
@ -114,7 +109,6 @@ class dbcheck(object):
return True
return c
def do_modify(self, m, controls, msg, validate=True):
'''perform a modify with optional verbose output'''
if self.verbose:
@ -126,9 +120,6 @@ class dbcheck(object):
return False
return True
################################################################
# handle empty attributes
def err_empty_attribute(self, dn, attrname):
'''fix empty attributes'''
self.report("ERROR: Empty attribute %s in %s" % (attrname, dn))
@ -143,15 +134,13 @@ class dbcheck(object):
"Failed to remove empty attribute %s" % attrname, validate=False):
self.report("Removed empty attribute %s" % attrname)
################################################################
# handle normalisation mismatches
def err_normalise_mismatch(self, dn, attrname, values):
'''fix attribute normalisation errors'''
self.report("ERROR: Normalisation error for attribute %s in %s" % (attrname, dn))
mod_list = []
for val in values:
normalised = self.samdb.dsdb_normalise_attributes(self.samdb_schema, attrname, [val])
normalised = self.samdb.dsdb_normalise_attributes(
self.samdb_schema, attrname, [val])
if len(normalised) != 1:
self.report("Unable to normalise value '%s'" % val)
mod_list.append((val, ''))
@ -168,7 +157,8 @@ class dbcheck(object):
(val, nval) = mod_list[i]
m['value_%u' % i] = ldb.MessageElement(val, ldb.FLAG_MOD_DELETE, attrname)
if nval != '':
m['normv_%u' % i] = ldb.MessageElement(nval, ldb.FLAG_MOD_ADD, attrname)
m['normv_%u' % i] = ldb.MessageElement(nval, ldb.FLAG_MOD_ADD,
attrname)
if self.do_modify(m, ["relax:0", "show_recycled:1"],
"Failed to normalise attribute %s" % attrname,
@ -179,10 +169,8 @@ class dbcheck(object):
'''see if a dsdb_Dn is the special Deleted Objects DN'''
return dsdb_dn.prefix == "B:32:18E2EA80684F11D2B9AA00C04F79F805:"
################################################################
# handle a DN pointing to a deleted object
def err_deleted_dn(self, dn, attrname, val, dsdb_dn, correct_dn):
"""handle a DN pointing to a deleted object"""
self.report("ERROR: target DN is deleted for %s in object %s - %s" % (attrname, dn, val))
self.report("Target GUID points at deleted DN %s" % correct_dn)
if not self.confirm_all('Remove DN link?', 'remove_all_deleted_DN_links'):
@ -195,9 +183,8 @@ class dbcheck(object):
"Failed to remove deleted DN attribute %s" % attrname):
self.report("Removed deleted DN on attribute %s" % attrname)
################################################################
# handle a missing target DN (both GUID and DN string form are missing)
def err_missing_dn_GUID(self, dn, attrname, val, dsdb_dn):
"""handle a missing target DN (both GUID and DN string form are missing)"""
# check if its a backlink
linkID = self.samdb_schema.get_linkId_from_lDAPDisplayName(attrname)
if (linkID & 1 == 0) and str(dsdb_dn).find('DEL\\0A') == -1:
@ -205,10 +192,8 @@ class dbcheck(object):
return
self.err_deleted_dn(dn, attrname, val, dsdb_dn, dsdb_dn)
################################################################
# handle a missing GUID extended DN component
def err_incorrect_dn_GUID(self, dn, attrname, val, dsdb_dn, errstr):
"""handle a missing GUID extended DN component"""
self.report("ERROR: %s component for %s in object %s - %s" % (errstr, attrname, dn, val))
controls=["extended_dn:1:1", "show_recycled:1"]
try:
@ -236,10 +221,8 @@ class dbcheck(object):
"Failed to fix %s on attribute %s" % (errstr, attrname)):
self.report("Fixed %s on attribute %s" % (errstr, attrname))
################################################################
# handle a DN string being incorrect
def err_dn_target_mismatch(self, dn, attrname, val, dsdb_dn, correct_dn, errstr):
"""handle a DN string being incorrect"""
self.report("ERROR: incorrect DN string component for %s in object %s - %s" % (attrname, dn, val))
dsdb_dn.dn = correct_dn
@ -254,8 +237,6 @@ class dbcheck(object):
"Failed to fix incorrect DN string on attribute %s" % attrname):
self.report("Fixed incorrect DN string on attribute %s" % (attrname))
################################################################
# handle an unknown attribute error
def err_unknown_attribute(self, obj, attrname):
'''handle an unknown attribute error'''
self.report("ERROR: unknown attribute '%s' in %s" % (attrname, obj.dn))
@ -269,9 +250,6 @@ class dbcheck(object):
"Failed to remove unknown attribute %s" % attrname):
self.report("Removed unknown attribute %s" % (attrname))
################################################################
# handle a missing backlink
def err_missing_backlink(self, obj, attrname, val, backlink_name, target_dn):
'''handle a missing backlink value'''
self.report("ERROR: missing backlink attribute '%s' in %s for link %s in %s" % (backlink_name, target_dn, attrname, obj.dn))
@ -286,9 +264,6 @@ class dbcheck(object):
"Failed to fix missing backlink %s" % backlink_name):
self.report("Fixed missing backlink %s" % (backlink_name))
################################################################
# handle a orphaned backlink
def err_orphaned_backlink(self, obj, attrname, val, link_name, target_dn):
'''handle a orphaned backlink value'''
self.report("ERROR: orphaned backlink attribute '%s' in %s for link %s in %s" % (attrname, obj.dn, link_name, target_dn))
@ -302,9 +277,6 @@ class dbcheck(object):
"Failed to fix orphaned backlink %s" % link_name):
self.report("Fixed orphaned backlink %s" % (link_name))
################################################################
# specialised checking for a dn attribute
def check_dn(self, obj, attrname, syntax_oid):
'''check a DN attribute for correctness'''
error_count = 0
@ -315,12 +287,13 @@ class dbcheck(object):
guid = dsdb_dn.dn.get_extended_component("GUID")
if guid is None:
error_count += 1
self.err_incorrect_dn_GUID(obj.dn, attrname, val, dsdb_dn, "missing GUID")
self.err_incorrect_dn_GUID(obj.dn, attrname, val, dsdb_dn,
"missing GUID")
continue
guidstr = str(misc.GUID(guid))
attrs=['isDeleted']
attrs = ['isDeleted']
linkID = self.samdb_schema.get_linkId_from_lDAPDisplayName(attrname)
reverse_link_name = self.samdb_schema.get_backlink_from_lDAPDisplayName(attrname)
if reverse_link_name is not None:
@ -415,9 +388,6 @@ class dbcheck(object):
"Failed to fix metadata for attribute %s" % attr):
self.report("Fixed metadata for attribute %s" % attr)
################################################################
# check one object - calls to individual error handlers above
def check_object(self, dn, attrs=['*']):
'''check one object'''
if self.verbose:
@ -454,7 +424,6 @@ class dbcheck(object):
got_repl_property_meta_data = True
continue
# check for empty attributes
for val in obj[attrname]:
if val == '':

3
source4/scripting/python/samba/tests/dcerpc/rpc_talloc.py
View File

@ -37,6 +37,7 @@ import talloc
talloc.enable_null_tracking()
class TallocTests(samba.tests.TestCase):
'''test talloc behaviour of pidl generated python code'''
@ -54,7 +55,7 @@ class TallocTests(samba.tests.TestCase):
# we expect one block for the object, and one for the structure
self.check_blocks(partial_attribute_set, 2)
attids = [ 1, 2, 3]
attids = [1, 2, 3]
partial_attribute_set.version = 1
partial_attribute_set.attids = attids
partial_attribute_set.num_attids = len(attids)

1
source4/scripting/python/samba/tests/strings.py
View File

@ -67,6 +67,7 @@ class strcasecmp_m_Tests(samba.tests.TestCase):
class strstr_m_Tests(samba.tests.TestCase):
"""strstr_m tests in simple ASCII and unicode strings"""
def test_strstr_m(self):
# A, B, strstr_m(A, B)
cases = [('hello', 'hello', 'hello'),

7
source4/scripting/python/samba/tests/upgrade.py
View File

@ -22,6 +22,7 @@
from samba.upgrade import import_wins
from samba.tests import LdbTestCase
class WinsUpgradeTests(LdbTestCase):
def test_upgrade(self):
@ -30,8 +31,10 @@ class WinsUpgradeTests(LdbTestCase):
}
import_wins(self.ldb, winsdb)
self.assertEquals(['name=FOO,type=0x20'],
[str(m.dn) for m in self.ldb.search(expression="(objectClass=winsRecord)")])
self.assertEquals(
['name=FOO,type=0x20'],
[str(m.dn) for m in
self.ldb.search(expression="(objectClass=winsRecord)")])
def test_version(self):
import_wins(self.ldb, {})

3
source4/scripting/python/samba/tests/upgradeprovisionneeddc.py
View File

@ -161,7 +161,8 @@ class UpgradeProvisionWithLdbTestCase(TestCaseInTempDir):
self.assertNotEquals(oem, "")
def test_update_dns_account(self):
update_dns_account_password(self.ldbs.sam, self.ldbs.secrets, self.names)
update_dns_account_password(self.ldbs.sam, self.ldbs.secrets,
self.names)
def test_updateOEMInfo(self):
realm = self.lp.get("realm")