# Path: afkak/_protocol.py
# class KafkaBootstrapProtocol(_BaseKafkaProtocol):
#     """
#     `KafkaBootstrapProtocol` sends and receives Kafka messages.
# 
#     It knows just enough about Kafka message framing to correlate responses
#     with requests. A deferred is issued for every request and fires when
#     a response is received or the connection is lost.
# 
#     :ivar dict _pending:
#         Map of correlation ID to Deferred.
#     """
#     _log = Logger()
# 
#     def connectionMade(self):
#         self._pending = {}
#         self._failed = None
# 
#     def stringReceived(self, response):
#         """
#         Handle a response from the broker.
#         """
#         correlation_id = response[0:4]
#         try:
#             d = self._pending.pop(correlation_id)
#         except KeyError:
#             self._log.warn((
#                 "Response has unknown correlation ID {correlation_id!r}."
#                 " Dropping connection to {peer}."
#             ), correlation_id=correlation_id, peer=self.transport.getPeer())
#             self.transport.loseConnection()
#         else:
#             d.callback(response)
# 
#     def connectionLost(self, reason=connectionDone):
#         """
#         Mark the protocol as failed and fail all pending operations.
#         """
#         self._failed = reason
#         pending, self._pending = self._pending, None
#         for d in pending.values():
#             d.errback(reason)
# 
#     def lengthLimitExceeded(self, length):
#         self._log.error(
#             "Broker at {peer} sent a {length:,d} byte message, exceeding the size limit of {max_length:,d}.",
#             peer=self.transport.getPeer(), length=length, max_length=self.MAX_LENGTH,
#         )
#         self.transport.loseConnection()
# 
#     def request(self, request):
#         """
#         Send a request to the Kafka broker.
# 
#         :param bytes request:
#             The bytes of a Kafka `RequestMessage`_ structure. It must have
#             a unique (to this connection) correlation ID.
# 
#         :returns:
#             `Deferred` which will:
# 
#               - Succeed with the bytes of a Kafka `ResponseMessage`_
#               - Fail when the connection terminates
# 
#         .. _RequestMessage:: https://kafka.apache.org/protocol.html#protocol_messages
# 
#         """
#         if self._failed is not None:
#             return fail(self._failed)
#         correlation_id = request[4:8]
#         assert correlation_id not in self._pending
#         d = Deferred()
#         self.sendString(request)
#         self._pending[correlation_id] = d
#         return d
# 
# class KafkaProtocol(_BaseKafkaProtocol):
#     """
#     Very thin wrapper around the Int32StringReceiver
#     Simply knows to call its factory.handleResponse()
#     method with the string received by stringReceived() and
#     to cleanup the factory reference when the connection is lost
#     """
#     factory = None
# 
#     def stringReceived(self, string):
#         self.factory.handleResponse(string)
# 
#     def connectionLost(self, reason=connectionDone):
#         self.factory._connectionLost(reason)
#         self.factory = None
# 
#     def lengthLimitExceeded(self, length):
#         log.error("Broker at %s sent a %d byte message, exceeding the size limit of %d. "
#                   "Terminating connection.", self.transport.getPeer(), length,
#                   self.MAX_LENGTH)
#         self.transport.loseConnection()
# 
# Path: afkak/test/logtools.py
# @contextmanager
# def capture_logging(logger, level=logging.DEBUG):
#     """Context manager that captures log records
# 
#     :param logger: :class:`logging.Logger`. Propagation of messages is disabled
#         while the context manager is active.
#     :param level:
#         Log level assigned to the logger while the context manager is active.
#         Defaults to `logging.DEBUG`.
# 
#         If you pass `logging.NOTSET` the level will be inherited from the
#         global configuration, which may vary depending on the test harness in
#         use (e.g., nose meddles with it). This may lead to brittle tests.
# 
#     :returns: :class:`list` of :class:`logging.LogRecord` objects
#     """
#     handler = _CaptureHandler()
#     was_propagating = logger.propagate
#     was_level = logger.level
#     logger.propagate = False
#     logger.setLevel(level)
#     logger.addHandler(handler)
#     try:
#         yield handler.records
#     finally:
#         logger.removeHandler(handler)
#         logger.propagate = was_propagating
#         logger.setLevel(was_level)


