Commits

Thomas Waldmann committed 0765b7e

kyoto cabinet tests: fix some minor issues

Comments (0)

Files changed (8)

storage/_tests/__init__.py

 class BytesStorageTestBase(object):
     pass
 
-
-
-
-
-

storage/_tests/conftest.py

+# Copyright: 2011 MoinMoin:RonnyPfannschmidt
+# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
+
+"""
+MoinMoin - storage test magic
+"""
+
+
+from __future__ import absolute_import, division
+
 import pytest
 from storage.wrappers import ByteToStreamWrappingStore
 
 # memcached is not in the loop
 stores = 'fs kc memory sqlite sqlite:compressed'.split()
 
+
 constructors = {
     'memory': lambda store, _: store(),
     'fs': lambda store, tmpdir: store(str(tmpdir.join('store'))),
                                           'test_table', compression_level=0),
     'sqlite:compressed': lambda store, tmpdir: store(str(tmpdir.join('store.sqlite')),
                                           'test_table', compression_level=1),
-    'kc': lambda store, tmpdir: store(str(tmpdir.join('store'))),
+    'kc': lambda store, tmpdir: store(str(tmpdir.join('store.kch'))),
 }
 
 
 
     multi_mark = getattr(metafunc.function, 'multi', None)
     if multi_mark is not None:
-        #XXX: hack
+        # XXX: hack
         storages = multi_mark.kwargs['Storage']
         for storage in storages:
             metafunc.addcall(id=storage.__name__, funcargs={
     store = construct(klass, tmpdir)
     store.create()
     store.open()
-
     # no destroy in the normal finalizer
-    # so we can keep the data for example if its a tmpdir
+    # so we can keep the data for example if it's a tmpdir
     request.addfinalizer(store.close)
     return store
 

storage/_tests/test_all.py

+# Copyright: 2011 MoinMoin:RonnyPfannschmidt
+# Copyright: 2011 MoinMoin:ThomasWaldmann
+# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
+
+"""
+MoinMoin - storage tests
+"""
+
+
+from __future__ import absolute_import, division
+
 import pytest
 
 def test_getitem_raises(store):
     del store['foo']
     assert len(store) == 0
 
+
 def test_perf(store):
-    #XXX: introduce perf test option
+    # XXX: introduce perf test option
     pytest.skip("usually we do no performance tests")
     for i in range(1000):
         key = value = str(i)
     for i in range(1000):
         key = str(i)
         del store[key]
+

storage/_tests/test_fs.py

+# Copyright: 2011 MoinMoin:RonnyPfannschmidt
 # Copyright: 2011 MoinMoin:ThomasWaldmann
 # License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
 

storage/_tests/test_kc.py

+# Copyright: 2011 MoinMoin:RonnyPfannschmidt
 # Copyright: 2011 MoinMoin:ThomasWaldmann
 # License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
 
 
 @pytest.mark.multi(Storage=[BytesStorage, FileStorage])
 def test_create(tmpdir, Storage):
-    target = tmpdir.join('store.kc')
+    target = tmpdir.join('store.kch')
     assert not target.check()
 
     store = Storage(str(target))
 @pytest.mark.multi(Storage=[BytesStorage, FileStorage])
 def test_destroy(tmpdir, Storage):
     store = test_create(tmpdir, Storage)
-    target = tmpdir.join('store.kc')
+    target = tmpdir.join('store.kch')
     store.destroy()
     assert not target.check()
 

storage/_tests/test_memory.py

+# Copyright: 2011 MoinMoin:RonnyPfannschmidt
 # Copyright: 2011 MoinMoin:ThomasWaldmann
 # License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
 
 import pytest
 from storage.memory import BytesStorage, FileStorage
 
+
 @pytest.mark.multi(Storage=[BytesStorage, FileStorage])
-def test_create( Storage):
+def test_create(Storage):
     store = Storage()
     assert store._st is None
 
     store.destroy()
     assert store._st is None
 
-

storage/_tests/test_sqlite.py

+# Copyright: 2011 MoinMoin:RonnyPfannschmidt
 # Copyright: 2011 MoinMoin:ThomasWaldmann
 # License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
 
     dbfile = tmpdir.join('store.sqlite')
     store = test_create(tmpdir, Storage)
     store.destroy()
-    #XXX: check for dropped table
+    # XXX: check for dropped table
 
-
 
 from __future__ import absolute_import, division
 
+import os
 from StringIO import StringIO
 
 from kyotocabinet import *
         self.close()
 
     def destroy(self):
-        self.open(mode=self.mode|DB.OTRUNCATE)
-        self.close()
+        os.remove(self.path)
 
     def open(self, mode=None):
         self._db = DB(self.db_opts)