73 lines
1.7 KiB
Python
73 lines
1.7 KiB
Python
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
|