def test_suite_event_loop(self):
     '''Test event loop in test worker'''
     worker = pulsar.get_actor()
     loop = get_event_loop()
     self.assertTrue(loop.is_running())
     self.assertTrue(worker._loop.is_running())
     self.assertNotEqual(worker._loop, loop)

 def test_coroutine1(self):
     loop = get_event_loop()
     d1 = Future()
     loop.call_later(0.2, d1.set_result, 1)
     a = yield c_summation(d1)
     self.assertEqual(a, 3)
     self.assertEqual(d1.result(), 1)

 def test_chain(self):
     loop = get_event_loop()
     future = Future()
     next = chain_future(future, callback=lambda r: r+2)
     loop.call_later(0.2, future.set_result, 1)
     result = yield next
     self.assertEqual(result, 3)

def create_store(url, loop=None, **kw):
    '''Create a new client :class:`Store` for a valid ``url``.

    A valid ``url`` taks the following forms::

        pulsar://user:[email protected]:6410
        redis://user:[email protected]:6500/11?namespace=testdb.
        postgresql://user:[email protected]:6500/testdb
        couchdb://user:[email protected]:6500/testdb

    :param loop: optional event loop, if not provided it is obtained
        via the ``get_event_loop`` method. If not loop is installed a bright
        new event loop is created via the :func:`.new_event_loop`.
        In the latter case the event loop is employed only for synchronous type
        requests via the :meth:`~.EventLoop.run_until_complete` method.
    :param kw: additional key-valued parameters to pass to the :class:`Store`
        initialisation method.
    :return: a :class:`Store`.
    '''
    if isinstance(url, Store):
        return url
    scheme, address, params = parse_store_url(url)
    dotted_path = data_stores.get(scheme)
    if not dotted_path:
        raise ImproperlyConfigured('%s store not available' % scheme)
    loop = loop or get_event_loop()
    if not loop:
        loop = new_event_loop(logger=logging.getLogger(dotted_path))
    store_class = module_attribute(dotted_path)
    params.update(kw)
    return store_class(scheme, address, loop, **params)

