Converted tests to nose.

src/whoosh/support/testing.py

@@ -1,5 +1,6 @@
 import shutil
 import tempfile
+from functools import wraps
 
 from whoosh.filedb.filestore import FileStorage
 
@@ -19,16 +20,17 @@ def __enter__(self):
         return FileStorage(self.dir)
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        if exc_type is not None:
-            if exc_type not in self.suppress:
-                print "Temp dir=", self.dir
-                return False
-
         if not self.keepdir:
             try:
                 shutil.rmtree(self.dir)
             except OSError, e:
                 print "Can't remove temp dir: " + str(e)
+
+        if exc_type is not None:
+            if self.keepdir:
+                print "Temp dir=", self.dir
+            if exc_type not in self.suppress:
+                return False
 
 
 class TempIndex(TempStorage):
@@ -41,3 +43,43 @@ def __enter__(self):
         return fstore.create_index(self.schema, indexname=self.basename)
 
 
+def skip_if(cond):
+    """A Nose test decorator that skips the decorated test if the given
+    function returns True at runtime.
+    """
+
+    def decorating_function(testfn):
+        @wraps(testfn)
+        def wrapper(*args, **kwargs):
+            if cond():
+                from nose.plugins.skip import SkipTest
+                raise SkipTest
+            else:
+                return testfn(*args, **kwargs)
+
+        return wrapper
+    return decorating_function
+
+
+def skip_if_unavailable(modulename):
+    """A Nose test decorator that only runs the decorated test if a module
+    can be imported::
+    
+        @skip_if_unavailable("multiprocessing")
+        def test_mp():
+    
+    Raises ``SkipTest`` if the module cannot be imported.
+    """
+
+    def cantimport():
+        try:
+            __import__(modulename)
+        except ImportError:
+            return True
+        else:
+            return False
+
+    return skip_if(cantimport)
+
+
+

stress/test_bigindex.py

@@ -1,5 +1,4 @@
 from __future__ import with_statement
-import unittest
 
 import random
 
@@ -8,66 +7,63 @@
 from whoosh.util import now
 
 
-class Test(unittest.TestCase):
-    def test_20000_single(self):
-        sc = fields.Schema(id=fields.ID(stored=True), text=fields.TEXT)
-        with TempIndex(sc, "20000single") as ix:
-            domain = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot",
-                      "golf", "hotel", "india", "juliet", "kilo", "lima"]
-
-            t = now()
-            for i in xrange(20000):
-                w = ix.writer()
-                w.add_document(id=unicode(i), text=u" ".join(random.sample(domain, 5)))
-                w.commit()
-            print "Write single:", now() - t
-
-            t = now()
-            ix.optimize()
-            print "Optimize single:", now() - t
-
-    def test_20000_buffered(self):
-        from whoosh.writing import BufferedWriter
+def test_20000_single():
+    sc = fields.Schema(id=fields.ID(stored=True), text=fields.TEXT)
+    with TempIndex(sc, "20000single") as ix:
+        domain = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot",
+                  "golf", "hotel", "india", "juliet", "kilo", "lima"]
 
-        sc = fields.Schema(id=fields.ID(stored=True), text=fields.TEXT)
-        with TempIndex(sc, "20000buffered") as ix:
-            domain = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot",
-                      "golf", "hotel", "india", "juliet", "kilo", "lima"]
-
-            t = now()
-            w = BufferedWriter(ix, limit=100, period=None)
-            for i in xrange(20000):
-                w.add_document(id=unicode(i),
-                               text = u" ".join(random.sample(domain, 5)))
-            w.close()
-            print "Write buffered:", now() - t
-
-            t = now()
-            ix.optimize()
-            print "Optimize buffered:", now() - t
-
-    def test_20000_batch(self):
-        sc = fields.Schema(id=fields.ID(stored=True), text=fields.TEXT)
-        with TempIndex(sc, "20000batch") as ix:
-            domain = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot",
-                      "golf", "hotel", "india", "juliet", "kilo", "lima"]
-
-            t = now()
+        t = now()
+        for i in xrange(20000):
             w = ix.writer()
-            for i in xrange(20000):
-                w.add_document(id=unicode(i),
-                               text = u" ".join(random.sample(domain, 5)))
-                if not i % 100:
-                    w.commit()
-                    w = ix.writer()
+            w.add_document(id=unicode(i), text=u" ".join(random.sample(domain, 5)))
             w.commit()
-            print "Write batch:", now() - t
-
-            t = now()
-            ix.optimize()
-            print "Optimize batch:", now() - t
+        print "Write single:", now() - t
+
+        t = now()
+        ix.optimize()
+        print "Optimize single:", now() - t
+
+def test_20000_buffered():
+    from whoosh.writing import BufferedWriter
+
+    sc = fields.Schema(id=fields.ID(stored=True), text=fields.TEXT)
+    with TempIndex(sc, "20000buffered") as ix:
+        domain = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot",
+                  "golf", "hotel", "india", "juliet", "kilo", "lima"]
+
+        t = now()
+        w = BufferedWriter(ix, limit=100, period=None)
+        for i in xrange(20000):
+            w.add_document(id=unicode(i),
+                           text = u" ".join(random.sample(domain, 5)))
+        w.close()
+        print "Write buffered:", now() - t
+
+        t = now()
+        ix.optimize()
+        print "Optimize buffered:", now() - t
+
+def test_20000_batch():
+    sc = fields.Schema(id=fields.ID(stored=True), text=fields.TEXT)
+    with TempIndex(sc, "20000batch") as ix:
+        domain = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot",
+                  "golf", "hotel", "india", "juliet", "kilo", "lima"]
+
+        t = now()
+        w = ix.writer()
+        for i in xrange(20000):
+            w.add_document(id=unicode(i),
+                           text = u" ".join(random.sample(domain, 5)))
+            if not i % 100:
+                w.commit()
+                w = ix.writer()
+        w.commit()
+        print "Write batch:", now() - t
+
+        t = now()
+        ix.optimize()
+        print "Optimize batch:", now() - t
 
 
 
