nilmdb/tests/test_serializer.py

import nilmdb
from nilmdb.printf import *

import nose
from nose.tools import *
from nose.tools import assert_raises
import threading
import time

from test_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.serializer.WrapObject(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