|
- import nilmdb
- from nilmdb.utils.printf import *
-
- import nose
- from nose.tools import *
- from nose.tools import assert_raises
- import threading
- import time
- import nilmdb.server
-
- from testutil.helpers import *
-
- class Foo(object):
- val = 0
-
- def __init__(self, asdf = "asdf"):
- self.init_thread = threading.current_thread().name
-
- @classmethod
- def foo(self):
- pass
-
- def fail(self):
- raise Exception("you asked me to do this")
-
- def test(self, debug = False):
- self.tester(debug)
-
- def t(self):
- pass
-
- def reent(self, func):
- func()
-
- def tester(self, debug = False):
- # purposely not thread-safe
- self.test_thread = threading.current_thread().name
- 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 range(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()
-
- def verify_result(self):
- eq_(self.foo.val, 20)
- eq_(self.foo.init_thread, self.foo.test_thread)
-
- class ListLike(object):
- def __init__(self):
- self.thread = threading.current_thread().name
- self.foo = 0
-
- def __iter__(self):
- eq_(threading.current_thread().name, self.thread)
- self.foo = 0
- return self
-
- def __getitem__(self, key):
- eq_(threading.current_thread().name, self.thread)
- return key
-
- def __next__(self):
- eq_(threading.current_thread().name, self.thread)
- if self.foo < 5:
- self.foo += 1
- return self.foo
- else:
- raise StopIteration
-
- class TestUnserialized(Base):
- def setUp(self):
- self.foo = Foo()
-
- def verify_result(self):
- # This should have failed to increment properly
- ne_(self.foo.val, 20)
- # Init and tests ran in different threads
- ne_(self.foo.init_thread, self.foo.test_thread)
-
- class TestSerializer(Base):
- def setUp(self):
- self.foo = nilmdb.utils.serializer_proxy(Foo)("qwer")
-
- def test_multi(self):
- sp = nilmdb.utils.serializer_proxy
- sp(Foo("x")).t()
- sp(sp(Foo)("x")).t()
- sp(sp(Foo))("x").t()
- sp(sp(Foo("x"))).t()
- sp(sp(Foo)("x")).t()
- sp(sp(Foo))("x").t()
-
- def test_iter(self):
- sp = nilmdb.utils.serializer_proxy
- i = sp(ListLike)()
- eq_(list(i), [1,2,3,4,5])
- eq_(i[3], 3)
-
- def test_del(self):
- sp = nilmdb.utils.serializer_proxy
- foo = sp(Foo("x"))
-
- # trigger exception in __del__, which should be ignored
- foo._SerializerObjectProxy__call_queue = None
- del foo
-
- def test_rocket(self):
- # Serializer works on a C module?
- sp = nilmdb.utils.serializer_proxy
- rkt = sp(nilmdb.server.rocket.Rocket("int32_8", None))
- eq_(rkt.binary_size, 40)
|