from samba.dcerpc import drsblobs
from samba import dsdb
import ldb
-import os
import samba
-import gc
-import time
class DsdbTests(TestCase):
self.samdb.set_attribute_replmetadata_version(dn, "description", version + 2)
self.assertEqual(self.samdb.get_attribute_replmetadata_version(dn, "description"), version + 2)
- def test_db_lock1(self):
- basedn = self.samdb.get_default_basedn()
- (r1, w1) = os.pipe()
-
- pid = os.fork()
- if pid == 0:
- # In the child, close the main DB, re-open just one DB
- del(self.samdb)
- gc.collect()
- self.samdb = SamDB(session_info=self.session,
- credentials=self.creds,
- lp=self.lp)
-
- self.samdb.transaction_start()
-
- dn = "cn=test_db_lock_user,cn=users," + str(basedn)
- self.samdb.add({
- "dn": dn,
- "objectclass": "user",
- })
- self.samdb.delete(dn)
-
- # Obtain a write lock
- self.samdb.transaction_prepare_commit()
- os.write(w1, b"prepared")
- time.sleep(2)
-
- # Drop the write lock
- self.samdb.transaction_cancel()
- os._exit(0)
-
- self.assertEqual(os.read(r1, 8), b"prepared")
-
- start = time.time()
-
- # We need to hold this iterator open to hold the all-record lock.
- res = self.samdb.search_iterator()
-
- # This should take at least 2 seconds because the transaction
- # has a write lock on one backend db open
-
- # Release the locks
- for l in res:
- pass
-
- end = time.time()
- self.assertGreater(end - start, 1.9)
-
- (got_pid, status) = os.waitpid(pid, 0)
- self.assertEqual(got_pid, pid)
- self.assertTrue(os.WIFEXITED(status))
- self.assertEqual(os.WEXITSTATUS(status), 0)
-
- def test_db_lock2(self):
- basedn = self.samdb.get_default_basedn()
- (r1, w1) = os.pipe()
- (r2, w2) = os.pipe()
-
- pid = os.fork()
- if pid == 0:
- # In the child, close the main DB, re-open
- del(self.samdb)
- gc.collect()
- self.samdb = SamDB(session_info=self.session,
- credentials=self.creds,
- lp=self.lp)
-
- # We need to hold this iterator open to hold the all-record lock.
- res = self.samdb.search_iterator()
-
- os.write(w2, b"start")
- if (os.read(r1, 7) != b"started"):
- os._exit(1)
-
- os.write(w2, b"add")
- if (os.read(r1, 5) != b"added"):
- os._exit(2)
-
- # Wait 2 seconds to block prepare_commit() in the child.
- os.write(w2, b"prepare")
- time.sleep(2)
-
- # Release the locks
- for l in res:
- pass
-
- if (os.read(r1, 8) != b"prepared"):
- os._exit(3)
-
- os._exit(0)
-
- # We can start the transaction during the search
- # because both just grab the all-record read lock.
- self.assertEqual(os.read(r2, 5), b"start")
- self.samdb.transaction_start()
- os.write(w1, b"started")
-
- self.assertEqual(os.read(r2, 3), b"add")
- dn = "cn=test_db_lock_user,cn=users," + str(basedn)
- self.samdb.add({
- "dn": dn,
- "objectclass": "user",
- })
- self.samdb.delete(dn)
- os.write(w1, b"added")
-
- # Obtain a write lock, this will block until
- # the parent releases the read lock.
- self.assertEqual(os.read(r2, 7), b"prepare")
- start = time.time()
- self.samdb.transaction_prepare_commit()
- end = time.time()
- try:
- self.assertGreater(end - start, 1.9)
- except:
- raise
- finally:
- os.write(w1, b"prepared")
-
- # Drop the write lock
- self.samdb.transaction_cancel()
-
- (got_pid, status) = os.waitpid(pid, 0)
- self.assertEqual(got_pid, pid)
- self.assertTrue(os.WIFEXITED(status))
- self.assertEqual(os.WEXITSTATUS(status), 0)
-
- def test_db_lock3(self):
- basedn = self.samdb.get_default_basedn()
- (r1, w1) = os.pipe()
- (r2, w2) = os.pipe()
-
- pid = os.fork()
- if pid == 0:
- # In the child, close the main DB, re-open
- del(self.samdb)
- gc.collect()
- self.samdb = SamDB(session_info=self.session,
- credentials=self.creds,
- lp=self.lp)
-
- # We need to hold this iterator open to hold the all-record lock.
- res = self.samdb.search_iterator()
-
- os.write(w2, b"start")
- if (os.read(r1, 7) != b"started"):
- os._exit(1)
-
- os.write(w2, b"add")
- if (os.read(r1, 5) != b"added"):
- os._exit(2)
-
- # Wait 2 seconds to block prepare_commit() in the child.
- os.write(w2, b"prepare")
- time.sleep(2)
-
- # Release the locks
- for l in res:
- pass
-
- if (os.read(r1, 8) != b"prepared"):
- os._exit(3)
-
- os._exit(0)
-
- # We can start the transaction during the search
- # because both just grab the all-record read lock.
- self.assertEqual(os.read(r2, 5), b"start")
- self.samdb.transaction_start()
- os.write(w1, b"started")
-
- self.assertEqual(os.read(r2, 3), b"add")
-
- # This will end up in the top level db
- dn = "@DSDB_LOCK_TEST"
- self.samdb.add({
- "dn": dn})
- self.samdb.delete(dn)
- os.write(w1, b"added")
-
- # Obtain a write lock, this will block until
- # the child releases the read lock.
- self.assertEqual(os.read(r2, 7), b"prepare")
- start = time.time()
- self.samdb.transaction_prepare_commit()
- end = time.time()
- self.assertGreater(end - start, 1.9)
- os.write(w1, b"prepared")
-
- # Drop the write lock
- self.samdb.transaction_cancel()
-
- (got_pid, status) = os.waitpid(pid, 0)
- self.assertTrue(os.WIFEXITED(status))
- self.assertEqual(os.WEXITSTATUS(status), 0)
- self.assertEqual(got_pid, pid)
-
-
- def _test_full_db_lock1(self, backend_path):
- (r1, w1) = os.pipe()
-
- pid = os.fork()
- if pid == 0:
- # In the child, close the main DB, re-open just one DB
- del(self.samdb)
- gc.collect()
-
- backenddb = ldb.Ldb(backend_path)
-
-
- backenddb.transaction_start()
-
- backenddb.add({"dn":"@DSDB_LOCK_TEST"})
- backenddb.delete("@DSDB_LOCK_TEST")
-
- # Obtain a write lock
- backenddb.transaction_prepare_commit()
- os.write(w1, b"prepared")
- time.sleep(2)
-
- # Drop the write lock
- backenddb.transaction_cancel()
- os._exit(0)
-
- self.assertEqual(os.read(r1, 8), b"prepared")
-
- start = time.time()
-
- # We need to hold this iterator open to hold the all-record lock.
- res = self.samdb.search_iterator()
-
- # This should take at least 2 seconds because the transaction
- # has a write lock on one backend db open
-
- end = time.time()
- self.assertGreater(end - start, 1.9)
-
- # Release the locks
- for l in res:
- pass
-
- (got_pid, status) = os.waitpid(pid, 0)
- self.assertEqual(got_pid, pid)
- self.assertTrue(os.WIFEXITED(status))
- self.assertEqual(os.WEXITSTATUS(status), 0)
-
- def test_full_db_lock1(self):
- basedn = self.samdb.get_default_basedn()
- backend_filename = "%s.ldb" % basedn.get_casefold()
- backend_subpath = os.path.join("sam.ldb.d",
- backend_filename)
- backend_path = self.lp.private_path(backend_subpath)
- self._test_full_db_lock1(backend_path)
-
-
- def test_full_db_lock1_config(self):
- basedn = self.samdb.get_config_basedn()
- backend_filename = "%s.ldb" % basedn.get_casefold()
- backend_subpath = os.path.join("sam.ldb.d",
- backend_filename)
- backend_path = self.lp.private_path(backend_subpath)
- self._test_full_db_lock1(backend_path)
-
-
- def _test_full_db_lock2(self, backend_path):
- (r1, w1) = os.pipe()
- (r2, w2) = os.pipe()
-
- pid = os.fork()
- if pid == 0:
-
- # In the child, close the main DB, re-open
- del(self.samdb)
- gc.collect()
- self.samdb = SamDB(session_info=self.session,
- credentials=self.creds,
- lp=self.lp)
-
- # We need to hold this iterator open to hold the all-record lock.
- res = self.samdb.search_iterator()
-
- os.write(w2, b"start")
- if (os.read(r1, 7) != b"started"):
- os._exit(1)
- os.write(w2, b"add")
- if (os.read(r1, 5) != b"added"):
- os._exit(2)
-
- # Wait 2 seconds to block prepare_commit() in the child.
- os.write(w2, b"prepare")
- time.sleep(2)
-
- # Release the locks
- for l in res:
- pass
-
- if (os.read(r1, 8) != b"prepared"):
- os._exit(3)
-
- os._exit(0)
-
- # In the parent, close the main DB, re-open just one DB
- del(self.samdb)
- gc.collect()
- backenddb = ldb.Ldb(backend_path)
-
- # We can start the transaction during the search
- # because both just grab the all-record read lock.
- self.assertEqual(os.read(r2, 5), b"start")
- backenddb.transaction_start()
- os.write(w1, b"started")
-
- self.assertEqual(os.read(r2, 3), b"add")
- backenddb.add({"dn":"@DSDB_LOCK_TEST"})
- backenddb.delete("@DSDB_LOCK_TEST")
- os.write(w1, b"added")
-
- # Obtain a write lock, this will block until
- # the child releases the read lock.
- self.assertEqual(os.read(r2, 7), b"prepare")
- start = time.time()
- backenddb.transaction_prepare_commit()
- end = time.time()
-
- try:
- self.assertGreater(end - start, 1.9)
- except:
- raise
- finally:
- os.write(w1, b"prepared")
-
- # Drop the write lock
- backenddb.transaction_cancel()
-
- (got_pid, status) = os.waitpid(pid, 0)
- self.assertEqual(got_pid, pid)
- self.assertTrue(os.WIFEXITED(status))
- self.assertEqual(os.WEXITSTATUS(status), 0)
-
- def test_full_db_lock2(self):
- basedn = self.samdb.get_default_basedn()
- backend_filename = "%s.ldb" % basedn.get_casefold()
- backend_subpath = os.path.join("sam.ldb.d",
- backend_filename)
- backend_path = self.lp.private_path(backend_subpath)
- self._test_full_db_lock2(backend_path)
-
- def test_full_db_lock2_config(self):
- basedn = self.samdb.get_config_basedn()
- backend_filename = "%s.ldb" % basedn.get_casefold()
- backend_subpath = os.path.join("sam.ldb.d",
- backend_filename)
- backend_path = self.lp.private_path(backend_subpath)
- self._test_full_db_lock2(backend_path)
-
def test_no_error_on_invalid_control(self):
try:
res = self.samdb.search(scope=ldb.SCOPE_SUBTREE,
--- /dev/null
+# Unix SMB/CIFS implementation. Tests for SamDB
+# Copyright (C) Jelmer Vernooij <jelmer@samba.org> 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/>.
+#
+
+"""Tests for samba's dsdb modules"""
+
+from samba.tests.samdb import SamDBTestCase
+from samba.samdb import SamDB
+import ldb
+import os
+import samba
+import gc
+import time
+
+class DsdbLockTestCase(SamDBTestCase):
+ def test_db_lock1(self):
+ basedn = self.samdb.get_default_basedn()
+ (r1, w1) = os.pipe()
+
+ pid = os.fork()
+ if pid == 0:
+ # In the child, close the main DB, re-open just one DB
+ del(self.samdb)
+ gc.collect()
+ self.samdb = SamDB(session_info=self.session,
+ lp=self.lp)
+
+ self.samdb.transaction_start()
+
+ dn = "cn=test_db_lock_user,cn=users," + str(basedn)
+ self.samdb.add({
+ "dn": dn,
+ "objectclass": "user",
+ })
+ self.samdb.delete(dn)
+
+ # Obtain a write lock
+ self.samdb.transaction_prepare_commit()
+ os.write(w1, b"prepared")
+ time.sleep(2)
+
+ # Drop the write lock
+ self.samdb.transaction_cancel()
+ os._exit(0)
+
+ self.assertEqual(os.read(r1, 8), b"prepared")
+
+ start = time.time()
+
+ # We need to hold this iterator open to hold the all-record lock.
+ res = self.samdb.search_iterator()
+
+ # This should take at least 2 seconds because the transaction
+ # has a write lock on one backend db open
+
+ # Release the locks
+ for l in res:
+ pass
+
+ end = time.time()
+ self.assertGreater(end - start, 1.9)
+
+ (got_pid, status) = os.waitpid(pid, 0)
+ self.assertEqual(got_pid, pid)
+ self.assertTrue(os.WIFEXITED(status))
+ self.assertEqual(os.WEXITSTATUS(status), 0)
+
+ def test_db_lock2(self):
+ basedn = self.samdb.get_default_basedn()
+ (r1, w1) = os.pipe()
+ (r2, w2) = os.pipe()
+
+ pid = os.fork()
+ if pid == 0:
+ # In the child, close the main DB, re-open
+ del(self.samdb)
+ gc.collect()
+ self.samdb = SamDB(session_info=self.session,
+ lp=self.lp)
+
+ # We need to hold this iterator open to hold the all-record lock.
+ res = self.samdb.search_iterator()
+
+ os.write(w2, b"start")
+ if (os.read(r1, 7) != b"started"):
+ os._exit(1)
+
+ os.write(w2, b"add")
+ if (os.read(r1, 5) != b"added"):
+ os._exit(2)
+
+ # Wait 2 seconds to block prepare_commit() in the child.
+ os.write(w2, b"prepare")
+ time.sleep(2)
+
+ # Release the locks
+ for l in res:
+ pass
+
+ if (os.read(r1, 8) != b"prepared"):
+ os._exit(3)
+
+ os._exit(0)
+
+ # We can start the transaction during the search
+ # because both just grab the all-record read lock.
+ self.assertEqual(os.read(r2, 5), b"start")
+ self.samdb.transaction_start()
+ os.write(w1, b"started")
+
+ self.assertEqual(os.read(r2, 3), b"add")
+ dn = "cn=test_db_lock_user,cn=users," + str(basedn)
+ self.samdb.add({
+ "dn": dn,
+ "objectclass": "user",
+ })
+ self.samdb.delete(dn)
+ os.write(w1, b"added")
+
+ # Obtain a write lock, this will block until
+ # the parent releases the read lock.
+ self.assertEqual(os.read(r2, 7), b"prepare")
+ start = time.time()
+ self.samdb.transaction_prepare_commit()
+ end = time.time()
+ try:
+ self.assertGreater(end - start, 1.9)
+ except:
+ raise
+ finally:
+ os.write(w1, b"prepared")
+
+ # Drop the write lock
+ self.samdb.transaction_cancel()
+
+ (got_pid, status) = os.waitpid(pid, 0)
+ self.assertEqual(got_pid, pid)
+ self.assertTrue(os.WIFEXITED(status))
+ self.assertEqual(os.WEXITSTATUS(status), 0)
+
+ def test_db_lock3(self):
+ basedn = self.samdb.get_default_basedn()
+ (r1, w1) = os.pipe()
+ (r2, w2) = os.pipe()
+
+ pid = os.fork()
+ if pid == 0:
+ # In the child, close the main DB, re-open
+ del(self.samdb)
+ gc.collect()
+ self.samdb = SamDB(session_info=self.session,
+ lp=self.lp)
+
+ # We need to hold this iterator open to hold the all-record lock.
+ res = self.samdb.search_iterator()
+
+ os.write(w2, b"start")
+ if (os.read(r1, 7) != b"started"):
+ os._exit(1)
+
+ os.write(w2, b"add")
+ if (os.read(r1, 5) != b"added"):
+ os._exit(2)
+
+ # Wait 2 seconds to block prepare_commit() in the child.
+ os.write(w2, b"prepare")
+ time.sleep(2)
+
+ # Release the locks
+ for l in res:
+ pass
+
+ if (os.read(r1, 8) != b"prepared"):
+ os._exit(3)
+
+ os._exit(0)
+
+ # We can start the transaction during the search
+ # because both just grab the all-record read lock.
+ self.assertEqual(os.read(r2, 5), b"start")
+ self.samdb.transaction_start()
+ os.write(w1, b"started")
+
+ self.assertEqual(os.read(r2, 3), b"add")
+
+ # This will end up in the top level db
+ dn = "@DSDB_LOCK_TEST"
+ self.samdb.add({
+ "dn": dn})
+ self.samdb.delete(dn)
+ os.write(w1, b"added")
+
+ # Obtain a write lock, this will block until
+ # the child releases the read lock.
+ self.assertEqual(os.read(r2, 7), b"prepare")
+ start = time.time()
+ self.samdb.transaction_prepare_commit()
+ end = time.time()
+ self.assertGreater(end - start, 1.9)
+ os.write(w1, b"prepared")
+
+ # Drop the write lock
+ self.samdb.transaction_cancel()
+
+ (got_pid, status) = os.waitpid(pid, 0)
+ self.assertTrue(os.WIFEXITED(status))
+ self.assertEqual(os.WEXITSTATUS(status), 0)
+ self.assertEqual(got_pid, pid)
+
+
+ def _test_full_db_lock1(self, backend_path):
+ (r1, w1) = os.pipe()
+
+ pid = os.fork()
+ if pid == 0:
+ # In the child, close the main DB, re-open just one DB
+ del(self.samdb)
+ gc.collect()
+
+ backenddb = ldb.Ldb(backend_path)
+
+
+ backenddb.transaction_start()
+
+ backenddb.add({"dn":"@DSDB_LOCK_TEST"})
+ backenddb.delete("@DSDB_LOCK_TEST")
+
+ # Obtain a write lock
+ backenddb.transaction_prepare_commit()
+ os.write(w1, b"prepared")
+ time.sleep(2)
+
+ # Drop the write lock
+ backenddb.transaction_cancel()
+ os._exit(0)
+
+ self.assertEqual(os.read(r1, 8), b"prepared")
+
+ start = time.time()
+
+ # We need to hold this iterator open to hold the all-record lock.
+ res = self.samdb.search_iterator()
+
+ # This should take at least 2 seconds because the transaction
+ # has a write lock on one backend db open
+
+ end = time.time()
+ self.assertGreater(end - start, 1.9)
+
+ # Release the locks
+ for l in res:
+ pass
+
+ (got_pid, status) = os.waitpid(pid, 0)
+ self.assertEqual(got_pid, pid)
+ self.assertTrue(os.WIFEXITED(status))
+ self.assertEqual(os.WEXITSTATUS(status), 0)
+
+ def test_full_db_lock1(self):
+ basedn = self.samdb.get_default_basedn()
+ backend_filename = "%s.ldb" % basedn.get_casefold()
+ backend_subpath = os.path.join("sam.ldb.d",
+ backend_filename)
+ backend_path = self.lp.private_path(backend_subpath)
+ self._test_full_db_lock1(backend_path)
+
+
+ def test_full_db_lock1_config(self):
+ basedn = self.samdb.get_config_basedn()
+ backend_filename = "%s.ldb" % basedn.get_casefold()
+ backend_subpath = os.path.join("sam.ldb.d",
+ backend_filename)
+ backend_path = self.lp.private_path(backend_subpath)
+ self._test_full_db_lock1(backend_path)
+
+
+ def _test_full_db_lock2(self, backend_path):
+ (r1, w1) = os.pipe()
+ (r2, w2) = os.pipe()
+
+ pid = os.fork()
+ if pid == 0:
+
+ # In the child, close the main DB, re-open
+ del(self.samdb)
+ gc.collect()
+ self.samdb = SamDB(session_info=self.session,
+ lp=self.lp)
+
+ # We need to hold this iterator open to hold the all-record lock.
+ res = self.samdb.search_iterator()
+
+ os.write(w2, b"start")
+ if (os.read(r1, 7) != b"started"):
+ os._exit(1)
+ os.write(w2, b"add")
+ if (os.read(r1, 5) != b"added"):
+ os._exit(2)
+
+ # Wait 2 seconds to block prepare_commit() in the child.
+ os.write(w2, b"prepare")
+ time.sleep(2)
+
+ # Release the locks
+ for l in res:
+ pass
+
+ if (os.read(r1, 8) != b"prepared"):
+ os._exit(3)
+
+ os._exit(0)
+
+ # In the parent, close the main DB, re-open just one DB
+ del(self.samdb)
+ gc.collect()
+ backenddb = ldb.Ldb(backend_path)
+
+ # We can start the transaction during the search
+ # because both just grab the all-record read lock.
+ self.assertEqual(os.read(r2, 5), b"start")
+ backenddb.transaction_start()
+ os.write(w1, b"started")
+
+ self.assertEqual(os.read(r2, 3), b"add")
+ backenddb.add({"dn":"@DSDB_LOCK_TEST"})
+ backenddb.delete("@DSDB_LOCK_TEST")
+ os.write(w1, b"added")
+
+ # Obtain a write lock, this will block until
+ # the child releases the read lock.
+ self.assertEqual(os.read(r2, 7), b"prepare")
+ start = time.time()
+ backenddb.transaction_prepare_commit()
+ end = time.time()
+
+ try:
+ self.assertGreater(end - start, 1.9)
+ except:
+ raise
+ finally:
+ os.write(w1, b"prepared")
+
+ # Drop the write lock
+ backenddb.transaction_cancel()
+
+ (got_pid, status) = os.waitpid(pid, 0)
+ self.assertEqual(got_pid, pid)
+ self.assertTrue(os.WIFEXITED(status))
+ self.assertEqual(os.WEXITSTATUS(status), 0)
+
+ def test_full_db_lock2(self):
+ basedn = self.samdb.get_default_basedn()
+ backend_filename = "%s.ldb" % basedn.get_casefold()
+ backend_subpath = os.path.join("sam.ldb.d",
+ backend_filename)
+ backend_path = self.lp.private_path(backend_subpath)
+ self._test_full_db_lock2(backend_path)
+
+ def test_full_db_lock2_config(self):
+ basedn = self.samdb.get_config_basedn()
+ backend_filename = "%s.ldb" % basedn.get_casefold()
+ backend_subpath = os.path.join("sam.ldb.d",
+ backend_filename)
+ backend_path = self.lp.private_path(backend_subpath)
+ self._test_full_db_lock2(backend_path)
import logging
import os
import uuid
+import shutil
from samba.auth import system_session
-from samba.provision import (setup_samdb, guess_names, make_smbconf,
- provision_paths_from_lp)
-from samba.provision import DEFAULT_POLICY_GUID, DEFAULT_DC_POLICY_GUID
-from samba.provision.backend import ProvisionBackend
+from samba.provision import provision
from samba.tests import TestCaseInTempDir
-from samba.dcerpc import security
-from samba.schema import Schema
-from samba import param
+from samba.dsdb import DS_DOMAIN_FUNCTION_2008_R2
class SamDBTestCase(TestCaseInTempDir):
def setUp(self):
super(SamDBTestCase, self).setUp()
- invocationid = str(uuid.uuid4())
- domaindn = "DC=COM,DC=EXAMPLE"
- self.domaindn = domaindn
- configdn = "CN=Configuration," + domaindn
- schemadn = "CN=Schema," + configdn
- domainguid = str(uuid.uuid4())
- policyguid = DEFAULT_POLICY_GUID
- domainsid = security.random_sid()
- path = os.path.join(self.tempdir, "samdb.ldb")
- session_info = system_session()
-
- hostname="foo"
- domain="EXAMPLE"
- dnsdomain="example.com"
- serverrole="domain controller"
- policyguid_dc = DEFAULT_DC_POLICY_GUID
-
- smbconf = os.path.join(self.tempdir, "smb.conf")
- make_smbconf(smbconf, hostname, domain, dnsdomain,
- self.tempdir, serverrole=serverrole)
-
- self.lp = param.LoadParm()
- self.lp.load(smbconf)
-
- names = guess_names(lp=self.lp, hostname=hostname,
- domain=domain, dnsdomain=dnsdomain,
- serverrole=serverrole,
- domaindn=self.domaindn, configdn=configdn,
- schemadn=schemadn)
-
- paths = provision_paths_from_lp(self.lp, names.dnsdomain)
-
- logger = logging.getLogger("provision")
-
- provision_backend = ProvisionBackend("ldb", paths=paths,
- lp=self.lp, credentials=None,
- names=names, logger=logger)
-
- schema = Schema(domainsid, invocationid=invocationid,
- schemadn=names.schemadn, serverdn=names.serverdn,
- am_rodc=False)
-
- self.samdb = setup_samdb(path, session_info,
- provision_backend, self.lp, names, logger,
- domainsid, domainguid, policyguid, policyguid_dc, False,
- "secret", "secret", "secret", invocationid, "secret",
- None, "domain controller", schema=schema)
+ self.session = system_session()
+ logger = logging.getLogger("selftest")
+ domain = "dsdb"
+ realm = "dsdb.samba.example.com"
+ host_name = "test"
+ server_role = "active directory domain controller"
+ dns_backend = "SAMBA_INTERNAL"
+ self.result = provision(logger,
+ self.session, targetdir=self.tempdir,
+ realm=realm, domain=domain,
+ hostname=host_name,
+ use_ntvfs=True,
+ serverrole=server_role,
+ dns_backend="SAMBA_INTERNAL",
+ dom_for_fun_level=DS_DOMAIN_FUNCTION_2008_R2)
+ self.samdb = self.result.samdb
+ self.lp = self.result.lp
def tearDown(self):
- for f in ['schema.ldb', 'configuration.ldb',
- 'users.ldb', 'samdb.ldb', 'smb.conf']:
+ for f in ['names.tdb']:
os.remove(os.path.join(self.tempdir, f))
+
+ for d in ['etc', 'msg.lock', 'private', 'state']:
+ shutil.rmtree(os.path.join(self.tempdir, d))
+
super(SamDBTestCase, self).tearDown()
planoldpythontestsuite("none", "simple", extra_path=["%s/lib/tdb/python/tests" % srcdir()], name="tdb.python")
planpythontestsuite("ad_dc_ntvfs:local", "samba.tests.dcerpc.sam")
planpythontestsuite("ad_dc_ntvfs:local", "samba.tests.dsdb")
+planpythontestsuite("none", "samba.tests.dsdb_lock")
planpythontestsuite("ad_dc_ntvfs:local", "samba.tests.dcerpc.bare")
planpythontestsuite("ad_dc_ntvfs:local", "samba.tests.dcerpc.unix")
planpythontestsuite("ad_dc_ntvfs:local", "samba.tests.dcerpc.srvsvc")
git.samba.org / metze / samba / wip.git / commitdiff