How to use the pebble.process.concurrent function in Pebble

To help you get started, we’ve selected a few Pebble examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
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)
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
@process.concurrent(callback=callback)
def long_decorated_callback():
    time.sleep(1)
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
@process.concurrent(timeout=0.2)
def sigterm_decorated():
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
    time.sleep(100)
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
def test_wrong_parameters(self):
        """Process Concurrent raises ValueError if wrong params."""
        self.assertRaises(ValueError, process.concurrent, undecorated,
                          args=[1])
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
@process.concurrent(callback=callback)
def decorated_callback(argument, keyword_argument=0):
    """A docstring."""
    return argument + keyword_argument
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
@process.concurrent
def decorated(argument, keyword_argument=0):
    """A docstring."""
    return argument + keyword_argument
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
    @process.concurrent
    def clsmethod(cls):
        return cls.a
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
def test_undecorated_results(self):
        """Process Concurrent undecorated results are produced."""
        task = process.concurrent(target=undecorated_simple)
        self.assertEqual(task.get(), 0)
github noxdafox / pebble / tests / test_process_concurrent.py View on Github external
            @process.concurrent(5, name='foo')
            def wrong():
                return
        except Exception as error: