s4 upgradeprovision: Move functions to helpers and improve code

Among code improvement the most significant part is that we now
compare DN object instead of their string representation. It allow
 to better react to case an white space difference.
Some new move objects have been added (ie. System into well known
security principals).

This will allow more unittesting

Signed-off-by: Jelmer Vernooij <jelmer@samba.org>

2 files changed, 194 insertions, 89 deletions

diff --git a/source4/scripting/python/samba/tests/upgradeprovision.py b/source4/scripting/python/samba/tests/upgradeprovision.py
index 7adb97f2983..c5e1094e0dd 100644
--- a/source4/scripting/python/samba/tests/upgradeprovision.py
+++ b/source4/scripting/python/samba/tests/upgradeprovision.py
@@ -18,90 +18,39 @@
 #
 
 import os
-from samba.credentials import Credentials
-from samba.auth import system_session
-from samba.upgradehelpers import get_paths, usn_in_range, get_ldbs,\
-                                 find_provision_key_parameters, dn_sort,\
-                                 identic_rename, get_diff_sddls
-from samba import param
+from samba.upgradehelpers import  usn_in_range, dn_sort,\
+                                  get_diff_sddls, update_secrets
+
+
+from samba.tests.provision import create_dummy_secretsdb
 from samba.tests import env_loadparm, TestCaseInTempDir
-import ldb
+from samba import Ldb
+from ldb import SCOPE_SUBTREE
+import samba.tests
 
 lp = env_loadparm()
 
+def dummymessage(a=None, b=None):
+    if 0:
+        print "none"
+
 
 class UpgradeProvisionTestCase(TestCaseInTempDir):
     """Some simple tests for individual functions in the provisioning code.
     """
-    def test_get_paths(self):
-        smbConfPath = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
-        targetdir = os.path.join(os.environ["SELFTEST_PREFIX"], "dc")
-        privatePath = os.path.join(targetdir, "private")
-
-        paths = get_paths(param, None, smbConfPath)
-        self.assertEquals(paths.private_dir, privatePath)
-
-        paths2 = get_paths(param, targetdir)
-        self.assertEquals(paths2.private_dir, privatePath)
-
     def test_usn_in_range(self):
+        range = [5, 25, 35, 55]
 
-        range = []
-        range.append(5)
-        range.append(25)
-        range.append(35)
-        range.append(55)
-
-        vals = []
-        vals.append(3)
-        vals.append(26)
-        vals.append(56)
+        vals = [3, 26, 56]
 
         for v in vals:
             self.assertFalse(usn_in_range(v, range))
 
-        vals = []
-        vals.append(5)
-        vals.append(20)
-        vals.append(25)
-        vals.append(35)
-        vals.append(36)
+        vals = [5, 20, 25, 35, 36]
 
         for v in vals:
             self.assertTrue(usn_in_range(v, range))
 
-
-    def test_get_ldbs(self):
-        smbConfPath = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
-        paths = get_paths(param, None, smbConfPath)
-        creds = Credentials()
-        creds.guess(lp)
-        try:
-            get_ldbs(paths, creds, system_session(), lp)
-        except:
-            self.assertTrue(0)
-
-    def test_find_key_param(self):
-        smbConfPath = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
-        paths = get_paths(param, None, smbConfPath)
-        creds = Credentials()
-        creds.guess(lp)
-        rootdn = "dc=samba,dc=example,dc=com"
-        ldbs = get_ldbs(paths, creds, system_session(), lp)
-        find_provision_key_parameters(ldbs.sam, ldbs.secrets, paths,
-                                                     smbConfPath, lp)
-        try:
-            names = find_provision_key_parameters(ldbs.sam, ldbs.secrets, paths,
-                                                     smbConfPath, lp)
-        except:
-            self.assertTrue(0)
-
-        self.assertTrue(names.realm == "SAMBA.EXAMPLE.COM")
-        self.assertTrue(str(names.rootdn).lower() == rootdn.lower())
-        self.assertTrue(names.ntdsguid != "")
-
-
-
     def test_dn_sort(self):
         # higher level comes after lower even if lexicographicaly closer
         # ie dc=tata,dc=toto (2 levels), comes after dc=toto
