[PATCH 6/8] urgent_replication: Use subunit reporting, remove allow_empty_output.
Matthieu Patou
mat at samba.org
Thu Oct 16 00:11:44 MDT 2014
On Sun, Oct 12, 2014 at 09:19:02PM -0700, Jelmer Vernooij wrote:
> Change-Id: I6d479b218eff6c4292fbb99e4760bbd62ce1f380
> Signed-Off-By: Jelmer Vernooij <jelmer at samba.org>
> ---
> python/samba/tests/__init__.py | 7 +-
> source4/dsdb/tests/python/urgent_replication.py | 116 +++++++++---------------
> source4/selftest/tests.py | 2 +-
> 3 files changed, 50 insertions(+), 75 deletions(-)
>
> diff --git a/python/samba/tests/__init__.py b/python/samba/tests/__init__.py
> index 2df30a6..8d3b4dd 100644
> --- a/python/samba/tests/__init__.py
> +++ b/python/samba/tests/__init__.py
> @@ -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,
> diff --git a/source4/dsdb/tests/python/urgent_replication.py b/source4/dsdb/tests/python/urgent_replication.py
> index 2a07a58..8083ce4 100755
> --- a/source4/dsdb/tests/python/urgent_replication.py
> +++ b/source4/dsdb/tests/python/urgent_replication.py
> @@ -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)
> diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py
> index ce728f9..60017f9 100755
> --- a/source4/selftest/tests.py
> +++ b/source4/selftest/tests.py
> @@ -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"]:
> --
> 2.1.1
>
Reviewed-by: Matthieu Patou <mat at matws.net>
--
Matthieu Patou
Samba Team
http://samba.org
More information about the samba-technical
mailing list