def testIllegalInboxDelete(self):
        """
        Test what happens if we try to delete the user Inbox.
        We expect that operation to fail.
        """
        self.stashed = None

        def login():
            return self.client.login(TEST_USER, TEST_PASSWD)

        def delete():
            return self.client.delete('inbox')

        def stash(result):
            self.stashed = result

        d1 = self.connected.addCallback(strip(login))
        d1.addCallbacks(strip(delete), self._ebGeneral)
        d1.addBoth(stash)
        d1.addCallbacks(self._cbStopClient, self._ebGeneral)
        d2 = self.loopback()
        d = defer.gatherResults([d1, d2])
        d.addCallback(lambda _: self.failUnless(isinstance(self.stashed,
                                                           failure.Failure)))
        return d 

def send(self, cmd, args, expect_ack=None, ack_callback=None, response_callback=None, expect_status=None, status_callback=None, update_callback=None, update_handler=None):
        if self.ready:
            self.cmd = cmd
            if 'request' in cmd:
                self.response_rows = []
            ret = self.api.gateway_send('%s %s %s' % (cmd, self.id, args))
            self.ack_pending = expect_ack
            self.ack_callback = ack_callback
            self.response_pending = bool(response_callback)
            self.response_callback = response_callback
            self.status_pending = expect_status
            self.status_callback = status_callback
            self.update_callback = update_callback
            self.update_handler = update_handler
        else:
            if self.on_connect_action:
                self.api.error_handler(self.id, 'Failure: on_connect_action already exists: %s' % repr(self.on_connect_action))
                ret = False
            else:
                self.api.output('%s storing on_connect_action (%s)...' % (self, cmd))
                self.on_connect_action = (cmd, args, expect_ack, ack_callback, response_callback, expect_status, status_callback, update_callback, update_handler)
                ret = True
        return ret 

def set_order_route(self, route, callback):
        #print('set_order_route(%s, %s) type=%s %s' % (repr(route), repr(callback), type(route), (type(route) in [str, unicode])))
        if type(route) in [str, unicode]:
            if route.startswith('{'):
                route = json.loads(route)
	    elif route.startswith('"'):
                route = {json.loads(route): None}
            else:
                route = {route: None}
        if (type(route)==dict) and (len(route.keys()) == 1) and (type(route.keys()[0]) in [str, unicode]):
            self.order_route = route
            if callback:
                self.get_order_route(callback)
        else:
            if callback:
                callback.errback(Failure(Exception('cannot set order route %s' % route)))
            else:
                self.error_handler(None, 'Cannot set order route %s' % repr(route)) 

def _complete_batch_send(self, resp):
        """Complete the processing of our batch send operation

        Clear the deferred tracking our current batch processing
        and reset our retry count and retry interval
        Return none to eat any errors coming from up the deferred chain
        """
        self._batch_send_d = None
        self._req_attempts = 0
        self._retry_interval = self._init_retry_interval
        if isinstance(resp, Failure) and not resp.check(tid_CancelledError,
                                                        CancelledError):
            log.error(
                "Failure detected in _complete_batch_send: %r", resp,
                exc_info=(resp.type, resp.value, resp.getTracebackObject()),
            )
        return 

def test_consumer_error_during_offset(self):
        topic = 'error_during_offset'
        part = 991
        reqs_ds = [Deferred(), Deferred()]
        clock = MemoryReactorClock()
        mockclient = Mock(reactor=clock)
        mockclient.send_offset_request.side_effect = reqs_ds
        consumer = Consumer(mockclient, topic, part, Mock())
        d = consumer.start(OFFSET_LATEST)
        # Make sure request for offset was made
        request = OffsetRequest(topic, part, OFFSET_LATEST, 1)
        mockclient.send_offset_request.assert_called_once_with([request])
        # Errback the first request
        f = Failure(KafkaUnavailableError())  # Perhaps kafka wasn't up yet...
        with patch.object(kconsumer, 'log'):
            reqs_ds[0].errback(f)
        # Advance the clock to trigger the 2nd request
        clock.advance(consumer.retry_delay + 1)  # fire the callLater
        self.assertEqual(2, mockclient.send_offset_request.call_count)

        # Stop the consumer to cleanup any outstanding operations
        self.assertIsNone(consumer.stop())
        self.assertIsNone(self.successResultOf(d)) 

