s4 dsdb: Use the changereplmetadata control

This control allow to specify the replPropertyMetaData attribute to
be specified on modify request. It can be used for very specific needs
to tweak the content of the replication data.

Signed-off-by: Andrew Bartlett <abartlet@samba.org>

1 files changed, 78 insertions, 11 deletions

diff --git a/source4/scripting/python/samba/tests/dsdb.py b/source4/scripting/python/samba/tests/dsdb.py
index c19dbaab47..d28be82984 100644
--- a/source4/scripting/python/samba/tests/dsdb.py
+++ b/source4/scripting/python/samba/tests/dsdb.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Unix SMB/CIFS implementation. Tests for dsdb 
+# Unix SMB/CIFS implementation. Tests for dsdb
 # Copyright (C) Matthieu Patou <mat@matws.net> 2010
 #
 # This program is free software; you can redistribute it and/or modify
@@ -17,26 +17,93 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 
-import samba.dsdb
 from samba.credentials import Credentials
 from samba.samdb import SamDB
 from samba.auth import system_session
 from testtools.testcase import TestCase
+from samba.ndr import ndr_unpack, ndr_pack
+from samba.dcerpc import drsblobs
+import ldb
 import os
+import samba.dsdb
 
 
 class DsdbTests(TestCase):
 
-    def _baseprovpath(self):
+
+    def setUp(self):
+        super(DsdbTests, self).setUp()
+        self.lp = samba.param.LoadParm()
+        self.lp.load(os.path.join(os.path.join(self.baseprovpath(), "etc"), "smb.conf"))
+        self.creds = Credentials()
+        self.creds.guess(self.lp)
+        self.session = system_session()
+        self.samdb = SamDB(os.path.join(self.baseprovpath(), "private", "sam.ldb"),
+            session_info=self.session, credentials=self.creds,lp=self.lp)
+
+
+    def baseprovpath(self):
         return os.path.join(os.environ['SELFTEST_PREFIX'], "dc")
 
+
     def test_get_oid_from_attrid(self):
-        lp = samba.param.LoadParm()
-        lp.load(os.path.join(os.path.join(self._baseprovpath(), "etc"), "smb.conf"))
-        creds = Credentials()
-        creds.guess(lp)
-        session = system_session()
-        test_ldb = SamDB(os.path.join(self._baseprovpath(), "private", "sam.ldb"),
-            session_info=session, credentials=creds,lp=lp)
-        oid = test_ldb.get_oid_from_attid(591614)
+        oid = self.samdb.get_oid_from_attid(591614)
         self.assertEquals(oid, "1.2.840.113556.1.4.1790")
+
+    def test_error_replpropertymetadata(self):
+        res = self.samdb.search(expression="cn=Administrator",
+                            scope=ldb.SCOPE_SUBTREE,
+                            attrs=["replPropertyMetaData"])
+        repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
+                            str(res[0]["replPropertyMetaData"]))
+        ctr = repl.ctr
+        for o in ctr.array:
+            # Search for Description
+            if o.attid == 13:
+                old_version = o.version
+                o.version = o.version + 1
+        replBlob = ndr_pack(repl)
+        msg = ldb.Message()
+        msg.dn = res[0].dn
+        msg["replPropertyMetaData"] = ldb.MessageElement(replBlob, ldb.FLAG_MOD_REPLACE, "replPropertyMetaData")
+        self.assertRaises(ldb.LdbError, self.samdb.modify, msg, ["local_oid:1.3.6.1.4.1.7165.4.3.14:0"])
+
+    def test_twoatt_replpropertymetadata(self):
+        res = self.samdb.search(expression="cn=Administrator",
+                            scope=ldb.SCOPE_SUBTREE,
+                            attrs=["replPropertyMetaData", "uSNChanged"])
+        repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
+                            str(res[0]["replPropertyMetaData"]))
+        ctr = repl.ctr
+        for o in ctr.array:
+            # Search for Description
+            if o.attid == 13:
+                old_version = o.version
+                o.version = o.version + 1
+                o.local_usn = long(str(res[0]["uSNChanged"])) + 1
+        replBlob = ndr_pack(repl)
+        msg = ldb.Message()
+        msg.dn = res[0].dn
+        msg["replPropertyMetaData"] = ldb.MessageElement(replBlob, ldb.FLAG_MOD_REPLACE, "replPropertyMetaData")
+        msg["description"] = ldb.MessageElement("new val", ldb.FLAG_MOD_REPLACE, "description")
+        self.assertRaises(ldb.LdbError, self.samdb.modify, msg, ["local_oid:1.3.6.1.4.1.7165.4.3.14:0"])
+
+    def test_set_replpropertymetadata(self):
+        res = self.samdb.search(expression="cn=Administrator",
+                            scope=ldb.SCOPE_SUBTREE,
+                            attrs=["replPropertyMetaData", "uSNChanged"])
+        repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
+                            str(res[0]["replPropertyMetaData"]))
+        ctr = repl.ctr
+        for o in ctr.array:
+            # Search for Description
+            if o.attid == 13:
+                old_version = o.version
+                o.version = o.version + 1
+                o.local_usn = long(str(res[0]["uSNChanged"])) + 1
+                o.originating_usn = long(str(res[0]["uSNChanged"])) + 1
+        replBlob = ndr_pack(repl)
+        msg = ldb.Message()
+        msg.dn = res[0].dn
+        msg["replPropertyMetaData"] = ldb.MessageElement(replBlob, ldb.FLAG_MOD_REPLACE, "replPropertyMetaData")
+        self.samdb.modify(msg, ["local_oid:1.3.6.1.4.1.7165.4.3.14:0"])