def test_max_dbs(self):
     self.assertRaises(OverflowError,
         lambda: testlib.temp_env(max_dbs=-1))
     for val in 0, 10, 20:
         _, env = testlib.temp_env(max_dbs=val)
         dbs = [env.open_db(B('db%d' % i)) for i in range(val)]
         self.assertRaises(lmdb.DbsFullError,
             lambda: env.open_db(B('toomany')))

 def test_dupsort(self):
     _, env = testlib.temp_env()
     db1 = env.open_db(B('db1'), dupsort=True)
     txn = env.begin(write=True, db=db1)
     assert txn.put(B('a'), B('a'))
     assert txn.put(B('a'), B('b'))
     assert txn.get(B('a')) == B('a')

 def test_integerdup(self):
     _, env = testlib.temp_env()
     db1 = env.open_db(B('db1'), dupsort=True, integerdup=True)
     txn = env.begin(write=True, db=db1)
     assert txn.put(UINT_0001, UINT_0002)
     assert txn.put(UINT_0001, UINT_0001)
     assert txn.get(UINT_0001) == UINT_0001

 def test_big_reverse(self):
     path, env = testlib.temp_env()
     txn = env.begin(write=True)
     keys = [B('%05d' % i) for i in range(0xffff)]
     for k in keys:
         txn.put(k, k, append=True)
     assert list(txn.cursor().iterprev(values=False)) == list(reversed(keys))

 def test_sync(self):
     _, env = testlib.temp_env()
     env.sync(False)
     env.sync(True)
     env.close()
     self.assertRaises(Exception,
         lambda: env.sync(False))

 def test_readonly_false(self):
     path, env = testlib.temp_env(readonly=False)
     with env.begin(write=True) as txn:
         txn.put(B('a'), B(''))
     with env.begin() as txn:
         assert txn.get(B('a')) == B('')
     assert not env.flags()['readonly']

    def test_reader_check(self):
        path, env = testlib.temp_env(max_spare_txns=0)
        rc = env.reader_check()
        assert rc == 0

        # We need to open a separate env since Transaction.abort() always calls
        # reset for a read-only txn, the actual abort doesn't happen until
        # __del__, when Transaction discovers there is no room for it on the
        # freelist.
        env1 = lmdb.open(path)
        txn1 = env1.begin()
        assert env.readers() != NO_READERS
        assert env.reader_check() == 0

        # Start a child, open a txn, then crash the child.
        rc = os.spawnl(os.P_WAIT, sys.executable, sys.executable,
                       __file__, 'test_reader_check_child', path)

        assert rc == 0
        assert env.reader_check() == 1
        assert env.reader_check() == 0
        assert env.readers() != NO_READERS

        txn1.abort()
        env1.close()
        assert env.readers() == NO_READERS

        env.close()
        self.assertRaises(Exception,
            lambda: env.reader_check())

    def test_reader_check(self):
        path, env = testlib.temp_env()
        rc = env.reader_check()
        assert rc == 0

        txn1 = env.begin()
        assert env.readers() != NO_READERS
        assert env.reader_check() == 0

        # Start a child, open a txn, then crash the child.
        child_pid = os.fork()
        if not child_pid:
            env2 = lmdb.open(path)
            txn = env2.begin()
            os.kill(os.getpid(), signal.SIGKILL)

        assert os.waitpid(child_pid, 0) == (child_pid, signal.SIGKILL)
        assert env.reader_check() == 1
        assert env.reader_check() == 0
        assert env.readers() != NO_READERS
        txn1.abort()
        assert env.readers() == NO_READERS

        env.close()
        self.assertRaises(Exception,
            lambda: env.reader_check())

 def test_begin_readonly(self):
     _, env = testlib.temp_env()
     txn = env.begin()
     # Read txn can't write.
     self.assertRaises(lmdb.ReadonlyError,
         lambda: txn.put(B('a'), B('')))
     txn.abort()

 def test_readonly(self):
     _, env = testlib.temp_env()
     txn = lmdb.Transaction(env)
     # Read txn can't write.
     self.assertRaises(lmdb.ReadonlyError,
         lambda: txn.put(B('a'), B('')))
     txn.abort()

 def test_abort_rw(self):
     _, env = testlib.temp_env()
     txn = env.begin(write=True)
     assert txn.put(B('a'), B('a'))
     txn.abort()
     txn = env.begin()
     assert txn.get(B('a')) is None

 def test_ok(self):
     path, env = testlib.temp_env()
     with env as env_:
         assert env_ is env
         with env.begin() as txn:
             txn.get(B('foo'))
     self.assertRaises(Exception, lambda: env.begin())

    def test_one(self):
        _, env = testlib.temp_env(max_spare_txns=1)
        # 1 here, since CFFI uses a temporary reader during init.
        assert 1 >= reader_count(env)

        t1 = env.begin()
        assert 1 == reader_count(env)

        t2 = env.begin()
        assert 2 == reader_count(env)

        t1.abort()
        del t1
        assert 2 == reader_count(env)  # 1 live, 1 cached

        t2.abort()
        del t2
        assert 1 == reader_count(env)  # 1 cached

        t3 = env.begin()
        assert 1 == reader_count(env)  # 1 live

        t3.abort()
        del t3
        assert 1 == reader_count(env)  # 1 cached

    def test_stat(self):
        _, env = testlib.temp_env()
        db1 = env.open_db(B('db1'))
        db2 = env.open_db(B('db2'))

        txn = lmdb.Transaction(env)
        for db in db1, db2:
            stat = txn.stat(db)
            for k in 'psize', 'depth', 'branch_pages', 'overflow_pages',\
                     'entries':
                assert isinstance(stat[k], INT_TYPES), k
                assert stat[k] >= 0
            assert stat['entries'] == 0

        txn = lmdb.Transaction(env, write=True)
        txn.put(B('a'), B('b'), db=db1)
        txn.commit()

        txn = lmdb.Transaction(env)
        stat = txn.stat(db1)
        assert stat['entries'] == 1

        stat = txn.stat(db2)
        assert stat['entries'] == 0

        txn.abort()
        self.assertRaises(Exception,
            lambda: env.stat(db1))
        env.close()
        self.assertRaises(Exception,
            lambda: env.stat(db1))

 def test_subdir_false_ok(self):
     path = testlib.temp_file(create=False)
     _, env = testlib.temp_env(path, subdir=False)
     assert os.path.exists(path)
     assert os.path.isfile(path)
     assert os.path.isfile(path + '-lock')
     assert not env.flags()['subdir']

 def setUp(self):
     self.path, self.env = testlib.temp_env()
     db = self.env.open_db(B("db1"), dupsort=True)
     self.txn = self.env.begin(db, write=True)
     for _ in range(2):
         putData(self.txn)
     self.c = self.txn.cursor()
     self.empty_entry = ("", "")

    def test_readonly_env_sub_eperm(self):
        # https://github.com/dw/py-lmdb/issues/109
        path, env = testlib.temp_env()
        env.close()

        env = lmdb.open(path, max_dbs=10, readonly=True)
        self.assertRaises(lmdb.ReadonlyError,
            lambda: env.open_db(B('node_schedules'), create=True))

 def test_integerkey(self):
     _, env = testlib.temp_env()
     db1 = env.open_db(B('db1'), integerkey=True)
     txn = env.begin(write=True, db=db1)
     assert txn.put(UINT_0001, B('a'))
     assert txn.put(UINT_0002, B('b'))
     assert txn.get(UINT_0001) == B('a')
     assert txn.get(UINT_0002) == B('b')

    def test_path(self):
        path, env = testlib.temp_env()
        assert path == env.path()
        assert isinstance(env.path(), UnicodeType)

        env.close()
        self.assertRaises(Exception,
            lambda: env.path())

 def test_sub_txn(self):
     _, env = testlib.temp_env()
     txn = env.begin(write=True)
     db1 = env.open_db(B('subdb1'), txn=txn)
     db2 = env.open_db(B('subdb2'), txn=txn)
     for db in db1, db2:
         assert db.flags(txn) == {'reverse_key': False, 'dupsort': False}
     txn.commit()