def wait_fd(fd, read=True):
    '''Wait for an event on file descriptor ``fd``.

    :param fd: file descriptor
    :param read=True: wait for a read event if ``True``, otherwise a wait
        for write event.

    This function must be invoked from a coroutine with parent, therefore
    invoking it from the main greenlet will raise an exception.
    Check how this function is used in the :func:`.psycopg2_wait_callback`
    function.
    '''
    current = greenlet.getcurrent()
    parent = current.parent
    assert parent, '"wait_fd" must be called by greenlet with a parent'
    try:
        fileno = fd.fileno()
    except AttributeError:
        fileno = fd
    loop = get_event_loop()
    future = Future(loop=loop)
    # When the event on fd occurs switch back to the current greenlet
    if read:
        loop.add_reader(fileno, _done_wait_fd, fileno, future, read)
    else:
        loop.add_writer(fileno, _done_wait_fd, fileno, future, read)
    # switch back to parent greenlet
    parent.switch(future)
    return future.result()

    def test_periodic(self):
        test = self
        loop = get_event_loop()
        waiter = Future()

        class p:
            def __init__(self, loops):
                self.loops = loops
                self.c = 0

            def __call__(self):
                self.c += 1
                if self.c == self.loops:
                    try:
                        raise ValueError('test periodic')
                    except Exception:
                        waiter.set_result(self.c)
                        raise

        every = 2
        loops = 2
        track = p(loops)
        start = loop.time()
        periodic = call_repeatedly(loop, every, track)
        self.assertIsInstance(periodic, LoopingCall)
        done = yield waiter
        taken = loop.time() - start
        self.assertEqual(done, loops)
        self.assertTrue(taken > every*loops)
        self.assertTrue(taken < every*loops + 2)
        self.assertTrue(periodic.cancelled)
        self.assertFalse(has_callback(loop, periodic.handler))

 def test_call_at(self):
     loop = get_event_loop()
     d1 = Future()
     d2 = Future()
     c1 = loop.call_at(loop.time()+1, lambda: d1.set_result(loop.time()))
     c2 = loop.call_later(1, lambda: d2.set_result(loop.time()))
     t1, t2 = yield pulsar.multi_async((d1, d2))
     self.assertTrue(t1 <= t2)

 def test_getnameinfo(self):
     loop = get_event_loop()
     info = yield from loop.getaddrinfo('www.bbc.co.uk', 'http')
     info = yield from loop.getnameinfo(('212.58.244.66', 80))
     info = yield from loop.getaddrinfo('github.com', 'https')
     self.assertTrue(info)
     addr = info[0][4]
     info = yield from loop.getnameinfo(addr)
     self.assertTrue(info)

 def __init__(self, max_workers=None, loop=None, maxtasks=None):
     self._loop = loop or get_event_loop()
     self._max_workers = min(max_workers or _DEFAULT_WORKERS, _MAX_WORKERS)
     self._greenlets = set()
     self._available = set()
     self._maxtasks = maxtasks
     self._queue = deque()
     self._shutdown = False
     self._shutdown_lock = threading.Lock()

 def _(o):
     try:
         pool = o.app.green_pool
     except AttributeError:
         pool = None
     if pool:
         return pool.submit(test_fun, o)
     else:
         loop = get_event_loop()
         return loop.run_in_executor(None, test_fun, o)

 def __init__(self, max_workers=None, loop=None):
     self._loop = loop or get_event_loop()
     self._max_workers = min(max_workers or _DEFAULT_WORKERS, _MAX_WORKERS)
     self._greenlets = set()
     self._available = set()
     self._queue = deque()
     self._shutdown = False
     self._waiter = None
     self._logger = logging.getLogger('pulsar.greenpool')
     self._shutdown_lock = threading.Lock()

 def test_call_soon(self):
     ioloop = get_event_loop()
     tid = yield loop_thread_id(ioloop)
     d = Future()
     callback = lambda: d.set_result(current_thread().ident)
     cbk = ioloop.call_soon(callback)
     self.assertEqual(cbk._callback, callback)
     self.assertEqual(cbk._args, ())
     # we should be able to wait less than a second
     result = yield d
     self.assertEqual(result, tid)

 def test_pool(self):
     pool = greenio.GreenPool()
     self.assertTrue(pool._loop)
     self.assertEqual(pool._loop, get_event_loop())
     self.assertFalse(pool._greenlets)
     future = pool.submit(lambda: 'Hi!')
     self.assertIsInstance(future, Future)
     result = yield from future
     self.assertEqual(result, 'Hi!')
     self.assertEqual(len(pool._greenlets), 1)
     self.assertEqual(len(pool._available), 1)

    def run_until_complete(self, options, **params):
        '''Execute the :meth:`run` method using pulsar asynchronous engine.

        Most commands are run using this method.
        '''
        loop = get_event_loop()
        run = run_in_greenlet(self.run)
        result = run(options, **params)
        if not loop.is_running():
            return loop.run_until_complete(result)
        else:
            return result

 def run_async(self, argv, **params):
     '''Run a command using pulsar asynchronous engine.'''
     loop = get_event_loop()
     run_until_complete = False
     if loop is None:
         run_until_complete = True
         loop = new_event_loop()
     future = async(self.run(argv, **params), loop)
     if run_until_complete:
         return loop.run_until_complete(future)
     else:
         return future

    def run_until_complete(self, options, **params):
        '''Execute the :meth:`run` method using pulsar asynchronous engine.

        Most commands are run using this method.
        '''
        pool = self.app.green_pool
        loop = get_event_loop()
        run = partial(self.run, options, **params)
        if pool:
            result = pool.submit(run)
        else:
            result = loop.run_in_executor(None, run)
        return result if loop.is_running() else loop.run_until_complete(result)

 def test_connection_pool(self):
     """Test the connection pool. A very important test!"""
     client = Echo(self.server_cfg.addresses[0], pool_size=2)
     self.assertNotEqual(client._loop, get_event_loop())
     #
     self.assertEqual(client.pool.pool_size, 2)
     self.assertEqual(client.pool.in_use, 0)
     self.assertEqual(client.pool.available, 0)
     self.assertEqual(client.sessions, 0)
     self.assertEqual(client._requests_processed, 0)
     #
     response = yield client(b"test connection")
     self.assertEqual(response, b"test connection")
     self.assertEqual(client.pool.in_use, 0)
     self.assertEqual(client.pool.available, 1)
     self.assertEqual(client.sessions, 1)
     self.assertEqual(client._requests_processed, 1)
     #
     response = yield client(b"test connection 2")
     self.assertEqual(response, b"test connection 2")
     self.assertEqual(client.pool.in_use, 0)
     self.assertEqual(client.pool.available, 1)
     self.assertEqual(client.sessions, 1)
     self.assertEqual(client._requests_processed, 2)
     #
     result = yield multi_async((client(b"ciao"), client(b"pippo"), client(b"foo")))
     self.assertEqual(len(result), 3)
     self.assertTrue(b"ciao" in result)
     self.assertTrue(b"pippo" in result)
     self.assertTrue(b"foo" in result)
     self.assertEqual(client.pool.in_use, 0)
     self.assertEqual(client.pool.available, 2)
     self.assertEqual(client.sessions, 2)
     self.assertEqual(client._requests_processed, 5)
     #
     # drop a connection
     yield run_in_loop(client._loop, self._drop_conection, client)
     #
     result = yield multi_async((client(b"ciao"), client(b"pippo"), client(b"foo")))
     self.assertEqual(len(result), 3)
     self.assertEqual(client.pool.in_use, 0)
     self.assertEqual(client.pool.available, 2)
     self.assertEqual(client.sessions, 3)
     self.assertEqual(client._requests_processed, 8)
     #
     yield run_in_loop(client._loop, client.pool.close)
     #
     self.assertEqual(client.pool.in_use, 0)
     self.assertEqual(client.pool.available, 0)
     self.assertEqual(client.sessions, 3)
     self.assertEqual(client._requests_processed, 8)