@@ -111,27 +60,7 @@ class UpgradeProvisionTestCase(TestCaseInTempDir):
         self.assertEquals(dn_sort("dc=toto,dc=tata",
                                     "cn=foo,dc=toto,dc=tata"), -1)
         self.assertEquals(dn_sort("cn=bar, dc=toto,dc=tata",
-                                    "cn=foo, dc=toto,dc=tata"), -1)
-
-    def test_identic_rename(self):
-        smbConfPath = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
-        paths = get_paths(param, None, smbConfPath)
-        creds = Credentials()
-        creds.guess(lp)
-        rootdn = "DC=samba,DC=example,DC=com"
-        ldbs = get_ldbs(paths, creds, system_session(), lp)
-
-        guestDN = ldb.Dn(ldbs.sam, "CN=Guest,CN=Users,%s" % rootdn)
-        try:
-            identic_rename(ldbs.sam, guestDN)
-            res = ldbs.sam.search(expression="(name=Guest)", base=rootdn,
-                                    scope=ldb.SCOPE_SUBTREE, attrs=["dn"])
-        except:
-            self.assertTrue(0)
-
-        self.assertEquals(len(res), 1)
-        self.assertEquals(str(res[0]["dn"]), "CN=Guest,CN=Users,%s" % rootdn)
-
+                                    "cn=foo, dc=toto,dc=tata"),-1)
     def test_get_diff_sddl(self):
         sddl = "O:SAG:DUD:AI(A;CIID;RPWPCRCCLCLORCWOWDSW;;;SA)\
 (A;CIID;RP LCLORC;;;AU)(A;CIID;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)S:AI(AU;CIIDSA;WP;;;WD)"
@@ -148,9 +77,9 @@ class UpgradeProvisionTestCase(TestCaseInTempDir):
 
         self.assertEquals(get_diff_sddls(sddl, sddl1) ,"")
         txt = get_diff_sddls(sddl, sddl2)
-        self.assertEquals(txt ,"\tOwner mismatch: SA (in ref) BA (in current)\n")
+        self.assertEquals(txt ,"\tOwner mismatch: SA (in ref) BA(in current)\n")
         txt = get_diff_sddls(sddl, sddl3)
-        self.assertEquals(txt ,"\tGroup mismatch: DU (in ref) BA (in current)\n")
+        self.assertEquals(txt ,"\tGroup mismatch: DU (in ref) BA(in current)\n")
         txt = get_diff_sddls(sddl, sddl4)
         txtmsg = "\tPart dacl is different between reference and current here\
  is the detail:\n\t\t(A;CIID;RPWPCRCCLCLORCWOWDSW;;;BA) ACE is not present in\
@@ -159,3 +88,41 @@ class UpgradeProvisionTestCase(TestCaseInTempDir):
         self.assertEquals(txt , txtmsg)
         txt = get_diff_sddls(sddl, sddl5)
         self.assertEquals(txt ,"\tCurrent ACL hasn't a sacl part\n")
