|
- import nilmdb
- from nilmdb.utils.printf import *
-
- import nose
- from nose.tools import *
- from nose.tools import assert_raises
- import threading
- import time
-
- from testutil.helpers import *
-
- #raise nose.exc.SkipTest("Skip these")
-
- class Foo(object):
- val = 0
-
- def fail(self):
- raise Exception("you asked me to do this")
-
- def test(self, debug = False):
- # purposely not thread-safe
- oldval = self.val
- newval = oldval + 1
- time.sleep(0.05)
- self.val = newval
- if debug:
- printf("[%s] value changed: %d -> %d\n",
- threading.current_thread().name, oldval, newval)
-
- class Base(object):
-
- def test_wrapping(self):
- self.foo.test()
- with assert_raises(Exception):
- self.foo.fail()
-
- def test_threaded(self):
- def func(foo):
- foo.test()
- threads = []
- for i in xrange(20):
- threads.append(threading.Thread(target = func, args = (self.foo,)))
- for t in threads:
- t.start()
- for t in threads:
- t.join()
- self.verify_result()
-
- class TestUnserialized(Base):
- def setUp(self):
- self.foo = Foo()
-
- def verify_result(self):
- # This should have failed to increment properly
- assert(self.foo.val != 20)
-
- class TestSerialized(Base):
- def setUp(self):
- self.realfoo = Foo()
- self.foo = nilmdb.utils.Serializer(self.realfoo)
-
- def tearDown(self):
- del self.foo
-
- def verify_result(self):
- # This should have worked
- eq_(self.realfoo.val, 20)
-
- def test_attribute(self):
- # Can't wrap attributes yet
- with assert_raises(TypeError):
- self.foo.val
|