def test_consumer_offset_out_of_range_error_without_reset(self):
        topic = 'offset_out_of_range_error'
        part = 911
        offset = 10000
        fetch_ds = [Deferred()]
        clock = MemoryReactorClock()
        mockclient = Mock(reactor=clock)
        mockclient.send_fetch_request.side_effect = fetch_ds

        consumer = Consumer(mockclient, topic, part, Mock())
        d = consumer.start(offset)

        f = Failure(OffsetOutOfRangeError())
        fetch_ds[0].errback(f)

        self.assertEqual(self.failureResultOf(d), f)

        consumer.stop() 

def test_consumer_error(self):
        """
            get an unexpected stop error from a consumer
        """
        client = self.mock_client([])
        processor = Mock()
        group = ConsumerGroup(client, "group_id", "topic1", processor)
        start_d = group.start()
        self.assertNoResult(start_d)
        with patch('afkak._group.Consumer') as mock_consumer:
            mock_consumer.return_value.start.return_value = d = defer.Deferred()
            group.on_join_complete({"topic1": [1]})
            self.assertEqual(mock_consumer.return_value.start.called, True)
            d.errback(Failure(AssertionError()))
            self.failureResultOf(start_d, AssertionError)
            d.addErrback(lambda result: None) 

def test_producer_send_messages_no_retry_fail(self):
        client = Mock(reactor=MemoryReactorClock())
        f = Failure(BrokerNotAvailableError())
        client.send_produce_request.side_effect = [fail(f)]
        client.topic_partitions = {self.topic: [0, 1, 2, 3]}
        client.metadata_error_for_topic.return_value = False
        msgs = [self.msg("one"), self.msg("two")]

        producer = Producer(client, max_req_attempts=1)
        d = producer.send_messages(self.topic, msgs=msgs)
        # Check the expected request was sent
        msgSet = create_message_set(
            make_send_requests(msgs), producer.codec)
        req = ProduceRequest(self.topic, 0, msgSet)
        client.send_produce_request.assert_called_once_with(
            [req], acks=producer.req_acks, timeout=producer.ack_timeout,
            fail_on_error=False)
        self.failureResultOf(d, BrokerNotAvailableError)

        producer.stop() 

def test_producer_send_messages_unexpected_err(self):
        client = Mock(reactor=MemoryReactorClock())
        f = Failure(TypeError())
        client.send_produce_request.side_effect = [fail(f)]
        client.topic_partitions = {self.topic: [0, 1, 2, 3]}
        client.metadata_error_for_topic.return_value = False
        msgs = [self.msg("one"), self.msg("two")]

        producer = Producer(client)
        # FIXME: Don't use patch to test logging
        with patch.object(aProducer, 'log') as klog:
            d = producer.send_messages(self.topic, msgs=msgs)
            klog.error.assert_called_once_with(
                'Unexpected failure: %r in _handle_send_response', f)
        self.failureResultOf(d, TypeError)

        producer.stop() 

def test_producer_stop_during_request(self):
        """
        Test stopping producer while it's waiting for reply from client
        """
        clock = MemoryReactorClock()
        client = Mock(reactor=clock)
        f = Failure(BrokerNotAvailableError())
        ret = [fail(f), Deferred()]
        client.send_produce_request.side_effect = ret
        client.topic_partitions = {self.topic: [0, 1, 2, 3]}
        client.metadata_error_for_topic.return_value = False
        msgs = [self.msg("one"), self.msg("two")]
        batch_n = 2

        producer = Producer(client, batch_every_n=batch_n, batch_send=True)
        d = producer.send_messages(self.topic, msgs=msgs)
        # At first, there's no result. Have to retry due to first failure
        self.assertNoResult(d)
        clock.advance(producer._retry_interval)

        producer.stop()
        self.failureResultOf(d, tid_CancelledError) 

