selftest: Add ldap rodc python test

Signed-off-by: Garming Sam <garming@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
Pair-programmed-with: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>

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

diff --git a/selftest/knownfail b/selftest/knownfail
index 1a606c8154a4eb116fdd650ede5456cc3caa4130..7e204edcd017b81f6a61984eb093c9411a4cda03 100644 (file)
--- a/selftest/knownfail
+++ b/selftest/knownfail
@@ -326,3 +326,6 @@
 ^samba4.blackbox.trust_ntlm.Test08.*client.*with.ADDOM.SAMBA.EXAMPLE.COM\\Administrator%locDCpass1\(fl2003dc:local\)
 ^samba4.blackbox.trust_ntlm.Test09.*client.*with.Administrator@ADDOMAIN%locDCpass1\(fl2003dc:local\)
 ^samba4.blackbox.trust_ntlm.Test10.*client.*with.Administrator@ADDOM.SAMBA.EXAMPLE.COM%locDCpass1\(fl2003dc:local\)
+^samba4.ldap.rodc.python\(rodc\).__main__.RodcTests.test_add.*
+^samba4.ldap.rodc.python\(rodc\).__main__.RodcTests.test_delete.*
+^samba4.ldap.rodc.python\(rodc\).__main__.RodcTests.test_modify_nonreplicated.*

diff --git a/source4/dsdb/tests/python/rodc.py b/source4/dsdb/tests/python/rodc.py
new file mode 100755 (executable)
index 0000000..173f900
--- /dev/null
+++ b/source4/dsdb/tests/python/rodc.py
@@ -0,0 +1,209 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+import optparse
+import sys
+import os
+import base64
+import random
+import re
+import uuid
+
+sys.path.insert(0, "bin/python")
+import samba
+from samba.tests.subunitrun import SubunitOptions, TestProgram
+
+import samba.getopt as options
+
+from samba.auth import system_session
+import ldb
+from samba.samdb import SamDB
+from samba.ndr import ndr_pack, ndr_unpack
+from samba.dcerpc import drsblobs
+
+import time
+
+
+class RodcTestException(Exception):
+    pass
+
+
+class RodcTests(samba.tests.TestCase):
+
+    def setUp(self):
+        super(RodcTests, self).setUp()
+        self.samdb = SamDB(HOST, credentials=CREDS,
+                           session_info=system_session(LP), lp=LP)
+
+        self.base_dn = self.samdb.domain_dn()
+
+        root = self.samdb.search(base='', scope=ldb.SCOPE_BASE,
+                                 attrs=['dsServiceName'])
+        self.service = root[0]['dsServiceName'][0]
+        self.tag = uuid.uuid4().hex
+
+    def test_add_replicated_objects(self):
+        for o in (
+                {
+                    'dn': "ou=%s1,%s" % (self.tag, self.base_dn),
+                    "objectclass": "organizationalUnit"
+                },
+                {
+                    'dn': "cn=%s2,%s" % (self.tag, self.base_dn),
+                    "objectclass": "user"
+                },
+                {
+                    'dn': "cn=%s3,%s" % (self.tag, self.base_dn),
+                    "objectclass": "group"
+                },
+                {
+                    'dn': "cn=%s4,%s" % (self.tag, self.service),
+                    "objectclass": "NTDSConnection",
+                    'enabledConnection': 'TRUE',
+                    'fromServer': self.base_dn,
+                    'options': '0'
+                },
+        ):
+            try:
+                self.samdb.add(o)
+                self.fail("Failed to fail to add %s" % o['dn'])
+            except ldb.LdbError as (ecode, emsg):
+                if ecode != ldb.ERR_REFERRAL:
+                    print emsg
+                    self.fail("Adding %s: ldb error: %s %s, wanted referral" %
+                              (o['dn'], ecode, emsg))
+                else:
+                    m = re.search(r'(ldap://[^>]+)>', emsg)
+                    if m is None:
+                        self.fail("referral seems not to refer to anything")
+                    address = m.group(1)
+
+                    try:
+                        tmpdb = SamDB(address, credentials=CREDS,
+                                      session_info=system_session(LP), lp=LP)
+                        tmpdb.add(o)
+                        tmpdb.delete(o['dn'])
+                    except ldb.LdbError, e:
+                        self.fail("couldn't modify referred location %s" %
+                                  address)
+
+    def test_modify_replicated_attributes(self):
+        # some timestamp ones
+        dn = 'CN=Guest,CN=Users,' + self.base_dn
+        value = 'hallooo'
+        for attr in ['carLicense', 'middleName']:
+            m = ldb.Message()
+            m.dn = ldb.Dn(self.samdb, dn)
+            m[attr] = ldb.MessageElement(value,
+                                         ldb.FLAG_MOD_REPLACE,
+                                         attr)
+            try:
+                self.samdb.modify(m)
+                self.fail("Failed to fail to modify %s %s" % (dn, attr))
+            except ldb.LdbError as (ecode, emsg):
+                if ecode != ldb.ERR_REFERRAL:
+                    self.fail("Failed to REFER when trying to modify %s %s" %
+                              (dn, attr))
+
+    def test_modify_nonreplicated_attributes(self):
+        # some timestamp ones
+        dn = 'CN=Guest,CN=Users,' + self.base_dn
+        value = '123456789'
+        for attr in ['badPwdCount', 'lastLogon', 'lastLogoff']:
+            m = ldb.Message()
+            m.dn = ldb.Dn(self.samdb, dn)
+            m[attr] = ldb.MessageElement(value,
+                                         ldb.FLAG_MOD_REPLACE,
+                                         attr)
+            # Windows refers these ones even though they are non-replicated
+            try:
+                self.samdb.modify(m)
+                self.fail("Failed to fail to modify %s %s" % (dn, attr))
+            except ldb.LdbError as (ecode, emsg):
+                if ecode != ldb.ERR_REFERRAL:
+                    self.fail("Failed to REFER when trying to modify %s %s" %
+                              (dn, attr))
+
+    def test_modify_nonreplicated_reps_attributes(self):
+        # some timestamp ones
+        dn = self.base_dn
+
+        m = ldb.Message()
+        m.dn = ldb.Dn(self.samdb, dn)
+        attr = 'repsFrom'
+
+        res = self.samdb.search(dn, scope=ldb.SCOPE_BASE,
+                                attrs=['repsFrom'])
+        rep = ndr_unpack(drsblobs.repsFromToBlob, res[0]['repsFrom'][0],
+                         allow_remaining=True)
+        rep.ctr.result_last_attempt = -1
+        value = ndr_pack(rep)
+
+        m[attr] = ldb.MessageElement(value,
+                                     ldb.FLAG_MOD_REPLACE,
+                                     attr)
+        try:
+            self.samdb.modify(m)
+            self.fail("Failed to fail to modify %s %s" % (dn, attr))
+        except ldb.LdbError as (ecode, emsg):
+            if ecode != ldb.ERR_REFERRAL:
+                self.fail("Failed to REFER when trying to modify %s %s" %
+                          (dn, attr))
+
+    def test_delete_special_objects(self):
+        dn = 'CN=Guest,CN=Users,' + self.base_dn
+        try:
+            self.samdb.delete(dn)
+            self.fail("Failed to fail to delete %s" % (dn))
+        except ldb.LdbError as (ecode, emsg):
+            if ecode != ldb.ERR_REFERRAL:
+                print ecode, emsg
+                self.fail("Failed to REFER when trying to delete %s" % dn)
+
+    def test_no_delete_nonexistent_objects(self):
+        dn = 'CN=does-not-exist-%s,CN=Users,%s' % (self.tag, self.base_dn)
+        try:
+            self.samdb.delete(dn)
+            self.fail("Failed to fail to delete %s" % (dn))
+        except ldb.LdbError as (ecode, emsg):
+            if ecode != ldb.ERR_NO_SUCH_OBJECT:
+                print ecode, emsg
+                self.fail("Failed to NO_SUCH_OBJECT when trying to delete "
+                          "%s (which does not exist)" % dn)
+
+
+
+def main():
+    global HOST, CREDS, LP
+    parser = optparse.OptionParser("rodc.py [options] <host>")
+
+    sambaopts = options.SambaOptions(parser)
+    versionopts = options.VersionOptions(parser)
+    credopts = options.CredentialsOptions(parser)
+    subunitopts = SubunitOptions(parser)
+
+    parser.add_option_group(sambaopts)
+    parser.add_option_group(versionopts)
+    parser.add_option_group(credopts)
+    parser.add_option_group(subunitopts)
+
+    opts, args = parser.parse_args()
+
+    LP = sambaopts.get_loadparm()
+    CREDS = credopts.get_credentials(LP)
+
+    try:
+        HOST = args[0]
+    except IndexError:
+        parser.print_usage()
+        sys.exit(1)
+
+    if "://" not in HOST:
+        if os.path.isfile(HOST):
+            HOST = "tdb://%s" % HOST
+        else:
+            HOST = "ldap://%s" % HOST
+
+    TestProgram(module=__name__, opts=subunitopts)
+
+main()

diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py
index 623aca74c4138c15e78c93813bc30424a6ac697d..c1b2f2c24cabecfd79a0189dbafc1f211bbd1552 100755 (executable)
--- a/source4/selftest/tests.py
+++ b/source4/selftest/tests.py
@@ -645,6 +645,12 @@ plantestsuite_loadlist("samba4.ldap.sort.python(ad_dc_ntvfs)", "ad_dc_ntvfs", [p
 plantestsuite_loadlist("samba4.ldap.vlv.python(ad_dc_ntvfs)", "ad_dc_ntvfs", [python, os.path.join(samba4srcdir, "dsdb/tests/python/vlv.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT'])
 plantestsuite_loadlist("samba4.ldap.linked_attributes.python(ad_dc_ntvfs)", "ad_dc_ntvfs:local", [python, os.path.join(samba4srcdir, "dsdb/tests/python/linked_attributes.py"), '$PREFIX_ABS/ad_dc_ntvfs/private/sam.ldb', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT'])
 
+plantestsuite_loadlist("samba4.ldap.rodc.python(rodc)", "rodc",
+                       [python,
+                        os.path.join(samba4srcdir, "dsdb/tests/python/rodc.py"),
+                        '$SERVER', '-U"$USERNAME%$PASSWORD"',
+                        '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT'])
+
 for env in ["ad_dc_ntvfs", "fl2000dc", "fl2003dc", "fl2008r2dc"]:
     plantestsuite_loadlist("samba4.ldap_schema.python(%s)" % env, env, [python, os.path.join(samba4srcdir, "dsdb/tests/python/ldap_schema.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT'])
     plantestsuite("samba4.ldap.possibleInferiors.python(%s)" % env, env, [python, os.path.join(samba4srcdir, "dsdb/samdb/ldb_modules/tests/possibleinferiors.py"), "ldap://$SERVER", '-U"$USERNAME%$PASSWORD"', "-W$DOMAIN"])