+
+
+class UpdateSecretsTests(samba.tests.TestCaseInTempDir):
+    def setUp(self):
+        super(UpdateSecretsTests, self).setUp()
+        self.referencedb = create_dummy_secretsdb(
+            os.path.join(self.tempdir, "ref.ldb"))
+
+    def _getEmptyDb(self):
+        return Ldb(os.path.join(self.tempdir, "secrets.ldb"))
+
+    def _getCurrentFormatDb(self):
+        return create_dummy_secretsdb(
+            os.path.join(self.tempdir, "secrets.ldb"))
+
+    def test_trivial(self):
+        # Test that updating an already up-to-date secretsdb works fine
+        self.secretsdb = self._getCurrentFormatDb()
+        self.assertEquals(None,
+            update_secrets(self.referencedb, self.secretsdb, dummymessage))
+
+    def test_update_modules(self):
+        empty_db = self._getEmptyDb()
+        update_secrets(self.referencedb, empty_db, dummymessage)
+        newmodules = empty_db.search(
+            expression="dn=@MODULES", base="", scope=SCOPE_SUBTREE)
+        refmodules = self.referencedb.search(
+            expression="dn=@MODULES", base="", scope=SCOPE_SUBTREE)
+        self.assertEquals(newmodules, refmodules)
+
+    def tearDown(self):
+        for name in ["ref.ldb", "secrets.ldb"]:
+            path = os.path.join(self.tempdir, name)
+            if os.path.exists(path):
+                os.unlink(path)
+        super(UpdateSecretsTests, self).tearDown()
+
+
diff --git a/source4/scripting/python/samba/tests/upgradeprovisionneeddc.py b/source4/scripting/python/samba/tests/upgradeprovisionneeddc.py
new file mode 100644
index 00000000000..32fad14765b
--- /dev/null
+++ b/source4/scripting/python/samba/tests/upgradeprovisionneeddc.py
@@ -0,0 +1,138 @@
+#!/usr/bin/python
+
+# Unix SMB/CIFS implementation.
+# Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007-2008
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import os
+import re
+import shutil
+from samba.credentials import Credentials
+from samba.auth import system_session
+from samba.provision import getpolicypath
+from samba.upgradehelpers import (get_paths, get_ldbs,
+                                 find_provision_key_parameters, identic_rename,
+                                 updateOEMInfo, getOEMInfo, update_gpo,
+                                 delta_update_basesamdb)
+
+from samba.tests.provision import create_dummy_secretsdb
+from samba import param
+from samba.tests import env_loadparm, TestCaseInTempDir
+import ldb
+
+
+def dummymessage(a=None, b=None):
+    if 0:
+        print "none"
+
+lp = env_loadparm()
+smbConfPath = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
+
+class UpgradeProvisionBasicLdbHelpersTestCase(TestCaseInTempDir):
+    """Some simple tests for individual functions in the provisioning code.
+    """
+
+    def test_get_ldbs(self):
+        paths = get_paths(param, None, smbConfPath)
+        creds = Credentials()
+        creds.guess(lp)
+        get_ldbs(paths, creds, system_session(), lp)
+
+    def test_find_key_param(self):
+        paths = get_paths(param, None, smbConfPath)
+        creds = Credentials()
+        creds.guess(lp)
+        rootdn = "dc=samba,dc=example,dc=com"
+        ldbs = get_ldbs(paths, creds, system_session(), lp)
+        names = find_provision_key_parameters(ldbs.sam, ldbs.secrets, ldbs.idmap,
+                                                paths, smbConfPath, lp)
+        self.assertEquals(names.realm, "SAMBA.EXAMPLE.COM")
+        self.assertTrue(str(names.rootdn).lower() == rootdn.lower())
+        self.assertTrue(names.policyid_dc != None)
+        self.assertTrue(names.ntdsguid != "")
+
+
+class UpgradeProvisionWithLdbTestCase(TestCaseInTempDir):
+    def _getEmptyDbName(self):
+        return os.path.join(self.tempdir, "sam.ldb")
+
+    def setUp(self):
+        super(UpgradeProvisionWithLdbTestCase, self).setUp()
+        paths = get_paths(param, None, smbConfPath)
+        self.creds = Credentials()
+        self.creds.guess(lp)
+        self.paths = paths
+        self.ldbs = get_ldbs(paths, self.creds, system_session(), lp)
+        self.lp = lp
+        self.names = find_provision_key_parameters(self.ldbs.sam, self.ldbs.secrets,
+                                                       self.ldbs.idmap, paths, smbConfPath, lp)
+        self.referencedb = create_dummy_secretsdb(
+            os.path.join(self.tempdir, "ref.ldb"))
+
+    def test_identic_rename(self):
+        rootdn = "DC=samba,DC=example,DC=com"
+
+        guestDN = ldb.Dn(self.ldbs.sam, "CN=Guest,CN=Users,%s" % rootdn)
+        identic_rename(self.ldbs.sam, guestDN)
+        res = self.ldbs.sam.search(expression="(name=Guest)", base=rootdn,
+                                scope=ldb.SCOPE_SUBTREE, attrs=["dn"])
+        self.assertEquals(len(res), 1)
+        self.assertEquals(str(res[0]["dn"]), "CN=Guest,CN=Users,%s" % rootdn)
+
+    def test_delta_update_basesamdb(self):
+        dummysampath = self._getEmptyDbName()
+        delta_update_basesamdb(self.paths.samdb, dummysampath,
+                                self.creds, system_session(), self.lp, dummymessage)
+
+    def test_update_gpo_simple(self):
+        dir = getpolicypath(self.paths.sysvol, self.names.dnsdomain, self.names.policyid)
+        shutil.rmtree(dir)
+        self.assertFalse(os.path.isdir(dir))
+        update_gpo(self.paths, self.ldbs.sam, self.names, self.lp, dummymessage)
+        self.assertTrue(os.path.isdir(dir))
+
+    def test_update_gpo_acl(self):
+        path = os.path.join(self.tempdir, "testupdategpo")
+        save = self.paths.sysvol
+        self.paths.sysvol = path
+        os.mkdir(path)
+        os.mkdir(os.path.join(path, self.names.dnsdomain))
+        os.mkdir(os.path.join(os.path.join(path, self.names.dnsdomain), "Policies"))
+        update_gpo(self.paths, self.ldbs.sam, self.names, self.lp, dummymessage)
+        shutil.rmtree(path)
+        self.paths.sysvol = save
+
+    def test_getOEMInfo(self):
+        realm = self.lp.get("realm")
+        basedn = "DC=%s" % realm.replace(".", ", DC=")
+        oem = getOEMInfo(self.ldbs.sam, basedn)
+        self.assertTrue(oem != "")
+
+    def test_updateOEMInfo(self):
+        realm = self.lp.get("realm")
+        basedn = "DC=%s" % realm.replace(".", ", DC=")
+        oem = getOEMInfo(self.ldbs.sam, basedn)
+        updateOEMInfo(self.ldbs.sam, basedn)
+        oem2 = getOEMInfo(self.ldbs.sam, basedn)
+        self.assertTrue(str(oem) != str(oem2))
+        self.assertTrue(re.match(".*upgrade to.*", str(oem2)))
+
+    def tearDown(self):
+        for name in ["ref.ldb", "secrets.ldb", "sam.ldb"]:
+            path = os.path.join(self.tempdir, name)
+            if os.path.exists(path):
+                os.unlink(path)
+        super(UpgradeProvisionWithLdbTestCase, self).tearDown()