def test_producer_stop_waiting_to_retry(self):
        """
        Test stopping producer while it's waiting to retry a request
        """
        clock = MemoryReactorClock()
        client = Mock(reactor=clock)
        f = Failure(BrokerNotAvailableError())
        ret = [fail(f)]
        client.send_produce_request.side_effect = ret
        client.topic_partitions = {self.topic: [0, 1, 2, 3]}
        client.metadata_error_for_topic.return_value = False
        msgs = [self.msg("one"), self.msg("two")]
        batch_n = 2

        producer = Producer(client, batch_every_n=batch_n, batch_send=True)
        d = producer.send_messages(self.topic, msgs=msgs)
        # At first, there's no result. Have to retry due to first failure
        self.assertNoResult(d)
        # Advance the clock, some, but not enough to retry
        clock.advance(producer._retry_interval / 2)
        # Stop the producer before the retry
        producer.stop()
        self.failureResultOf(d, tid_CancelledError) 

def process(self):
        """
        Process a request.
        """

        # get site from channel
        self.site = self.channel.site

        # set various default headers
        self.setHeader(b'server', version)
        self.setHeader(b'date', http.datetimeToString())

        # Resource Identification
        self.prepath = []
        self.postpath = list(map(unquote, self.path[1:].split(b'/')))

        try:
            resrc = self.site.getResourceFor(self)
            if resource._IEncodingResource.providedBy(resrc):
                encoder = resrc.getEncoder(self)
                if encoder is not None:
                    self._encoder = encoder
            self.render(resrc)
        except:
            self.processingFailed(failure.Failure()) 

def handleResponse(self, response):
        if self.quietLoss:
            return
        if self.failed:
            self.factory.noPage(
                Failure(
                    error.Error(
                        self.status, self.message, response)))
        if self.factory.method == b'HEAD':
            # Callback with empty string, since there is never a response
            # body for HEAD requests.
            self.factory.page(b'')
        elif self.length != None and self.length != 0:
            self.factory.noPage(Failure(
                PartialDownloadError(self.status, self.message, response)))
        else:
            self.factory.page(response)
        # server might be stupid and not close connection. admittedly
        # the fact we do only one request per connection is also
        # stupid...
        self.transport.loseConnection() 

def request(self, method, uri, headers=None, bodyProducer=None):
        """
        Issue a request to the server indicated by the given C{uri}.

        An existing connection from the connection pool may be used or a new
        one may be created.

        I{HTTP} and I{HTTPS} schemes are supported in C{uri}.

        @see: L{twisted.web.iweb.IAgent.request}
        """
        parsedURI = URI.fromBytes(uri)
        try:
            endpoint = self._getEndpoint(parsedURI)
        except SchemeNotSupported:
            return defer.fail(Failure())
        key = (parsedURI.scheme, parsedURI.host, parsedURI.port)
        return self._requestWithEndpoint(key, endpoint, method, parsedURI,
                                         headers, bodyProducer,
                                         parsedURI.originForm) 

def _handleRedirect(self, response, method, uri, headers, redirectCount):
        """
        Handle a redirect response, checking the number of redirects already
        followed, and extracting the location header fields.
        """
        if redirectCount >= self._redirectLimit:
            err = error.InfiniteRedirection(
                response.code,
                b'Infinite redirection detected',
                location=uri)
            raise ResponseFailed([Failure(err)], response)
        locationHeaders = response.headers.getRawHeaders(b'location', [])
        if not locationHeaders:
            err = error.RedirectWithNoLocation(
                response.code, b'No location header field', uri)
            raise ResponseFailed([Failure(err)], response)
        location = self._resolveLocation(uri, locationHeaders[0])
        deferred = self._agent.request(method, location, headers)
        def _chainResponse(newResponse):
            newResponse.setPreviousResponse(response)
            return newResponse
        deferred.addCallback(_chainResponse)
        return deferred.addCallback(
            self._handleResponse, method, uri, headers, redirectCount + 1) 