-if __name__ == "__main__":
-    unittest.main()

stress/test_bigsort.py

@@ -1,68 +1,65 @@
-import unittest
-
 import os.path, random, shutil
 from datetime import datetime
 
-from whoosh import fields, index, query, scoring
+from whoosh import fields, index, query
 from whoosh.util import now
 
 
-class TestSorting(unittest.TestCase):
-    def test_bigsort(self):
-        times = 30000
-        dirname = "testindex"
-
-        df = fields.DATETIME(stored=True)
-        schema = fields.Schema(id=fields.ID(stored=True), date=df)
-
-        if os.path.exists(dirname):
-            shutil.rmtree(dirname)
-        os.mkdir(dirname)
-        ix = index.create_in(dirname, schema)
-
-        print "Writing..."
-        t = now()
-        w = ix.writer(limitmb=512)
-        for i in xrange(times):
-            dt = datetime.fromtimestamp(random.randint(15839593, 1294102139))
-            w.add_document(id=unicode(i), date=dt)
-        w.commit()
-        print "Writing took ", now() - t
-
-        ix = index.open_dir(dirname)
-        s = ix.searcher()
-        q = query.Wildcard("id", "1?2*")
-
-        t = now()
-        x = list(df.sortable_values(s.reader(), "date"))
-        print now() - t, len(x)
-
-        t = now()
-        for y in x:
-            p = list(s.postings("date", y).all_ids())
-        print now() - t
-
-
-
-        t = now()
-        r = s.search(q, limit=25, sortedby="date", reverse=True)
-        print "Search 1 took", now() - t
-        print "len=", r.scored_length()
-
-        t = now()
-        r = s.search(q, limit=25, sortedby="date")
-        print "Search 2 took", now() - t
-
-        t = now()
-        r = s.search(q, limit=25, sortedby="date")
-        print "Search 2 took", now() - t
-
-        from heapq import nlargest
-        t = now()
-        sf = s.stored_fields
-        gen = ((sf(n)["date"], n) for n in q.docs(s))
-        r = nlargest(25, gen)
-        print now() - t
+def test_bigsort():
+    times = 30000
+    dirname = "testindex"
+
+    df = fields.DATETIME(stored=True)
+    schema = fields.Schema(id=fields.ID(stored=True), date=df)
+
+    if os.path.exists(dirname):
+        shutil.rmtree(dirname)
+    os.mkdir(dirname)
+    ix = index.create_in(dirname, schema)
+
+    print "Writing..."
+    t = now()
+    w = ix.writer(limitmb=512)
+    for i in xrange(times):
+        dt = datetime.fromtimestamp(random.randint(15839593, 1294102139))
+        w.add_document(id=unicode(i), date=dt)
+    w.commit()
+    print "Writing took ", now() - t
+
+    ix = index.open_dir(dirname)
+    s = ix.searcher()
+    q = query.Wildcard("id", "1?2*")
+
+    t = now()
+    x = list(df.sortable_values(s.reader(), "date"))
+    print now() - t, len(x)
+
+    t = now()
+    for y in x:
+        p = list(s.postings("date", y).all_ids())
+    print now() - t
+
+
+
+    t = now()
+    r = s.search(q, limit=25, sortedby="date", reverse=True)
+    print "Search 1 took", now() - t
+    print "len=", r.scored_length()
+
+    t = now()
+    r = s.search(q, limit=25, sortedby="date")
+    print "Search 2 took", now() - t
+
+    t = now()
+    r = s.search(q, limit=25, sortedby="date")
+    print "Search 2 took", now() - t
+
+    from heapq import nlargest
+    t = now()
+    sf = s.stored_fields
+    gen = ((sf(n)["date"], n) for n in q.docs(s))
+    r = nlargest(25, gen)
+    print now() - t
 
 
 

stress/test_bigtable.py

@@ -1,16 +1,15 @@
-import unittest
+from __future__ import with_statement
 
-import os.path, shutil, tempfile
 from random import randint, shuffle
 
-from whoosh.filedb.filestore import FileStorage
-from whoosh.filedb.filetables import HashWriter, HashReader, dump_hash
+from nose.tools import assert_equal
 
-class Test(unittest.TestCase):
-    def test_bigtable(self):
-        dir = tempfile.mkdtemp(prefix="bigtable", suffix=".tmpix")
-        st = FileStorage(dir)
-
+from whoosh.filedb.filetables import HashWriter, HashReader
+from whoosh.support.testing import TempStorage
+
+
+def test_bigtable():
+    with TempStorage("bigtable") as st:
         def randstring(min, max):
             return "".join(chr(randint(1, 255))
                            for _ in xrange(randint(min, max)))
@@ -27,16 +26,12 @@ def randstring(min, max):
         keys = samp.keys()
         shuffle(keys)
         for key in keys:
-            self.assertEqual(samp[key], fhr[key])
+            assert_equal(samp[key], fhr[key])
 
         set1 = set(samp.iteritems())
         set2 = set(fhr.items())
-        self.assertEqual(set1, set2)
+        assert_equal(set1, set2)
 
         fhr.close()
-        shutil.rmtree(dir)
 
 
-if __name__ == "__main__":
-    #import sys;sys.argv = ['', 'Test.testName']
-    unittest.main()