def test_add_and_subscribe(self):
        """
        Create an application session that subscribes to some
        topic and add it to a router to run embedded.
        """
        d = txaio.create_future()

        class TestSession(ApplicationSession):
            def onJoin(self, details):
                # noinspection PyUnusedLocal
                def on_event(*arg, **kwargs):
                    pass

                d2 = self.subscribe(on_event, u'com.example.topic1')

                def ok(_):
                    txaio.resolve(d, None)

                def error(err):
                    txaio.reject(d, err)

                txaio.add_callbacks(d2, ok, error)

        session = TestSession(types.ComponentConfig(u'realm1'))

        self.session_factory.add(session)

        return d

def test_errback_reject_no_args():
    """
    txaio.reject() with no args
    """

    f = txaio.create_future()
    exception = RuntimeError("it failed")
    errors = []

    def err(f):
        errors.append(f)
    txaio.add_callbacks(f, None, err)
    try:
        raise exception
    except:
        txaio.reject(f)

    run_once()

    assert len(errors) == 1
    assert isinstance(errors[0], txaio.IFailedFuture)
    assert exception == errors[0].value
    tb = txaio.failure_format_traceback(errors[0])

    assert 'RuntimeError' in tb
    assert 'it failed' in tb
    assert txaio.failure_message(errors[0]) == 'RuntimeError: it failed'
    assert 'it failed' in str(errors[0])

def test_errback(framework):
    f = txaio.create_future()
    exception = RuntimeError("it failed")
    errors = []

    def err(f):
        errors.append(f)
    txaio.add_callbacks(f, None, err)
    try:
        raise exception
    except:
        fail = txaio.create_failure()
    txaio.reject(f, fail)

    run_once()

    assert len(errors) == 1
    assert isinstance(errors[0], txaio.IFailedFuture)
    assert exception == errors[0].value
    assert txaio.failure_traceback(errors[0]) is not None

    tb = txaio.failure_format_traceback(errors[0])

    assert 'RuntimeError' in tb
    assert 'it failed' in tb
    assert txaio.failure_message(errors[0]) == 'RuntimeError: it failed'
    assert 'it failed' in str(errors[0])

    def test_add_and_subscribe(self):
        """
        Create an application session that subscribes to some
        topic and add it to a router to run embedded.
        """
        d = txaio.create_future()

        class TestSession(ApplicationSession):

            def onJoin(self, details):
                d2 = self.subscribe(lambda: None, u'com.example.topic1')

                def ok(_):
                    txaio.resolve(d, None)

                def error(err):
                    txaio.reject(d, err)

                txaio.add_callbacks(d2, ok, error)

        session = TestSession(types.ComponentConfig(u'realm1'))

        self.session_factory.add(session, authrole=u'test_role')

        return d

    def _unsubscribe(self, subscription):
        """
        Called from :meth:`autobahn.wamp.protocol.Subscription.unsubscribe`
        """
        assert(isinstance(subscription, Subscription))
        assert subscription.active
        assert(subscription.id in self._subscriptions)
        assert(subscription in self._subscriptions[subscription.id])

        if not self._transport:
            raise exception.TransportLost()

        # remove handler subscription and mark as inactive
        self._subscriptions[subscription.id].remove(subscription)
        subscription.active = False

        # number of handler subscriptions left ..
        scount = len(self._subscriptions[subscription.id])

        if scount == 0:
            # if the last handler was removed, unsubscribe from broker ..
            request_id = util.id()

            on_reply = txaio.create_future()
            self._unsubscribe_reqs[request_id] = UnsubscribeRequest(request_id, on_reply, subscription.id)

            msg = message.Unsubscribe(request_id, subscription.id)

            self._transport.send(msg)
            return on_reply
        else:
            # there are still handlers active on the subscription!
            return txaio.create_future_success(scount)

def test_is_future_generic():
    '''
    Returning an immediate value from as_future
    '''
    f = txaio.create_future('result')

    assert txaio.is_future(f)

def test_create_error():
    f = txaio.create_future(error=RuntimeError("test"))
    if txaio.using_twisted:
        assert f.called
    else:
        assert f.done()
    # cancel the error; we expected it
    txaio.add_callbacks(f, None, lambda _: None)

        def on_connect(req):
            f = txaio.create_future()

            def cb(x):
                f = foo(42)
                f.add_callbacks(f, lambda v: values.append(v), None)
                return f
            txaio.add_callbacks(f, cb, None)
            return f