def writeTo(self, transport):
        """
        Format this L{Request} as an HTTP/1.1 request and write it to the given
        transport.  If bodyProducer is not None, it will be associated with an
        L{IConsumer}.

        @param transport: The transport to which to write.
        @type transport: L{twisted.internet.interfaces.ITransport} provider

        @return: A L{Deferred} which fires with L{None} when the request has
            been completely written to the transport or with a L{Failure} if
            there is any problem generating the request bytes.
        """
        if self.bodyProducer is None:
            # If the method semantics anticipate a body, include a
            # Content-Length even if it is 0.
            # https://tools.ietf.org/html/rfc7230#section-3.3.2
            if self.method in (b"PUT", b"POST"):
                self._writeToEmptyBodyContentLength(transport)
            else:
                self._writeHeaders(transport, None)
        elif self.bodyProducer.length is UNKNOWN_LENGTH:
            return self._writeToBodyProducerChunked(transport)
        else:
            return self._writeToBodyProducerContentLength(transport) 

def _disconnectParser(self, reason):
        """
        If there is still a parser, call its C{connectionLost} method with the
        given reason.  If there is not, do nothing.

        @type reason: L{Failure}
        """
        if self._parser is not None:
            parser = self._parser
            self._parser = None
            self._currentRequest = None
            self._finishedRequest = None
            self._responseDeferred = None

            # The parser is no longer allowed to do anything to the real
            # transport.  Stop proxying from the parser's transport to the real
            # transport before telling the parser it's done so that it can't do
            # anything.
            self._transportProxy._stopProxying()
            self._transportProxy = None
            parser.connectionLost(reason) 

def test_returnsBytes(self):
        """
        The return value of L{formatFailure} is a C{str} instance (not a
        C{unicode} instance) with numeric character references for any non-ASCII
        characters meant to appear in the output.
        """
        try:
            raise Exception("Fake bug")
        except:
            result = formatFailure(Failure())

        self.assertIsInstance(result, bytes)
        if _PY3:
            self.assertTrue(all(ch < 128 for ch in result))
        else:
            self.assertTrue(all(ord(ch) < 128 for ch in result))
        # Indentation happens to rely on NO-BREAK SPACE
        self.assertIn(b"&#160;", result) 

def test_renderNoFailure(self):
        """
        If the L{Deferred} fails, L{DeferredResource} reports the failure via
        C{processingFailed}, and does not cause an unhandled error to be
        logged.
        """
        request = DummyRequest([])
        d = request.notifyFinish()
        failure = Failure(RuntimeError())
        deferredResource = DeferredResource(defer.fail(failure))
        deferredResource.render(request)
        self.assertEqual(self.failureResultOf(d), failure)
        del deferredResource
        gc.collect()
        errors = self.flushLoggedErrors(RuntimeError)
        self.assertEqual(errors, []) 

def assertFlattensImmediately(self, root, target):
        """
        Assert that a root element, when flattened, is equal to a string, and
        performs no asynchronus Deferred anything.

        This version is more convenient in tests which wish to make multiple
        assertions about flattening, since it can be called multiple times
        without having to add multiple callbacks.

        @return: the result of rendering L{root}, which should be equivalent to
            L{target}.
        @rtype: L{bytes}
        """
        results = []
        it = self.assertFlattensTo(root, target)
        it.addBoth(results.append)
        # Do our best to clean it up if something goes wrong.
        self.addCleanup(it.cancel)
        if not results:
            self.fail("Rendering did not complete immediately.")
        result = results[0]
        if isinstance(result, Failure):
            result.raiseException()
        return results[0] 