import logging
import unittest
from unittest import mock
from twisted.internet.address import IPv4Address
from twisted.internet.error import ConnectionDone, ConnectionLost
from twisted.logger import LogLevel, globalLogPublisher
from twisted.python.failure import Failure
from twisted.test.iosim import FakeTransport
from twisted.test.proto_helpers import (
    EventLoggingObserver, StringTransportWithDisconnection,
)
from twisted.trial.unittest import SynchronousTestCase
from .._protocol import KafkaBootstrapProtocol, KafkaProtocol
from .logtools import capture_logging
        peer = IPv4Address('TCP', '1.2.3.4', 1234)
        kp = KafkaProtocol()
        kp.factory = factory_spy = mock.Mock(wraps=TheFactory())
        kp.transport = StringTransportWithDisconnection(peerAddress=peer)
        kp.transport.protocol = kp

        with capture_logging(logging.getLogger('afkak.protocol')) as records:
            kp.lengthLimitExceeded(too_long)

        self.assertEqual(1, len(factory_spy._connectionLost.mock_calls))

        [record] = records
        record.getMessage()  # Formats okay.
        self.assertEqual((
            'Broker at %s sent a %d byte message, exceeding the size limit of %d. '
            'Terminating connection.'
        ), record.msg)
        self.assertEqual((peer, too_long, kp.MAX_LENGTH), record.args)


class KafkaBootstrapProtocolTests(SynchronousTestCase):
    """Test `afkak._protocol.KafkaBootstrapProtocol`

    :ivar peer: Peer IAddress, an IPv4 address
    :ivar protocol: `KafkaBootstrapProtocol` object, connected in setUp
    :ivar transport: `FakeTransport` object associated with the protocol,
        connected in setUp
    """
    def setUp(self):
        self.peer = IPv4Address('TCP', 'kafka', 9072)

