def test__on_underlying_close_error(self, mocklog):
        tp = AMQPTransport(Mock())

        tp._on_underlying_close(404, sentinel.text)

        self.assertEquals(mocklog.error.call_count, 1)
        self.assertIn(sentinel.text, mocklog.error.call_args[0])
        self.assertEquals(mocklog.debug.call_count, 0)

    def test_close_while_locked(self):
        tp = AMQPTransport(Mock())
        tp.lock = True

        tp.close()

        self.assertEquals(tp._client.close.call_count, 0)
        self.assertEquals(tp._client.callbacks.remove.call_count, 0)

    def test__sync_call_with_normal_and_kwarg_rets(self):
        def async_func(*args, **kwargs):
            cbparam = kwargs.get('callback')
            cbparam(sentinel.arg, sup=sentinel.val, sup2=sentinel.val2)

        tp = AMQPTransport(Mock())
        rv = tp._sync_call(async_func, 'callback')
        self.assertEquals(rv, (sentinel.arg, {'sup':sentinel.val, 'sup2':sentinel.val2}))

    def test__sync_call_with_mult_rets(self):
        def async_func(*args, **kwargs):
            cbparam = kwargs.get('callback')
            cbparam(sentinel.val, sentinel.val2)

        tp = AMQPTransport(Mock())
        rv = tp._sync_call(async_func, 'callback')
        self.assertEquals(rv, (sentinel.val, sentinel.val2))

    def test__sync_call_no_ret_value(self):

        def async_func(*args, **kwargs):
            cbparam = kwargs.get('callback')
            cbparam()

        tp = AMQPTransport(Mock())
        rv = tp._sync_call(async_func, 'callback')
        self.assertIsNone(rv)

    def test__on_underlying_close(self, mocklog):
        tp = AMQPTransport(Mock())
        cb = Mock()
        tp.add_on_close_callback(cb)

        tp._on_underlying_close(200, sentinel.text)

        cb.assert_called_once_with(tp, 200, sentinel.text)
        self.assertEquals(mocklog.debug.call_count, 1)
        self.assertIn(sentinel.text, mocklog.debug.call_args[0])

    def test_close(self):
        client = Mock()
        tp = AMQPTransport(client)
        tp.close()

        client.close.assert_called_once_with()

        self.assertEquals(client.callbacks.remove.call_count, 4)
        self.assertEquals(client.callbacks.remove.call_args_list, [call(client.channel_number, 'Basic.GetEmpty'),
                                                                   call(client.channel_number, 'Channel.Close'),
                                                                   call(client.channel_number, '_on_basic_deliver'),
                                                                   call(client.channel_number, '_on_basic_get')])

    def test__on_underlying_close(self):
        client = Mock()
        tp = AMQPTransport(client)
        cb = Mock()
        tp.add_on_close_callback(cb)

        tp._on_underlying_close(200, sentinel.text)

        cb.assert_called_once_with(tp, 200, sentinel.text)

        self.assertEquals(client.callbacks.remove.call_count, 4)
        self.assertEquals(client.callbacks.remove.call_args_list, [call(client.channel_number, 'Basic.GetEmpty'),
                                                                   call(client.channel_number, 'Channel.Close'),
                                                                   call(client.channel_number, '_on_basic_deliver'),
                                                                   call(client.channel_number, '_on_basic_get')])

    def test_delete_xn(self):
        raise unittest.SkipTest("broken 2 mar, skipping for now")
        # same as the other deletes except with queues instead

        xn = self.container.ex_manager.create_xn_service('test_service')

        # prove queue is declared already (can't declare the same named one with diff params)
        ch = self.container.node.channel(RecvChannel)
        ch._queue_auto_delete = not xn._xn_auto_delete

        # must set recv_name
        ch._recv_name = xn
        self.assertRaises(TransportError, ch._declare_queue, xn.queue)

        # now let's delete via ex manager
        self.container.ex_manager.delete_xn(xn)

        # grab another channel and declare (should work fine this time)
        ch = self.container.node.channel(RecvChannel)
        ch._exchange_auto_delete = not xn._xs._xs_auto_delete

        # must set recv_name
        ch._recv_name = xn
        ch._declare_queue(xn.queue)

        # cool, now cleanup (we don't expose this via Channel)
        at = AMQPTransport.get_instance()
        at.delete_queue_impl(ch._amq_chan, xn.queue)

    def accept(self, n=1, timeout=None):
        """
        Accepts new connections for listening endpoints.

        Can accept more than one message at at time before it returns a new channel back to the
        caller. Optionally can specify a timeout - if n messages aren't received in that time,
        will raise an Empty exception.

        Sets the channel in the ACCEPTED state - caller is responsible for acking all messages
        received on the returned channel in order to put this channel back in the CONSUMING
        state.
        """

        assert self._fsm.current_state in [self.S_ACTIVE, self.S_CLOSED], "Channel must be in active/closed state to accept, currently %s (forget to ack messages?)" % str(self._fsm.current_state)

        was_consuming = self._consuming

        if not self._should_discard and not was_consuming:
            # tune QOS to get exactly n messages
            if not (self._queue_auto_delete and self._transport is AMQPTransport.get_instance()):
                self._transport.qos_impl(self._amq_chan, prefetch_count=n)

            # start consuming
            self.start_consume()

        with self._recv_queue.await_n(n=n) as ar:
            log.debug("accept: waiting for %s msgs, timeout=%s", n, timeout)
            ar.get(timeout=timeout)

        if not was_consuming:
            # turn consuming back off if we already were off
            if not (self._queue_auto_delete and self._transport is AMQPTransport.get_instance()):
                self.stop_consume()
            else:
                log.debug("accept should turn consume off, but queue is auto_delete and this would destroy the queue")

        ms = [self.recv() for x in xrange(n)]

        ch = self._create_accepted_channel(self._amq_chan, ms)
        map(ch._recv_queue.put, ms)

        # transition to ACCEPT
        self._fsm.process(self.I_ENTER_ACCEPT)

        # return the channel
        return ch

    def test_stop_consume_raises_warning_with_auto_delete(self):
        transport = AMQPTransport(Mock())
        transport.stop_consume_impl = Mock()
        self.ch.on_channel_open(transport)
        #transport.channel_number = sentinel.channel_number

        self.ch._consumer_tag = sentinel.consumer_tag
        self.ch._recv_name = NameTrio(sentinel.ex, sentinel.queue, sentinel.binding)
        self.ch._fsm.current_state = self.ch.S_ACTIVE
        self.ch._consuming = True

        #self.ch._ensure_transport = MagicMock()
        self.ch._queue_auto_delete = True

        self.ch.stop_consume()

        self.assertTrue(self.ch._transport.stop_consume_impl.called)
        self.assertIn(self.ch._consumer_tag, self.ch._transport.stop_consume_impl.call_args[0])

    def start(self):
        log.debug("ExchangeManager.start")

        total_count = 0

        def handle_failure(name, node):
            log.warn("Node %s could not be started", name)
            node.ready.set()        # let it fall out below

        # Establish connection(s) to broker
        for name, cfgkey in CFG.container.messaging.server.iteritems():
            if not cfgkey:
                continue

            if cfgkey not in CFG.server:
                raise ExchangeManagerError("Config key %s (name: %s) (from CFG.container.messaging.server) not in CFG.server" % (cfgkey, name))

            total_count += 1
            log.debug("Starting connection: %s", name)

            # start it with a zero timeout so it comes right back to us
            try:
                node, ioloop = messaging.make_node(CFG.server[cfgkey], name, 0)

                # install a finished handler directly on the ioloop just for this startup period
                fail_handle = lambda _: handle_failure(name, node)
                ioloop.link(fail_handle)

                # wait for the node ready event, with a large timeout just in case
                node_ready = node.ready.wait(timeout=15)

                # remove the finished handler, we don't care about it here
                ioloop.unlink(fail_handle)

                # only add to our list if we started successfully
                if not node.running:
                    ioloop.kill()      # make sure ioloop dead
                else:
                    self._nodes[name]   = node
                    self._ioloops[name] = ioloop

            except socket.error as e:
                log.warn("Could not start connection %s due to socket error, continuing", name)

        fail_count = total_count - len(self._nodes)
        if fail_count > 0 or total_count == 0:
            if fail_count == total_count:
                raise ExchangeManagerError("No node connection was able to start (%d nodes attempted, %d nodes failed)" % (total_count, fail_count))

            log.warn("Some nodes could not be started, ignoring for now")   # @TODO change when ready

        self._transport = AMQPTransport.get_instance()

        # load interceptors into each
        map(lambda x: x.setup_interceptors(CFG.interceptor), self._nodes.itervalues())

        log.debug("Started %d connections (%s)", len(self._nodes), ",".join(self._nodes.iterkeys()))

    def _new_transport(self, ch_number=None):
        """
        Creates a new AMQPTransport with an underlying Pika channel.
        """
        amq_chan = blocking_cb(self.client.channel, 'on_open_callback', channel_number=ch_number)
        if amq_chan is None:
            log.error("AMQCHAN IS NONE THIS SHOULD NEVER HAPPEN, chan number requested: %s", ch_number)
            from pyon.container.cc import Container
            if Container.instance is not None:
                Container.instance.fail_fast("AMQCHAN IS NONE, messaging has failed", True)
            raise StandardError("AMQCHAN IS NONE THIS SHOULD NEVER HAPPEN, chan number requested: %s" % ch_number)

        transport = AMQPTransport(amq_chan)

        # by default, everything should have a prefetch count of 1 (configurable)
        # this can be overridden by the channel get_n related methods
        transport.qos_impl(prefetch_count=CFG.get_safe('container.messaging.endpoint.prefetch_count', 1))

        return transport

    def __init__(self, transport=None, close_callback=None):
        """
        Initializes a BaseChannel instance.

        @param  transport       Underlying transport used for broker communication. Can be None, if so, will
                                use the AMQPTransport stateless singleton.
        @type   transport       BaseTransport
        @param  close_callback  The method to invoke when close() is called on this BaseChannel. May be left as None,
                                in which case close_impl() will be called. This expects to be a callable taking one
                                param, this channel instance.
        """
        self.set_close_callback(close_callback)
        self._transport = transport or AMQPTransport.get_instance()

    def start(self):
        log.debug("ExchangeManager starting ...")

        # Establish connection to broker
        # @TODO: raise error if sux
        node, ioloop = messaging.make_node()

        self._transport = AMQPTransport.get_instance()
        self._client    = self._get_channel(node)

        # Declare root exchange
        #self.default_xs.ensure_exists(self._get_channel())
        return node, ioloop

    def _on_stop_consume(self):
        """
        Stops consuming messages.

        If the queue has auto_delete, this will delete it.
        """
        #log.debug("RecvChannel._on_stop_consume")

        if self._queue_auto_delete and self._transport is AMQPTransport.get_instance():
            log.debug("Autodelete is on, this will destroy this queue: %s", self._recv_name.queue)

        self._ensure_amq_chan()

        self._sync_call(self._amq_chan.basic_cancel, 'callback', self._consumer_tag)

    def _on_start_consume(self):
        """
        Starts consuming messages.

        setup_listener must have been called first.
        """
        #log.debug("RecvChannel._on_start_consume")

        if self._consumer_tag and (self._queue_auto_delete and self._transport is AMQPTransport.get_instance()):
            log.warn("Attempting to start consuming on a queue that may have been auto-deleted")

        self._ensure_amq_chan()

        self._consumer_tag = self._amq_chan.basic_consume(self._on_deliver,
                                                          queue=self._recv_name.queue,
                                                          no_ack=self._consumer_no_ack,
                                                          exclusive=self._consumer_exclusive)

    def __init__(self, transport=None, close_callback=None):
        """
        Initializes a BaseChannel instance.

        @param  transport       Underlying transport used for broker communication. Can be None, if so, will
                                use the AMQPTransport stateless singleton.
        @type   transport       BaseTransport
        @param  close_callback  The method to invoke when close() is called on this BaseChannel. May be left as None,
                                in which case close_impl() will be called. This expects to be a callable taking one
                                param, this channel instance.
        """
        self.set_close_callback(close_callback)
        self._transport = transport or AMQPTransport.get_instance()

        # setup FSM for BaseChannel / SendChannel tree
        self._fsm = FSM(self.S_INIT)
        self._fsm.add_transition(self.I_ATTACH, self.S_INIT, None, self.S_ACTIVE)
        self._fsm.add_transition(self.I_CLOSE, self.S_ACTIVE, self._on_close, self.S_CLOSED)
        self._fsm.add_transition(self.I_CLOSE, self.S_CLOSED, None, self.S_CLOSED)              # closed is a goal state, multiple closes are ok and are no-ops
        self._fsm.add_transition(self.I_CLOSE, self.S_INIT, None, self.S_CLOSED)                # INIT to CLOSED is fine too

    def test_delete_xp(self):
        # same as test_delete_xs
        xp = self.container.ex_manager.create_xp('test_xp')

        # prove XS is declared already (can't declare the same named one with diff params)
        ch = self.container.node.channel(RecvChannel)
        ch._exchange_auto_delete = not xp._xs._xs_auto_delete

        self.assertRaises(TransportError, ch._declare_exchange, xp.exchange)

        # now let's delete via ex manager
        self.container.ex_manager.delete_xp(xp)

        # grab another channel and declare (should work fine this time)
        ch = self.container.node.channel(RecvChannel)
        ch._exchange_auto_delete = not xp._xs._xs_auto_delete

        ch._declare_exchange(xp.exchange)

        # cool, now cleanup (we don't expose this via Channel)
        at = AMQPTransport.get_instance()
        at.delete_exchange_impl(ch._amq_chan, xp.exchange)

    def test_delete_xs(self):
        # we get interesting here.  we have to create or xs, try to declare again to prove it is there,
        # then delete, then declare to prove we CAN create, then make sure to clean it up.  whew.
        xs = self.container.ex_manager.create_xs('test_xs')

        # prove XS is declared already (can't declare the same named one with diff params)
        ch = self.container.node.channel(RecvChannel)
        ch._exchange_auto_delete = not xs._xs_auto_delete

        self.assertRaises(TransportError, ch._declare_exchange, xs.exchange)

        # now let's delete via ex manager
        self.container.ex_manager.delete_xs(xs)

        # grab another channel and declare (should work fine this time)
        ch = self.container.node.channel(RecvChannel)
        ch._exchange_auto_delete = not xs._xs_auto_delete

        ch._declare_exchange(xs.exchange)

        # cool, now cleanup (we don't expose this via Channel)
        at = AMQPTransport.get_instance()
        at.delete_exchange_impl(ch._amq_chan, xs.exchange)