def _invoke_callback(self, fd, events):
        if fd not in self._fds:
            return
        (reader, writer) = self._fds[fd]
        if reader:
            err = None
            if reader.fileno() == -1:
                err = error.ConnectionLost()
            elif events & IOLoop.READ:
                err = log.callWithLogger(reader, reader.doRead)
            if err is None and events & IOLoop.ERROR:
                err = error.ConnectionLost()
            if err is not None:
                self.removeReader(reader)
                reader.readConnectionLost(failure.Failure(err))
        if writer:
            err = None
            if writer.fileno() == -1:
                err = error.ConnectionLost()
            elif events & IOLoop.WRITE:
                err = log.callWithLogger(writer, writer.doWrite)
            if err is None and events & IOLoop.ERROR:
                err = error.ConnectionLost()
            if err is not None:
                self.removeWriter(writer)
                writer.writeConnectionLost(failure.Failure(err)) 

def resolve(self, host, port, family=0):
        # getHostByName doesn't accept IP addresses, so if the input
        # looks like an IP address just return it immediately.
        if twisted.internet.abstract.isIPAddress(host):
            resolved = host
            resolved_family = socket.AF_INET
        elif twisted.internet.abstract.isIPv6Address(host):
            resolved = host
            resolved_family = socket.AF_INET6
        else:
            deferred = self.resolver.getHostByName(utf8(host))
            resolved = yield gen.Task(deferred.addBoth)
            if isinstance(resolved, failure.Failure):
                resolved.raiseException()
            elif twisted.internet.abstract.isIPAddress(resolved):
                resolved_family = socket.AF_INET
            elif twisted.internet.abstract.isIPv6Address(resolved):
                resolved_family = socket.AF_INET6
            else:
                resolved_family = socket.AF_UNSPEC
        if family != socket.AF_UNSPEC and family != resolved_family:
            raise Exception('Requested socket family %d but got %d' %
                            (family, resolved_family))
        result = [
            (resolved_family, (resolved, port)),
        ]
        raise gen.Return(result) 

def _invoke_callback(self, fd, events):
        if fd not in self._fds:
            return
        (reader, writer) = self._fds[fd]
        if reader:
            err = None
            if reader.fileno() == -1:
                err = error.ConnectionLost()
            elif events & IOLoop.READ:
                err = log.callWithLogger(reader, reader.doRead)
            if err is None and events & IOLoop.ERROR:
                err = error.ConnectionLost()
            if err is not None:
                self.removeReader(reader)
                reader.readConnectionLost(failure.Failure(err))
        if writer:
            err = None
            if writer.fileno() == -1:
                err = error.ConnectionLost()
            elif events & IOLoop.WRITE:
                err = log.callWithLogger(writer, writer.doWrite)
            if err is None and events & IOLoop.ERROR:
                err = error.ConnectionLost()
            if err is not None:
                self.removeWriter(writer)
                writer.writeConnectionLost(failure.Failure(err)) 

def resolve(self, host, port, family=0):
        # getHostByName doesn't accept IP addresses, so if the input
        # looks like an IP address just return it immediately.
        if twisted.internet.abstract.isIPAddress(host):
            resolved = host
            resolved_family = socket.AF_INET
        elif twisted.internet.abstract.isIPv6Address(host):
            resolved = host
            resolved_family = socket.AF_INET6
        else:
            deferred = self.resolver.getHostByName(utf8(host))
            resolved = yield gen.Task(deferred.addBoth)
            if isinstance(resolved, failure.Failure):
                resolved.raiseException()
            elif twisted.internet.abstract.isIPAddress(resolved):
                resolved_family = socket.AF_INET
            elif twisted.internet.abstract.isIPv6Address(resolved):
                resolved_family = socket.AF_INET6
            else:
                resolved_family = socket.AF_UNSPEC
        if family != socket.AF_UNSPEC and family != resolved_family:
            raise Exception('Requested socket family %d but got %d' %
                            (family, resolved_family))
        result = [
            (resolved_family, (resolved, port)),
        ]
        raise gen.Return(result) 

def _route_msg(self, encrypt_and_sign_result, recipient, raw):
        """
        Sends the msg using the ESMTPSenderFactory.

        :param encrypt_and_sign_result: A tuple containing the 'maybe'
                                        encrypted message and the recipient
        :type encrypt_and_sign_result: tuple
        """
        message, recipient = encrypt_and_sign_result
        msg = message.as_string(False)

        d = None
        for sender in self._senders:
            if sender.can_send(recipient.dest.addrstr):
                self.log.debug('Sending message to %s with: %s'
                               % (recipient, str(sender)))
                d = sender.send(recipient, msg)
                break

        if d is None:
            return self.sendError(Failure(), raw)

        emit_async(catalog.SMTP_SEND_MESSAGE_START,
                   self._from_address, recipient.dest.addrstr)
        d.addCallback(self.sendSuccess)
        d.addErrback(self.sendError, raw)
        return d 

def dispatch(self, msg):
        cmd = msg[0]
        _method = getattr(self, 'do_' + cmd.upper(), None)

        if not _method:
            return defer.fail(failure.Failure(RuntimeError('No such command')))

        return defer.maybeDeferred(_method, *msg) 

def testIllegalInboxRename(self):
        """
        Try to rename inbox. We expect it to fail. Then it would be not
        an inbox anymore, would it?
        """
        self.stashed = None

        def login():
            return self.client.login(TEST_USER, TEST_PASSWD)

        def rename():
            return self.client.rename('inbox', 'frotz')

        def stash(stuff):
            self.stashed = stuff

        d1 = self.connected.addCallback(strip(login))
        d1.addCallbacks(strip(rename), self._ebGeneral)
        d1.addBoth(stash)
        d1.addCallbacks(self._cbStopClient, self._ebGeneral)
        d2 = self.loopback()
        d = defer.gatherResults([d1, d2])
        d.addCallback(lambda _:
                      self.failUnless(isinstance(
                          self.stashed, failure.Failure)))
        return d 

def test_send_error_bounces_if_bouncer_is_provided(self):
        bouncer = MagicMock()
        outgoing_mail = OutgoingMail(self.from_address, self.keymanager,
                                     bouncer)

        failure = Failure(exc_value=Exception())
        origmsg = 'message'
        outgoing_mail.sendError(failure, origmsg)

        bouncer.bounce_message.assert_called() 

def emit_message(self, message):

        # Compute storage location from topology information.
        storage_location = self.strategy.topology_to_storage(message.topology, message_type=message.type)
        log.debug(u'Storage location: {storage}', storage=dict(storage_location))

        # Store data or event.
        if message.type in (MessageType.DATA_CONTAINER, MessageType.EVENT):
            self.store_message(storage_location, message.data)

        # Provision graphing subsystem.
        if message.type == MessageType.DATA_CONTAINER:
            # TODO: Purge message from fields to be used as tags
            # Namely:
            # 'geohash',
            # 'location', 'location_id', 'location_name', 'sensor_id', 'sensor_type',
            # 'latitude', 'longitude', 'lat', 'lon'
            for graphing_subsystem in self.graphing:

                # Mix in references to each other. A bit of a hack, but okay for now :-).
                graphing_subsystem.strategy = self.strategy

                subsystem_name = graphing_subsystem.__class__.__name__
                log.debug(u'Provisioning Grafana with {name}', name=subsystem_name)
                try:
                    graphing_subsystem.provision(storage_location, message.data, topology=message.topology)

                except Exception as ex:
                    log.failure(u'Grafana provisioning failed for storage={storage}, message={message}:\n{log_failure}',
                                storage=storage_location.dump(), message=message.data,
                                level=LogLevel.error)

                    return Failure(Exception('Grafana provisioning failed'))

        return True 

def mqtt_process_error(self, failure, topic, payload):
        """
        Failure handling

        :param failure: Failure object from Twisted
        :param topic:   Full MQTT topic
        :param payload: Raw MQTT payload
        """

        # Log failure
        log.failure(u'Processing MQTT message failed from topic "{topic}":\n{log_failure}', topic=topic, failure=failure, level=LogLevel.error)

        # MQTT error signalling
        self.mqtt_publish_error(failure, topic, payload)