diff options
Diffstat (limited to 'source4/scripting/python/samba/tests')
| -rw-r--r-- | source4/scripting/python/samba/tests/upgradeprovision.py | 145 | ||||
| -rw-r--r-- | source4/scripting/python/samba/tests/upgradeprovisionneeddc.py | 138 |
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() |