# Path: afkak/_protocol.py
# class KafkaBootstrapProtocol(_BaseKafkaProtocol):
#     """
#     `KafkaBootstrapProtocol` sends and receives Kafka messages.
# 
#     It knows just enough about Kafka message framing to correlate responses
#     with requests. A deferred is issued for every request and fires when
#     a response is received or the connection is lost.
# 
#     :ivar dict _pending:
#         Map of correlation ID to Deferred.
#     """
#     _log = Logger()
# 
#     def connectionMade(self):
#         self._pending = {}
#         self._failed = None
# 
#     def stringReceived(self, response):
#         """
#         Handle a response from the broker.
#         """
#         correlation_id = response[0:4]
#         try:
#             d = self._pending.pop(correlation_id)
#         except KeyError:
#             self._log.warn((
#                 "Response has unknown correlation ID {correlation_id!r}."
#                 " Dropping connection to {peer}."
#             ), correlation_id=correlation_id, peer=self.transport.getPeer())
#             self.transport.loseConnection()
#         else:
#             d.callback(response)
# 
#     def connectionLost(self, reason=connectionDone):
#         """
#         Mark the protocol as failed and fail all pending operations.
#         """
#         self._failed = reason
#         pending, self._pending = self._pending, None
#         for d in pending.values():
#             d.errback(reason)
# 
#     def lengthLimitExceeded(self, length):
#         self._log.error(
#             "Broker at {peer} sent a {length:,d} byte message, exceeding the size limit of {max_length:,d}.",
#             peer=self.transport.getPeer(), length=length, max_length=self.MAX_LENGTH,
#         )
#         self.transport.loseConnection()
# 
#     def request(self, request):
#         """
#         Send a request to the Kafka broker.
# 
#         :param bytes request:
#             The bytes of a Kafka `RequestMessage`_ structure. It must have
#             a unique (to this connection) correlation ID.
# 
#         :returns:
#             `Deferred` which will:
# 
#               - Succeed with the bytes of a Kafka `ResponseMessage`_
#               - Fail when the connection terminates
# 
#         .. _RequestMessage:: https://kafka.apache.org/protocol.html#protocol_messages
# 
#         """
#         if self._failed is not None:
#             return fail(self._failed)
#         correlation_id = request[4:8]
#         assert correlation_id not in self._pending
#         d = Deferred()
#         self.sendString(request)
#         self._pending[correlation_id] = d
#         return d
# 
# class KafkaProtocol(_BaseKafkaProtocol):
#     """
#     Very thin wrapper around the Int32StringReceiver
#     Simply knows to call its factory.handleResponse()
#     method with the string received by stringReceived() and
#     to cleanup the factory reference when the connection is lost
#     """
#     factory = None
# 
#     def stringReceived(self, string):
#         self.factory.handleResponse(string)
# 
#     def connectionLost(self, reason=connectionDone):
#         self.factory._connectionLost(reason)
#         self.factory = None
# 
#     def lengthLimitExceeded(self, length):
#         log.error("Broker at %s sent a %d byte message, exceeding the size limit of %d. "
#                   "Terminating connection.", self.transport.getPeer(), length,
#                   self.MAX_LENGTH)
#         self.transport.loseConnection()
# 
# Path: afkak/test/logtools.py
# @contextmanager
# def capture_logging(logger, level=logging.DEBUG):
#     """Context manager that captures log records
# 
#     :param logger: :class:`logging.Logger`. Propagation of messages is disabled
#         while the context manager is active.
#     :param level:
#         Log level assigned to the logger while the context manager is active.
#         Defaults to `logging.DEBUG`.
# 
#         If you pass `logging.NOTSET` the level will be inherited from the
#         global configuration, which may vary depending on the test harness in
#         use (e.g., nose meddles with it). This may lead to brittle tests.
# 
#     :returns: :class:`list` of :class:`logging.LogRecord` objects
#     """
#     handler = _CaptureHandler()
#     was_propagating = logger.propagate
#     was_level = logger.level
#     logger.propagate = False
#     logger.setLevel(level)
#     logger.addHandler(handler)
#     try:
#         yield handler.records
#     finally:
#         logger.removeHandler(handler)
#         logger.propagate = was_propagating
#         logger.setLevel(was_level)


# Path: afkak/test/test_protocol.py
import logging
import unittest
from unittest import mock
from twisted.internet.address import IPv4Address
from twisted.internet.error import ConnectionDone, ConnectionLost
from twisted.logger import LogLevel, globalLogPublisher
from twisted.python.failure import Failure
from twisted.test.iosim import FakeTransport
from twisted.test.proto_helpers import (
    EventLoggingObserver, StringTransportWithDisconnection,
)
from twisted.trial.unittest import SynchronousTestCase
from .._protocol import KafkaBootstrapProtocol, KafkaProtocol
from .logtools import capture_logging

        peer = IPv4Address('TCP', '1.2.3.4', 1234)
        kp = KafkaProtocol()
        kp.factory = factory_spy = mock.Mock(wraps=TheFactory())
        kp.transport = StringTransportWithDisconnection(peerAddress=peer)
        kp.transport.protocol = kp

        with capture_logging(logging.getLogger('afkak.protocol')) as records:
            kp.lengthLimitExceeded(too_long)

        self.assertEqual(1, len(factory_spy._connectionLost.mock_calls))

        [record] = records
        record.getMessage()  # Formats okay.
        self.assertEqual((
            'Broker at %s sent a %d byte message, exceeding the size limit of %d. '
            'Terminating connection.'
        ), record.msg)
        self.assertEqual((peer, too_long, kp.MAX_LENGTH), record.args)


class KafkaBootstrapProtocolTests(SynchronousTestCase):
    """Test `afkak._protocol.KafkaBootstrapProtocol`

    :ivar peer: Peer IAddress, an IPv4 address
    :ivar protocol: `KafkaBootstrapProtocol` object, connected in setUp
    :ivar transport: `FakeTransport` object associated with the protocol,
        connected in setUp
    """
    def setUp(self):
        self.peer = IPv4Address('TCP', 'kafka', 9072)

        self.protocol = KafkaBootstrapProtocol()