Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_undecorated_callback(self):
"""Process Concurrent undecorated results are forwarded to callback."""
task = process.concurrent(target=undecorated, args=[1],
kwargs={'keyword_argument': 1},
callback=self.callback)
event.wait()
self.assertEqual(task.get(), 2)
@process.concurrent(callback=callback)
def long_decorated_callback():
time.sleep(1)
@process.concurrent(timeout=0.2)
def sigterm_decorated():
signal.signal(signal.SIGTERM, signal.SIG_IGN)
time.sleep(100)
def test_wrong_parameters(self):
"""Process Concurrent raises ValueError if wrong params."""
self.assertRaises(ValueError, process.concurrent, undecorated,
args=[1])
@process.concurrent(callback=callback)
def decorated_callback(argument, keyword_argument=0):
"""A docstring."""
return argument + keyword_argument
@process.concurrent
def decorated(argument, keyword_argument=0):
"""A docstring."""
return argument + keyword_argument
@process.concurrent
def clsmethod(cls):
return cls.a
def test_undecorated_results(self):
"""Process Concurrent undecorated results are produced."""
task = process.concurrent(target=undecorated_simple)
self.assertEqual(task.get(), 0)
@process.concurrent(5, name='foo')
def wrong():
return
except Exception as error: