import gc
import time
+
class DsdbLockTestCase(SamDBTestCase):
def test_db_lock1(self):
basedn = self.samdb.get_default_basedn()
self.assertEqual(os.WEXITSTATUS(status), 0)
self.assertEqual(got_pid, pid)
-
def _test_full_db_lock1(self, backend_path):
(r1, w1) = os.pipe()
backenddb = ldb.Ldb(backend_path)
-
backenddb.transaction_start()
- backenddb.add({"dn":"@DSDB_LOCK_TEST"})
+ backenddb.add({"dn": "@DSDB_LOCK_TEST"})
backenddb.delete("@DSDB_LOCK_TEST")
# Obtain a write lock
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_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()
os.write(w1, b"started")
self.assertEqual(os.read(r2, 3), b"add")
- backenddb.add({"dn":"@DSDB_LOCK_TEST"})
+ backenddb.add({"dn": "@DSDB_LOCK_TEST"})
backenddb.delete("@DSDB_LOCK_TEST")
os.write(w1, b"added")