def test_errback_illegal_args(framework):
    '''
    non-Exception/Failures should be rejected
    '''
    f = txaio.create_future()
    try:
        txaio.reject(f, object())
        assert "should have raised exception."
    except RuntimeError:
        pass

 def notify_connect_error(fail):
     chain_f = txaio.create_future()
     # hmm, if connectfailure took a _Transport instead of
     # (or in addition to?) self it could .failed() the
     # transport and we could do away with the is_fatal
     # listener?
     handler_f = self.fire('connectfailure', self, fail.value)
     txaio.add_callbacks(
         handler_f,
         lambda _: txaio.reject(chain_f, fail),
         lambda _: txaio.reject(chain_f, fail)
     )
     return chain_f

def test_callback(framework):
    f = txaio.create_future()
    results = []

    def cb(f):
        results.append(f)
    txaio.add_callbacks(f, cb, None)
    txaio.resolve(f, "it worked")

    run_once()

    assert len(results) == 1
    assert results[0] == "it worked"

        def _register(obj, fn, procedure, options):
            request_id = util.id()
            on_reply = txaio.create_future()
            endpoint_obj = Endpoint(fn, obj, options.details_arg if options else None)
            self._register_reqs[request_id] = RegisterRequest(request_id, on_reply, procedure, endpoint_obj)

            if options:
                msg = message.Register(request_id, procedure, **options.message_attr())
            else:
                msg = message.Register(request_id, procedure)

            self._transport.send(msg)
            return on_reply

        def _subscribe(obj, fn, topic, options):
            request_id = util.id()
            on_reply = txaio.create_future()
            handler_obj = Handler(fn, obj, options.details_arg if options else None)
            self._subscribe_reqs[request_id] = SubscribeRequest(request_id, on_reply, handler_obj)

            if options:
                msg = message.Subscribe(request_id, topic, **options.message_attr())
            else:
                msg = message.Subscribe(request_id, topic)

            self._transport.send(msg)
            return on_reply

        def sign_challenge(self, session, challenge):
            """
            Sign WAMP-cryptosign challenge.

            :param challenge: The WAMP-cryptosign challenge object for which a signature should be computed.
            :type challenge: instance of autobahn.wamp.types.Challenge

            :returns: A Deferred/Future that resolves to the computed signature.
            :rtype: str
            """
            if not isinstance(challenge, Challenge):
                raise Exception("challenge must be instance of autobahn.wamp.types.Challenge, not {}".format(type(challenge)))

            if u'challenge' not in challenge.extra:
                raise Exception("missing challenge value in challenge.extra")

            # the challenge sent by the router (a 32 bytes random value)
            challenge_hex = challenge.extra[u'challenge']

            # the challenge for WAMP-cryptosign is a 32 bytes random value in Hex encoding (that is, a unicode string)
            challenge_raw = binascii.a2b_hex(challenge_hex)

            # if the transport has a channel ID, the message to be signed by the client actually
            # is the XOR of the challenge and the channel ID
            channel_id_raw = session._transport.get_channel_id()
            if channel_id_raw:
                data = util.xor(challenge_raw, channel_id_raw)
            else:
                data = challenge_raw

            # a raw byte string is signed, and the signature is also a raw byte string
            d1 = self.sign(data)

            # asyncio lacks callback chaining (and we cannot use co-routines, since we want
            # to support older Pythons), hence we need d2
            d2 = txaio.create_future()

            def process(signature_raw):
                # convert the raw signature into a hex encode value (unicode string)
                signature_hex = binascii.b2a_hex(signature_raw).decode('ascii')

                # we return the concatenation of the signature and the message signed (96 bytes)
                data_hex = binascii.b2a_hex(data).decode('ascii')

                sig = signature_hex + data_hex
                txaio.resolve(d2, sig)

            txaio.add_callbacks(d1, process, None)

            return d2

