mirror of
https://github.com/samba-team/samba.git
synced 2025-01-08 21:18:16 +03:00
urgent_replication: Use subunit reporting, remove allow_empty_output.
Change-Id: I6d479b218eff6c4292fbb99e4760bbd62ce1f380 Signed-Off-By: Jelmer Vernooij <jelmer@samba.org> Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
Jelmer Vernooij
Andrew Bartlett
parent
7043a8b62a
commit
23ac8d130c
@ -166,8 +166,9 @@ class BlackboxTestCase(TestCase):
|
||||
raise BlackboxProcessError(retcode, line, p.stdout.read(), p.stderr.read())
|
||||
return p.stdout.read()
|
||||
|
||||
|
||||
def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
|
||||
flags=0, ldb_options=None, ldap_only=False):
|
||||
flags=0, ldb_options=None, ldap_only=False, global_schema=True):
|
||||
"""Create SamDB instance and connects to samdb_url database.
|
||||
|
||||
:param samdb_url: Url for database to connect to.
|
||||
@ -176,6 +177,7 @@ def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
|
||||
:param credentials: Optional credentials, defaults to anonymous.
|
||||
:param flags: Optional LDB flags
|
||||
:param ldap_only: If set, only remote LDAP connection will be created.
|
||||
:param global_schema: Whether to use global schema.
|
||||
|
||||
Added value for tests is that we have a shorthand function
|
||||
to make proper URL for ldb.connect() while using default
|
||||
@ -207,7 +209,8 @@ def connect_samdb(samdb_url, lp=None, session_info=None, credentials=None,
|
||||
session_info=session_info,
|
||||
credentials=credentials,
|
||||
flags=flags,
|
||||
options=ldb_options)
|
||||
options=ldb_options,
|
||||
global_schema=global_schema)
|
||||
|
||||
|
||||
def connect_samdb_ex(samdb_url, lp=None, session_info=None, credentials=None,
|
||||
|
||||
@ -3,29 +3,26 @@
|
||||
|
||||
import optparse
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.insert(0, "bin/python")
|
||||
import samba
|
||||
samba.ensure_external_module("testtools", "testtools")
|
||||
samba.ensure_external_module("subunit", "subunit/python")
|
||||
|
||||
import samba.getopt as options
|
||||
|
||||
from samba.auth import system_session
|
||||
from ldb import (LdbError, ERR_NO_SUCH_OBJECT, Message,
|
||||
MessageElement, Dn, FLAG_MOD_REPLACE)
|
||||
from samba.samdb import SamDB
|
||||
import samba.tests
|
||||
import samba.dsdb as dsdb
|
||||
import samba.getopt as options
|
||||
|
||||
from subunit.run import SubunitTestRunner
|
||||
|
||||
import unittest
|
||||
|
||||
parser = optparse.OptionParser("urgent_replication.py [options] <host>")
|
||||
sambaopts = options.SambaOptions(parser)
|
||||
parser.add_option_group(sambaopts)
|
||||
parser.add_option_group(options.VersionOptions(parser))
|
||||
|
||||
# use command line creds if available
|
||||
credopts = options.CredentialsOptions(parser)
|
||||
parser.add_option_group(credopts)
|
||||
@ -37,8 +34,6 @@ if len(args) < 1:
|
||||
|
||||
host = args[0]
|
||||
|
||||
lp = sambaopts.get_loadparm()
|
||||
creds = credopts.get_credentials(lp)
|
||||
|
||||
class UrgentReplicationTests(samba.tests.TestCase):
|
||||
|
||||
@ -50,30 +45,29 @@ class UrgentReplicationTests(samba.tests.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(UrgentReplicationTests, self).setUp()
|
||||
self.ldb = ldb
|
||||
self.base_dn = ldb.domain_dn()
|
||||
self.ldb = samba.tests.connect_samdb(host, global_schema=False)
|
||||
self.base_dn = self.ldb.domain_dn()
|
||||
|
||||
print "baseDN: %s\n" % self.base_dn
|
||||
|
||||
def test_nonurgent_object(self):
|
||||
"""Test if the urgent replication is not activated
|
||||
when handling a non urgent object"""
|
||||
"""Test if the urgent replication is not activated when handling a non urgent object."""
|
||||
self.ldb.add({
|
||||
"dn": "cn=nonurgenttest,cn=users," + self.base_dn,
|
||||
"objectclass":"user",
|
||||
"samaccountname":"nonurgenttest",
|
||||
"description":"nonurgenttest description"})
|
||||
|
||||
# urgent replication should not be enabled when creating
|
||||
# urgent replication should not be enabled when creating
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
# urgent replication should not be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=nonurgenttest,cn=users," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=nonurgenttest,cn=users," + self.base_dn)
|
||||
m["description"] = MessageElement("new description", FLAG_MOD_REPLACE,
|
||||
"description")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
@ -82,12 +76,11 @@ class UrgentReplicationTests(samba.tests.TestCase):
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
def test_nTDSDSA_object(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling a nTDSDSA object'''
|
||||
"""Test if the urgent replication is activated when handling a nTDSDSA object."""
|
||||
self.ldb.add({
|
||||
"dn": "cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,%s" % self.ldb.get_config_basedn(),
|
||||
"dn": "cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,%s" %
|
||||
self.ldb.get_config_basedn(),
|
||||
"objectclass":"server",
|
||||
"cn":"test server",
|
||||
"name":"test server",
|
||||
@ -107,10 +100,10 @@ systemFlags: 33554432""", ["relax:0"])
|
||||
|
||||
# urgent replication should NOT be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=NTDS Settings test,cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=NTDS Settings test,cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn)
|
||||
m["options"] = MessageElement("0", FLAG_MOD_REPLACE,
|
||||
"options")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn("cn=Configuration," + self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
@ -121,15 +114,13 @@ systemFlags: 33554432""", ["relax:0"])
|
||||
|
||||
self.delete_force(self.ldb, "cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn)
|
||||
|
||||
|
||||
def test_crossRef_object(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling a crossRef object'''
|
||||
"""Test if the urgent replication is activated when handling a crossRef object."""
|
||||
self.ldb.add({
|
||||
"dn": "CN=test crossRef,CN=Partitions,CN=Configuration,"+ self.base_dn,
|
||||
"objectClass": "crossRef",
|
||||
"cn": "test crossRef",
|
||||
"dnsRoot": lp.get("realm").lower(),
|
||||
"dnsRoot": self.get_loadparm().get("realm").lower(),
|
||||
"instanceType": "4",
|
||||
"nCName": self.base_dn,
|
||||
"showInAdvancedViewOnly": "TRUE",
|
||||
@ -142,10 +133,10 @@ systemFlags: 33554432""", ["relax:0"])
|
||||
|
||||
# urgent replication should NOT be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=test crossRef,CN=Partitions,CN=Configuration," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=test crossRef,CN=Partitions,CN=Configuration," + self.base_dn)
|
||||
m["systemFlags"] = MessageElement("0", FLAG_MOD_REPLACE,
|
||||
"systemFlags")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn("cn=Configuration," + self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
@ -155,11 +146,8 @@ systemFlags: 33554432""", ["relax:0"])
|
||||
res = self.ldb.load_partition_usn("cn=Configuration," + self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
|
||||
def test_attributeSchema_object(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling an attributeSchema object'''
|
||||
"""Test if the urgent replication is activated when handling an attributeSchema object"""
|
||||
|
||||
try:
|
||||
self.ldb.add_ldif(
|
||||
@ -186,19 +174,17 @@ name: test attributeSchema""")
|
||||
except LdbError:
|
||||
print "Not testing urgent replication when creating attributeSchema object ...\n"
|
||||
|
||||
# urgent replication should be enabled when modifying
|
||||
# urgent replication should be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "CN=test attributeSchema,CN=Schema,CN=Configuration," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "CN=test attributeSchema,CN=Schema,CN=Configuration," + self.base_dn)
|
||||
m["lDAPDisplayName"] = MessageElement("updated test attributeSchema", FLAG_MOD_REPLACE,
|
||||
"lDAPDisplayName")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn("cn=Schema,cn=Configuration," + self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
def test_classSchema_object(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling a classSchema object'''
|
||||
"""Test if the urgent replication is activated when handling a classSchema object."""
|
||||
try:
|
||||
self.ldb.add_ldif(
|
||||
"""dn: CN=test classSchema,CN=Schema,CN=Configuration,%s""" % self.base_dn + """
|
||||
@ -231,17 +217,15 @@ defaultHidingValue: TRUE""")
|
||||
|
||||
# urgent replication should be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "CN=test classSchema,CN=Schema,CN=Configuration," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "CN=test classSchema,CN=Schema,CN=Configuration," + self.base_dn)
|
||||
m["lDAPDisplayName"] = MessageElement("updated test classSchema", FLAG_MOD_REPLACE,
|
||||
"lDAPDisplayName")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn("cn=Schema,cn=Configuration," + self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
def test_secret_object(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling a secret object'''
|
||||
"""Test if the urgent replication is activated when handling a secret object."""
|
||||
|
||||
self.ldb.add({
|
||||
"dn": "cn=test secret,cn=System," + self.base_dn,
|
||||
@ -256,22 +240,20 @@ defaultHidingValue: TRUE""")
|
||||
|
||||
# urgent replication should be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=test secret,cn=System," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=test secret,cn=System," + self.base_dn)
|
||||
m["currentValue"] = MessageElement("yyyyyyyy", FLAG_MOD_REPLACE,
|
||||
"currentValue")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
# urgent replication should NOT be enabled when deleting
|
||||
# urgent replication should NOT be enabled when deleting
|
||||
self.delete_force(self.ldb, "cn=test secret,cn=System," + self.base_dn)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
def test_rIDManager_object(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling a rIDManager object'''
|
||||
"""Test if the urgent replication is activated when handling a rIDManager object."""
|
||||
self.ldb.add_ldif(
|
||||
"""dn: CN=RID Manager test,CN=System,%s""" % self.base_dn + """
|
||||
objectClass: rIDManager
|
||||
@ -289,10 +271,10 @@ rIDAvailablePool: 133001-1073741823""", ["relax:0"])
|
||||
|
||||
# urgent replication should be enabled when modifying
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "CN=RID Manager test,CN=System," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "CN=RID Manager test,CN=System," + self.base_dn)
|
||||
m["systemFlags"] = MessageElement("0", FLAG_MOD_REPLACE,
|
||||
"systemFlags")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
@ -301,10 +283,8 @@ rIDAvailablePool: 133001-1073741823""", ["relax:0"])
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
def test_urgent_attributes(self):
|
||||
'''Test if the urgent replication is activated
|
||||
when handling urgent attributes of an object'''
|
||||
"""Test if the urgent replication is activated when handling urgent attributes of an object."""
|
||||
|
||||
self.ldb.add({
|
||||
"dn": "cn=user UrgAttr test,cn=users," + self.base_dn,
|
||||
@ -319,40 +299,40 @@ rIDAvailablePool: 133001-1073741823""", ["relax:0"])
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
# urgent replication should be enabled when modifying userAccountControl
|
||||
# urgent replication should be enabled when modifying userAccountControl
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m["userAccountControl"] = MessageElement(str(dsdb.UF_NORMAL_ACCOUNT+dsdb.UF_SMARTCARD_REQUIRED), FLAG_MOD_REPLACE,
|
||||
"userAccountControl")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
# urgent replication should be enabled when modifying lockoutTime
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m["lockoutTime"] = MessageElement("1", FLAG_MOD_REPLACE,
|
||||
"lockoutTime")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
# urgent replication should be enabled when modifying pwdLastSet
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m["pwdLastSet"] = MessageElement("1", FLAG_MOD_REPLACE,
|
||||
"pwdLastSet")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
# urgent replication should NOT be enabled when modifying a not-urgent
|
||||
# attribute
|
||||
m = Message()
|
||||
m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
|
||||
m["description"] = MessageElement("updated urgent attributes test description",
|
||||
FLAG_MOD_REPLACE, "description")
|
||||
ldb.modify(m)
|
||||
self.ldb.modify(m)
|
||||
res = self.ldb.load_partition_usn(self.base_dn)
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
@ -362,18 +342,10 @@ rIDAvailablePool: 133001-1073741823""", ["relax:0"])
|
||||
self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
|
||||
|
||||
|
||||
if not "://" in host:
|
||||
if os.path.isfile(host):
|
||||
host = "tdb://%s" % host
|
||||
else:
|
||||
host = "ldap://%s" % host
|
||||
|
||||
|
||||
ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp,
|
||||
global_schema=False)
|
||||
|
||||
runner = SubunitTestRunner()
|
||||
rc = 0
|
||||
#
|
||||
if not runner.run(unittest.makeSuite(UrgentReplicationTests)).wasSuccessful():
|
||||
rc = 1
|
||||
|
||||
sys.exit(rc)
|
||||
|
||||
@ -469,7 +469,7 @@ planoldpythontestsuite("dc", "dsdb_schema_info",
|
||||
extra_path=[os.path.join(samba4srcdir, 'dsdb/tests/python')],
|
||||
name="samba4.schemaInfo.python(dc)",
|
||||
extra_args=['-U"$DOMAIN/$DC_USERNAME%$DC_PASSWORD"'])
|
||||
plantestsuite("samba4.urgent_replication.python(dc)", "dc:local", [python, os.path.join(samba4srcdir, "dsdb/tests/python/urgent_replication.py"), '$PREFIX_ABS/dc/private/sam.ldb'], allow_empty_output=True)
|
||||
plantestsuite("samba4.urgent_replication.python(dc)", "dc:local", [python, os.path.join(samba4srcdir, "dsdb/tests/python/urgent_replication.py"), '$PREFIX_ABS/dc/private/sam.ldb'])
|
||||
plantestsuite("samba4.ldap.dirsync.python(dc)", "dc", [python, os.path.join(samba4srcdir, "dsdb/tests/python/dirsync.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN'])
|
||||
plantestsuite("samba4.ldap.sites.python(dc)", "dc", [python, os.path.join(samba4srcdir, "dsdb/tests/python/sites.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN'])
|
||||
for env in ["dc", "fl2000dc", "fl2003dc", "fl2008r2dc"]:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user