56 lines
1.4 KiB
Python
56 lines
1.4 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 *
|
|
|
|
def func_with_callback(a, b, callback):
|
|
callback(a)
|
|
callback(b)
|
|
callback(a+b)
|
|
|
|
class TestIteratorizer(object):
|
|
def test(self):
|
|
|
|
# First try it with a normal callback
|
|
self.result = ""
|
|
def cb(x):
|
|
self.result += str(x)
|
|
func_with_callback(1, 2, cb)
|
|
eq_(self.result, "123")
|
|
|
|
# Now make it an iterator
|
|
it = nilmdb.utils.Iteratorizer(
|
|
lambda x:
|
|
func_with_callback(1, 2, x))
|
|
result = ""
|
|
for i in it:
|
|
result += str(i)
|
|
eq_(result, "123")
|
|
|
|
# Make sure things work when an exception occurs
|
|
it = nilmdb.utils.Iteratorizer(
|
|
lambda x:
|
|
func_with_callback(1, "a", x))
|
|
result = ""
|
|
with assert_raises(TypeError) as e:
|
|
for i in it:
|
|
result += str(i)
|
|
eq_(result, "1a")
|
|
|
|
# Now try to trigger the case where we stop iterating
|
|
# mid-generator, and expect the iteratorizer to clean up after
|
|
# itself. This doesn't have a particular result in the test,
|
|
# but gains coverage.
|
|
def foo():
|
|
it = nilmdb.utils.Iteratorizer(
|
|
lambda x:
|
|
func_with_callback(1, 2, x))
|
|
it.next()
|
|
foo()
|