def test_explicit_reactor_future():
    """
    If we set an event-loop, Futures + Tasks should use it.
    """
    pytest.importorskip('asyncio')
    if txaio.using_twisted:
        pytest.skip()

    with patch.object(txaio.config, 'loop') as fake_loop:
        f = txaio.create_future('result')
        f.add_done_callback(lambda _: None)

        assert len(fake_loop.method_calls) == 2
        c = fake_loop.method_calls[1]
        assert c[0] == 'call_soon'

    def test_add(self):
        """
        Create an application session and add it to a router to
        run embedded.
        """
        d = txaio.create_future()

        class TestSession(ApplicationSession):
            def onJoin(self, details):
                txaio.resolve(d, None)

        session = TestSession(types.ComponentConfig(u"realm1"))

        self.session_factory.add(session)

        return d

    def call(self, procedure, *args, **kwargs):
        """
        Implements :func:`autobahn.wamp.interfaces.ICaller.call`
        """
        if six.PY2 and type(procedure) == str:
            procedure = six.u(procedure)
        assert(isinstance(procedure, six.text_type))

        if not self._transport:
            raise exception.TransportLost()

        request_id = util.id()

        if 'options' in kwargs and isinstance(kwargs['options'], types.CallOptions):
            options = kwargs.pop('options')
            msg = message.Call(request_id, procedure, args=args, kwargs=kwargs, **options.message_attr())
        else:
            options = None
            msg = message.Call(request_id, procedure, args=args, kwargs=kwargs)

        # FIXME
        # def canceller(_d):
        #   cancel_msg = message.Cancel(request)
        #   self._transport.send(cancel_msg)
        # d = Deferred(canceller)

        on_reply = txaio.create_future()
        self._call_reqs[request_id] = CallRequest(request_id, on_reply, options)

        try:
            # Notes:
            #
            # * this might raise autobahn.wamp.exception.SerializationError
            #   when the user payload cannot be serialized
            # * we have to setup a PublishRequest() in _publish_reqs _before_
            #   calling transpor.send(), because a mock- or side-by-side transport
            #   will immediately lead on an incoming WAMP message in onMessage()
            #
            self._transport.send(msg)
        except Exception as e:
            if request_id in self._call_reqs:
                del self._call_reqs[request_id]
            raise e

        return on_reply

    def publish(self, topic, *args, **kwargs):
        """
        Implements :func:`autobahn.wamp.interfaces.IPublisher.publish`
        """
        if six.PY2 and type(topic) == str:
            topic = six.u(topic)
        assert(type(topic) == six.text_type)

        if not self._transport:
            raise exception.TransportLost()

        request_id = util.id()

        if 'options' in kwargs and isinstance(kwargs['options'], types.PublishOptions):
            options = kwargs.pop('options')
            msg = message.Publish(request_id, topic, args=args, kwargs=kwargs, **options.message_attr())
        else:
            options = None
            msg = message.Publish(request_id, topic, args=args, kwargs=kwargs)

        if options and options.acknowledge:
            # only acknowledged publications expect a reply ..
            on_reply = txaio.create_future()
            self._publish_reqs[request_id] = PublishRequest(request_id, on_reply)
        else:
            on_reply = None

        try:
            # Notes:
            #
            # * this might raise autobahn.wamp.exception.SerializationError
            #   when the user payload cannot be serialized
            # * we have to setup a PublishRequest() in _publish_reqs _before_
            #   calling transpor.send(), because a mock- or side-by-side transport
            #   will immediately lead on an incoming WAMP message in onMessage()
            #
            self._transport.send(msg)
        except Exception as e:
            if request_id in self._publish_reqs:
                del self._publish_reqs[request_id]
            raise e

        return on_reply

    def _unregister(self, registration):
        """
        Called from :meth:`autobahn.wamp.protocol.Registration.unregister`
        """
        assert(isinstance(registration, Registration))
        assert registration.active
        assert(registration.id in self._registrations)

        if not self._transport:
            raise exception.TransportLost()

        request_id = util.id()

        on_reply = txaio.create_future()
        self._unregister_reqs[request_id] = UnregisterRequest(request_id, on_reply, registration.id)

        msg = message.Unregister(request_id, registration.id)

        self._transport.send(msg)
        return on_reply

def test_cancel(framework):
    cancels = []

    def it_died(f):
        cancels.append(f)

    f = txaio.create_future(canceller=it_died)
    # both Future and Deferred have .cancel() methods .. but seemed
    # more "symmetric"/expected to make a method? But could just stick
    # with "f.cancel()" here ...
    txaio.cancel(f)

    # at least for Twisted, we have to "handle" the "CancelledError"
    # -- in practice, dropping a future on the floor with no
    # error-handler is A Bad Thing anyway
    txaio.add_callbacks(f, None, lambda _: None)

    run_once()
    run_once()

    assert cancels == [f]