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