Jim Paris
cc4e3bdb76
More work with nilmdb.stream_insert. Checks for overlap between parser output and database intervals, and actually inserts the data into pytables. Need to benchmark/figure out whether we can use table.append rather than the row.append() nonsense -- it could simplify things quite a bit. Improve layout class and add tests to get more coverage. Better error handling on invalid inputs (reports the reason for the error) Replace layout.fillrow with layout.filltable, but turns out that we can probably just remove it anyway. Add nilmdb.Timer for simple timing tests Move some duplicated test suite helper functions into a new file, test_helper.py Add class to test_interval.py to match the others git-svn-id: https://bucket.mit.edu/svn/nilm/nilmdb@10661 ddd99763-3ecb-0310-9145-efcb8ce7c51f
73 lines
1.7 KiB
Python
73 lines
1.7 KiB
Python
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
|