def create_store(url, loop=None, **kw):
    '''Create a new :class:`Store` for a valid ``url``.

    :param url: a valid ``url`` takes the following forms:

        :ref:`Pulsar datastore <store_pulsar>`::

            pulsar://user:[email protected]:6410

        :ref:`Redis <store_redis>`::

            redis://user:[email protected]:6500/11?namespace=testdb

        :ref:`CouchDb <store_couchdb>`::

            couchdb://user:[email protected]:6500/testdb
            https+couchdb://user:[email protected]:6500/testdb

    :param loop: optional event loop, obtained by
        :func:`~asyncio.get_event_loop` if not provided.
        To create a synchronous client pass a new event loop created via
        the :func:`~asyncio.new_event_loop`.
        In the latter case the event loop is employed only for synchronous
        type requests via the :meth:`~asyncio.BaseEventLoop.run_until_complete`
        method.
    :param kw: additional key-valued parameters to pass to the :class:`.Store`
        initialisation method. It can contains parameters such as
        ``database``, ``user`` and ``password`` to override the
        ``url`` values. Additional parameters are processed by the
        :meth:`.Store._init` method.
    :return: a :class:`Store`.
    '''
    if isinstance(url, Store):
        return url
    scheme, address, params = parse_store_url(url)
    dotted_path = data_stores.get(scheme)
    if not dotted_path:
        raise ImproperlyConfigured('%s store not available' % scheme)
    loop = loop or get_event_loop()
    if not loop:
        loop = new_event_loop(logger=logging.getLogger(dotted_path))
    store_class = module_attribute(dotted_path)
    if not store_class.registered:
        store_class.registered = True
        store_class.register()
    params.update(kw)
    return store_class(scheme, address, loop, **params)

def search_engine(url, loop=None, **kw):
    if isinstance(url, SearchEngine):
        return url
    loop = loop or get_event_loop()
    if not loop:
        raise ImproperlyConfigured('no event loop')
    if isinstance(url, dict):
        extra = url.copy()
        url = extra.pop('url', None)
        extra.update(kw)
        kw = extra
    scheme, address, params = parse_store_url(url)
    dotted_path = search_engines.get(scheme)
    if not dotted_path:
        raise ImproperlyConfigured('%s search engine not available' % scheme)
    engine_class = module_attribute(dotted_path)
    params.update(kw)
    return engine_class(scheme, address, loop, **params)

    def test_run_in_thread_loop(self):
        event_loop = get_event_loop()

        def simple(a, b):
            return a + b

        d = run_in_loop(event_loop, simple, 1, 2)
        self.assertIsInstance(d, Future)
        result = yield d
        self.assertEqual(result, 3)
        d = run_in_loop(event_loop, simple, 1, 'a')
        self.assertIsInstance(d, Future)
        try:
            result = yield d
        except TypeError:
            pass
        else:
            assert False, "TypeError not raised"