diff --git a/kafka/future.py b/kafka/future.py index 5c94be734..f09a00526 100644 --- a/kafka/future.py +++ b/kafka/future.py @@ -19,7 +19,7 @@ class Future: selector), which subclasses ``Future`` and adds ``__await__``. Keeping ``__await__`` off the base makes the invariant type-enforced -- awaiting a plain handoff ``Future`` raises immediately rather than silently working on - one backend and breaking on another. See ``kafka.net.backend.BackendFuture`` + one backend and breaking on another. See ``kafka.net.backend.NetBackendFuture`` for the awaitable contract. """ __slots__ = ('is_done', 'value', 'exception', '_callbacks', '_errbacks', '_lock') diff --git a/kafka/net/asyncio_backend.py b/kafka/net/asyncio_backend.py index ef8cc4c20..15ee91de8 100644 --- a/kafka/net/asyncio_backend.py +++ b/kafka/net/asyncio_backend.py @@ -23,7 +23,7 @@ class AsyncioFuture(Future): Inherits ``kafka.future.Future``'s callback core and overrides only ``__await__`` to bridge to an ``asyncio.Future`` so an asyncio Task can - await it. Per the BackendFuture contract this is created and resolved on + await it. Per the NetBackendFuture contract this is created and resolved on the loop thread only; a fresh asyncio.Future is minted per awaiter, so fan-out (multiple awaiters / callbacks) is preserved. """ @@ -392,9 +392,6 @@ def data_received(self, data): self.transport._bump_read() self._conn.data_received(data) - def eof_received(self): - return self._conn.eof_received() - def connection_lost(self, exc): self._conn.connection_lost(exc) @@ -411,7 +408,7 @@ class _AsyncioTransport: Wraps an asyncio transport + its protocol adapter and exposes the surface KafkaConnection / the manager drive (a superset of the NetBackend Transport protocol): write/close/abort/is_closing, pause/resume_reading, set/get - protocol, host/host_port/getPeer, and last_activity for idle sweeping. + protocol, host/host_port/get_peer, and last_activity for idle sweeping. """ def __init__(self, transport, host, port): @@ -461,7 +458,7 @@ def resume_reading(self): except (RuntimeError, AttributeError): pass - def getPeer(self): + def get_peer(self): peer = self._t.get_extra_info('peername') return peer if peer is not None else (self.host, self._port) diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 823b0149a..3843d46a5 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -9,7 +9,7 @@ implementation; an asyncio backend (and eventually Twisted) implements the same surface so it can be swapped in via ``net=`` without touching core code. -The :class:`BackendFuture` contract is the surface of the +The :class:`NetBackendFuture` contract is the surface of the loop-awaitable futures a backend hands out from ``net.create_future()``. The selector's implementation is ``kafka.net.selector.SelectorFuture``; an asyncio (and eventually Twisted) backend supplies its own. @@ -43,7 +43,7 @@ * **Timing** -- ``sleep`` (backend-specific awaitable; core coroutines await it). * **Connection** -- ``create_connection`` (returns a :class:`Transport`). * **Cross-thread bridge** -- ``run`` (schedule on the loop, block the caller). -* **Future factory** -- ``create_future`` (see ``BackendFuture``). +* **Future factory** -- ``create_future`` (see ``NetBackendFuture``). * **Cross-thread wake** -- ``wakeup``. """ import importlib @@ -51,7 +51,7 @@ @runtime_checkable -class BackendFuture(Protocol): +class NetBackendFuture(Protocol): """Contract for the awaitable futures returned by ``net.create_future()``. A pluggable async backend (the kafka.net selector, asyncio, Twisted, ...) @@ -60,7 +60,7 @@ class BackendFuture(Protocol): backends. The selector's ``SelectorFuture`` is the reference implementation: it subclasses the thread-safe ``kafka.future.Future`` (the portable callback core) and adds ``__await__``. A plain ``Future`` is deliberately NOT a - BackendFuture -- it has no ``__await__`` -- so awaiting a cross-thread + NetBackendFuture -- it has no ``__await__`` -- so awaiting a cross-thread handoff future fails loudly instead of silently working on one backend. Pinned semantics -- the three axes where backends could otherwise diverge: @@ -93,19 +93,19 @@ class BackendFuture(Protocol): exception: Any def __await__(self): ... - def success(self, value: Any) -> 'BackendFuture': ... - def failure(self, e: Any) -> 'BackendFuture': ... - def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... - def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... - def add_both(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... - def chain(self, future: 'BackendFuture') -> 'BackendFuture': ... + def success(self, value: Any) -> 'NetBackendFuture': ... + def failure(self, e: Any) -> 'NetBackendFuture': ... + def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'NetBackendFuture': ... + def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'NetBackendFuture': ... + def add_both(self, f: Callable, *args: Any, **kwargs: Any) -> 'NetBackendFuture': ... + def chain(self, future: 'NetBackendFuture') -> 'NetBackendFuture': ... def succeeded(self) -> bool: ... def failed(self) -> bool: ... @runtime_checkable -class Transport(Protocol): - """The transport surface a backend's ``create_connection`` returns. +class NetTransport(Protocol): + """The transport surface used by NetProtocol / NetBackend. The subset of the ``asyncio.Transport`` / Twisted ``ITransport`` surface that ``KafkaConnection`` actually drives (plus ``last_activity``, which the @@ -113,8 +113,7 @@ class Transport(Protocol): builds one and wires it to the conn. The selector's ``KafkaTCPTransport`` and an asyncio-transport adapter both satisfy it. """ - - # Monotonic timestamp of the last read/write; manager idle-sweeping reads it. + # Monotonic timestamp of the last read/write; used for idle-sweeping last_activity: float def write(self, data: bytes) -> None: ... @@ -124,6 +123,20 @@ def is_closing(self) -> bool: ... def pause_reading(self) -> None: ... def resume_reading(self) -> None: ... def host_port(self) -> Tuple[str, int]: ... + def get_protocol(self) -> 'NetProtocol': ... + def set_protocol(self, protocol: 'NetProtocol') -> None: ... + + +@runtime_checkable +class NetProtocol(Protocol): + # Monotonic timestamp of the last read/write; used for idle-sweeping + last_activity: float + + def connection_made(self, transport: NetTransport) -> None: ... + def data_received(self, data: bytes) -> None: ... + def connection_lost(self, exc: Any = None) -> None: ... + def pause_writing(self) -> None: ... + def resume_writing(self) -> None: ... @runtime_checkable @@ -155,7 +168,7 @@ def call_soon(self, task: Any) -> Any: def call_soon_threadsafe(self, callback: Any) -> Any: """``call_soon`` from another thread; wakes the loop.""" - def call_soon_with_future(self, coro: Any, *args: Any) -> BackendFuture: + def call_soon_with_future(self, coro: Any, *args: Any) -> NetBackendFuture: """Schedule ``coro`` and return a future that resolves with its result.""" def call_at(self, when: float, task: Any) -> Any: @@ -174,7 +187,7 @@ def sleep(self, delay: float) -> Any: # --- connection seam -------------------------------------------------- async def create_connection( self, - protocol: Any, + protocol: NetProtocol, host: str, port: int, *, @@ -211,8 +224,8 @@ def run(self, coro: Any, *args: Any, timeout_ms: Optional[float] = None) -> Any: """ # --- future factory --------------------------------------------------- - def create_future(self) -> BackendFuture: - """Create a loop-awaitable future (see ``BackendFuture``).""" + def create_future(self) -> NetBackendFuture: + """Create a loop-awaitable future (see ``NetBackendFuture``).""" # --- misc ------------------------------------------------------------- def wakeup(self) -> None: diff --git a/kafka/net/connection.py b/kafka/net/connection.py index 1107ce164..414145b34 100644 --- a/kafka/net/connection.py +++ b/kafka/net/connection.py @@ -221,15 +221,6 @@ def data_received(self, data): self.unpause('max_in_flight') self._reauth.on_response_processed() - def eof_received(self): - """ Called when the other end calls write_eof() or equivalent. - - If this returns a false value (including None), the transport - will close itself. If it returns a true value, closing the - transport is up to the protocol. - """ - return False - def connection_lost(self, exc): """ Called when the connection is lost or closed. @@ -283,7 +274,7 @@ def connection_made(self, transport): self.initializing = True self.transport.resume_reading() try: - log_prefix = 'node=%s[%s:%s]' % (self.node_id, *self.transport.getPeer()[0:2]) + log_prefix = 'node=%s[%s:%s]' % (self.node_id, *self.transport.get_peer()[0:2]) except Exception: log.exception('Failed to build connection log_prefix') log_prefix = '' @@ -456,7 +447,7 @@ async def _sasl_authenticate(self, timeout_at=None): # Prefer the configured hostname (stored on the transport) so that # mechanisms like GSSAPI construct service principals against the # user-supplied name, not whichever IP getaddrinfo handed us. - sasl_host = self.transport.host if self.transport.host else self.transport.getPeer()[0] + sasl_host = self.transport.host if self.transport.host else self.transport.get_peer()[0] mechanism = get_sasl_mechanism(self.config['sasl_mechanism'])( host=sasl_host, **self.config) diff --git a/kafka/net/selector.py b/kafka/net/selector.py index bd473abb6..6520a7c6e 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -43,7 +43,7 @@ def _initialize_coro(maybe_coro): class SelectorFuture(Future): - """The NetworkSelector's loop-awaitable future (see backend.BackendFuture). + """The NetworkSelector's loop-awaitable future (see backend.NetBackendFuture). ``kafka.future.Future`` is the thread-safe callback/handoff core with no ``__await__``; ``SelectorFuture`` adds it: ``yield self`` suspends the @@ -550,7 +550,7 @@ def reschedule(self, when, task): return task def create_future(self): - """Create a loop-awaitable future (see backend.BackendFuture). + """Create a loop-awaitable future (see backend.NetBackendFuture). Portability seam for pluggable backends: core coroutines call this instead of constructing ``Future`` directly, so an alternate backend diff --git a/kafka/net/transport.py b/kafka/net/transport.py index 339d8198f..8846b848d 100644 --- a/kafka/net/transport.py +++ b/kafka/net/transport.py @@ -32,7 +32,6 @@ def __init__(self, net, sock, host=None): def last_activity(self): return max(self.last_write, self.last_read) - # AsyncIO def is_closing(self): """Return True if the transport is closing or closed.""" return self._closed @@ -128,39 +127,6 @@ def _sock_recv(self): recvd_data = b''.join(recvd) return recvd_data, err - """Interface for write-only transports.""" - - def set_write_buffer_limits(self, high=None, low=None): - """Set the high- and low-water limits for write flow control. - - These two values control when to call the protocol's - pause_writing() and resume_writing() methods. If specified, - the low-water limit must be less than or equal to the - high-water limit. Neither value can be negative. - - The defaults are implementation-specific. If only the - high-water limit is given, the low-water limit defaults to an - implementation-specific value less than or equal to the - high-water limit. Setting high to zero forces low to zero as - well, and causes pause_writing() to be called whenever the - buffer becomes non-empty. Setting low to zero causes - resume_writing() to be called only once the buffer is empty. - Use of zero for either limit is generally sub-optimal as it - reduces opportunities for doing I/O and computation - concurrently. - """ - raise NotImplementedError - - def get_write_buffer_size(self): - """Return the current size of the write buffer.""" - raise NotImplementedError - - def get_write_buffer_limits(self): - """Get the high and low watermarks for write flow control. - Return a tuple (low, high) where low and high are - positive number of bytes.""" - raise NotImplementedError - def write(self, data): """Write some data bytes to the transport. @@ -177,16 +143,6 @@ def write(self, data): self._write_task = self._net.call_soon(self._write_to_sock) return len(data) - def writelines(self, list_of_data): - """Write a list (or any iterable) of data bytes to the transport.""" - if not self._write or self._closed: - raise RuntimeError('Transport closed for writes') - self._write_buffer.extend(list_of_data) - if not self._writing: - self._writing = True - self._write_task = self._net.call_soon(self._write_to_sock) - return sum(len(data) for data in list_of_data) - async def _write_to_sock(self): try: while self._write_buffer: @@ -229,22 +185,6 @@ def _sock_send(self): return total_bytes, Errors.KafkaConnectionError(e) return total_bytes, None - def write_eof(self): - """Close the write end after flushing buffered data. - - (This is like typing ^D into a UNIX program reading from stdin.) - - Data may still be received. - """ - log.debug('%s: write_eof', self) - self._write = False - if not self._write_buffer: - self._sock.shutdown(socket.SHUT_WR) - - def can_write_eof(self): - """Return True if this transport supports write_eof(), False if not.""" - return True - def abort(self, error=None): """Close the transport immediately. @@ -282,19 +222,7 @@ def _close(self, error=None): if proto is not None: proto.connection_lost(error) - # Twisted - def abortConnection(self): - """Close the connection abruptly.""" - return self.abort() - - def getHost(self): - """Similar to getPeer, but returns an address describing this side of the connection. - - Returns IPv4Address or IPv6Address. - """ - return self._sock.getsockname() - - def getPeer(self): + def get_peer(self): """Get the remote address of this connection. Treat this method with caution. It is the unfortunate result of the CGI and Jabber standards, @@ -305,49 +233,6 @@ def getPeer(self): """ return self._sock.getpeername() - def getTcpKeepAlive(self): - """Return if SO_KEEPALIVE is enabled.""" - return self._sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE) - - def getTcpNoDelay(self): - """Return if TCP_NODELAY is enabled.""" - return self._sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) - - def loseWriteConnection(self): - """Half-close the write side of a TCP connection.""" - return self.write_eof() - - def setTcpKeepAlive(self, enabled): - """Enable/disable SO_KEEPALIVE.""" - return self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, enabled) - - def setTcpNoDelay(self, enabled): - """Enable/disable TCP_NODELAY.""" - return self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, enabled) - - def loseConnection(self): - """Close my connection, after writing all pending data. - - Note that if there is a registered producer on a transport it will not be closed until the producer has been unregistered. - """ - return self.close() - - #def write(self, data): - # """Write some data to the physical connection, in sequence, in a non-blocking fashion. - # - # If possible, make sure that it is all written. No data will ever be lost, - # although (obviously) the connection may be closed before it all gets through. - # """ - # pass - - def writeSequence(self, data): - """Write an iterable of byte strings to the physical connection. - - If possible, make sure that all of the data is written to the socket at once, - without first copying it all into a single byte string. - """ - return self.writelines(data) - async def handshake(self): log.info('%s: connected to %s', self, self._sock) diff --git a/test/mock_broker.py b/test/mock_broker.py index 3061ad113..d0cc70753 100644 --- a/test/mock_broker.py +++ b/test/mock_broker.py @@ -130,12 +130,6 @@ def write(self, data): self._net.call_soon(self._process_requests) return len(data) - def writelines(self, data_list): - for data in data_list: - self._write_buffer.extend(data) - self.last_write = time.monotonic() - self._net.call_soon(self._process_requests) - def close(self): if not self._closed: self._closed = True @@ -150,16 +144,7 @@ def abort(self, error=None): self._protocol.connection_lost(error) self._protocol = None - def can_write_eof(self): - return False - - def write_eof(self): - pass - - def getHost(self): - return (self._host, 0) - - def getPeer(self): + def get_peer(self): return (self._host, self._port) def host_port(self): diff --git a/test/net/test_asyncio_backend.py b/test/net/test_asyncio_backend.py index aea1ebdae..b86205832 100644 --- a/test/net/test_asyncio_backend.py +++ b/test/net/test_asyncio_backend.py @@ -1,7 +1,7 @@ """Tests for the asyncio NetBackend (kafka/net/asyncio_backend.py). Covers backend-specific behavior (lifecycle, timers, cross-thread run), reuses -the shared BackendFuture conformance suite against the asyncio-backed future, +the shared NetBackendFuture conformance suite against the asyncio-backed future, and drives a real protocol round-trip through a MockBroker on a started AsyncioBackend -- the both-backends coverage for the async paths. """ @@ -18,7 +18,7 @@ from kafka.net.manager import KafkaConnectionManager from kafka.protocol.metadata import MetadataRequest from test.mock_broker import MockBroker -from test.net.test_backend_future import BackendFutureContract +from test.net.test_net_backend_future import NetBackendFutureContract @pytest.fixture @@ -158,8 +158,8 @@ def test_create_future_type(self, backend): assert not fut.is_done -class TestAsyncioBackendFuture(BackendFutureContract): - """Reuse the shared BackendFuture conformance suite for the asyncio future.""" +class TestAsyncioNetBackendFuture(NetBackendFutureContract): + """Reuse the shared NetBackendFuture conformance suite for the asyncio future.""" @pytest.fixture(autouse=True) def _net(self): @@ -193,8 +193,6 @@ def connection_made(self, transport): transport.resume_reading() def data_received(self, data): self.received += data - def eof_received(self): - return None def connection_lost(self, exc): self.closed = True def pause_writing(self): @@ -238,7 +236,7 @@ async def do(): await started_backend.create_connection(proto, host, port) transport = proto.transport assert transport.host_port() == '%s:%s' % (host, port) - assert transport.getPeer()[0:2] == (host, port) + assert transport.get_peer()[0:2] == (host, port) transport.write(b'ping') for _ in range(100): if proto.received: diff --git a/test/net/test_backend.py b/test/net/test_backend.py index c86318d09..d7a92f19d 100644 --- a/test/net/test_backend.py +++ b/test/net/test_backend.py @@ -11,7 +11,7 @@ import pytest from kafka.net.backend import ( - NetBackend, Transport, resolve_backend, register_backend, _BACKENDS, + NetBackend, NetTransport, resolve_backend, register_backend, _BACKENDS, ) from kafka.net.selector import NetworkSelector from kafka.net.transport import KafkaTCPTransport @@ -59,15 +59,15 @@ def test_readiness_primitives_and_poll_excluded(self): assert hasattr(net, name), name # still present on the selector impl -class TestTransportContract: +class TestNetTransportContract: def test_kafkatcptransport_satisfies_transport(self): - # Method-presence check against the Transport protocol (no socket needed). + # Method-presence check against the NetTransport protocol (no socket needed). for name in ('write', 'close', 'abort', 'is_closing', 'pause_reading', 'resume_reading', 'host_port'): assert callable(getattr(KafkaTCPTransport, name)), name def test_plain_object_is_not_transport(self): - assert not isinstance(object(), Transport) + assert not isinstance(object(), NetTransport) class TestOnIoThread: diff --git a/test/net/test_connection.py b/test/net/test_connection.py index b458c3be0..0ab54103b 100644 --- a/test/net/test_connection.py +++ b/test/net/test_connection.py @@ -67,7 +67,7 @@ class TestKafkaConnectionCheckApiVersions: def _make_conn(self, net, **kwargs): conn = KafkaConnection(net, node_id='test-0', **kwargs) transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) conn.transport = transport conn.initializing = True return conn @@ -264,7 +264,7 @@ def _make_send_ready(self, connection): connection.initializing = False connection.connected = True transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092, 0, 0) + transport.get_peer.return_value = ('127.0.0.1', 9092, 0, 0) connection.transport = transport connection.parser = MagicMock() cid = [0] @@ -310,7 +310,7 @@ class TestKafkaConnectionConnectionLifecycle: def test_connection_made(self, connection): transport = MagicMock() transport.get_protocol.return_value = None - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) connection.connection_made(transport) assert connection.transport is transport assert connection.initializing is True @@ -350,7 +350,7 @@ def test_init_complete(self, connection): def test_init_complete_drains_buffer(self, net): proto = KafkaConnection(net, node_id='test') transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092, 0, 0) + transport.get_peer.return_value = ('127.0.0.1', 9092, 0, 0) proto.transport = transport request = MagicMock() request.API_VERSION = 1 @@ -458,7 +458,7 @@ def test_sasl_authenticate_handshake_error(self, net): security_protocol='SASL_PLAINTEXT', sasl_mechanism='PLAIN') transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) conn.transport = transport conn.initializing = True @@ -483,7 +483,7 @@ def test_sasl_authenticate_mechanism_not_supported(self, net): security_protocol='SASL_PLAINTEXT', sasl_mechanism='PLAIN') transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) conn.transport = transport conn.initializing = True @@ -510,7 +510,7 @@ def test_sasl_authenticate_success(self, net): sasl_plain_username='user', sasl_plain_password='pass') transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) conn.transport = transport conn.initializing = True @@ -550,7 +550,7 @@ def test_sasl_authenticate_auth_failure(self, net): sasl_plain_username='user', sasl_plain_password='wrong') transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) conn.transport = transport conn.initializing = True @@ -620,7 +620,7 @@ def test_sasl_uses_transport_host_for_mechanism(self, net): sasl_plain_username='user', sasl_plain_password='pass') transport = MagicMock() transport.host = 'kafka.example.com' - transport.getPeer.return_value = ('10.0.0.1', 9092) + transport.get_peer.return_value = ('10.0.0.1', 9092) conn.transport = transport conn.initializing = True @@ -634,7 +634,7 @@ def test_sasl_falls_back_to_peer_ip_when_transport_host_unset(self, net): sasl_plain_username='user', sasl_plain_password='pass') transport = MagicMock() transport.host = None - transport.getPeer.return_value = ('10.0.0.1', 9092) + transport.get_peer.return_value = ('10.0.0.1', 9092) conn.transport = transport conn.initializing = True diff --git a/test/net/test_backend_future.py b/test/net/test_net_backend_future.py similarity index 88% rename from test/net/test_backend_future.py rename to test/net/test_net_backend_future.py index b38e16e05..faa2dc463 100644 --- a/test/net/test_backend_future.py +++ b/test/net/test_net_backend_future.py @@ -1,6 +1,6 @@ -"""Conformance suite for the BackendFuture contract (kafka/future.py). +"""Conformance suite for the NetBackendFuture contract (kafka/future.py). -``BackendFutureContract`` is a reusable, backend-agnostic test mixin: it +``NetBackendFutureContract`` is a reusable, backend-agnostic test mixin: it pins the contract that every backend's ``create_future()`` must satisfy (callback core, fan-out, await semantics). A backend provides two hooks -- ``make_future()`` (a fresh pending future) and ``drive(coros)`` (run the @@ -11,11 +11,11 @@ import pytest from kafka.future import Future -from kafka.net.backend import BackendFuture +from kafka.net.backend import NetBackendFuture from kafka.net.selector import NetworkSelector -class BackendFutureContract: +class NetBackendFutureContract: # --- hooks a backend must provide ------------------------------------- def make_future(self): raise NotImplementedError @@ -26,7 +26,7 @@ def drive(self, coros): # --- typing / identity ------------------------------------------------ def test_satisfies_protocol(self): - assert isinstance(self.make_future(), BackendFuture) + assert isinstance(self.make_future(), NetBackendFuture) # --- callback core (no loop needed) ----------------------------------- def test_success_fires_callback(self): @@ -147,7 +147,7 @@ async def resolver(): assert caught == ['nope'] -class TestNetworkSelectorBackendFuture(BackendFutureContract): +class TestNetworkSelectorNetBackendFuture(NetBackendFutureContract): """The selector backend: create_future() returns a SelectorFuture, driven synchronously via NetworkSelector.drain() (no IO thread needed).""" @@ -168,16 +168,16 @@ def drive(self, coros): self.net.drain() -class TestBackendFutureTypeEnforcement: - """A backend's create_future() result satisfies BackendFuture -- covered by - TestNetworkSelectorBackendFuture.test_satisfies_protocol above. Here we pin +class TestNetBackendFutureTypeEnforcement: + """A backend's create_future() result satisfies NetBackendFuture -- covered by + TestNetworkSelectorNetBackendFuture.test_satisfies_protocol above. Here we pin the negative: the plain thread-safe Future (a cross-thread handoff) deliberately does NOT (no __await__), which turns 'awaited a handoff future' into a loud error rather than a silent backend-specific bug.""" def test_plain_future_is_not_backend_future(self): - assert not isinstance(Future(), BackendFuture) + assert not isinstance(Future(), NetBackendFuture) assert not hasattr(Future(), '__await__') def test_non_future_is_not_backend_future(self): - assert not isinstance(object(), BackendFuture) + assert not isinstance(object(), NetBackendFuture) diff --git a/test/net/test_sasl_reauthentication.py b/test/net/test_sasl_reauthentication.py index 76745c23a..e0c6f4615 100644 --- a/test/net/test_sasl_reauthentication.py +++ b/test/net/test_sasl_reauthentication.py @@ -191,7 +191,7 @@ def test_session_lifetime_captured_from_v1_response(self, net): conn = KafkaConnection(net, node_id='test', **SASL_CONFIG) transport = MagicMock() - transport.getPeer.return_value = ('127.0.0.1', 9092) + transport.get_peer.return_value = ('127.0.0.1', 9092) transport.host = 'broker' conn.transport = transport conn.initializing = True diff --git a/test/net/test_transport.py b/test/net/test_transport.py index 36af1abc0..849eacfc6 100644 --- a/test/net/test_transport.py +++ b/test/net/test_transport.py @@ -107,12 +107,6 @@ def test_write_after_close_raises(self, net): with pytest.raises(RuntimeError): t.write(b'hello') - def test_writelines(self, net): - sock = _make_mock_sock() - t = KafkaTCPTransport(net, sock) - t.writelines([b'hello', b'world']) - assert len(t._write_buffer) == 2 - def test_close_marks_closed(self, net): sock = _make_mock_sock() t = KafkaTCPTransport(net, sock) @@ -223,17 +217,6 @@ def fake_sock_recv(): assert t._sock is None proto.connection_lost.assert_called_once_with(err) - def test_can_write_eof(self, net): - sock = _make_mock_sock() - t = KafkaTCPTransport(net, sock) - assert t.can_write_eof() - - def test_write_eof_sets_flag(self, net): - sock = _make_mock_sock() - t = KafkaTCPTransport(net, sock) - t.write_eof() - assert not t._write - def test_end_to_end_write_read(self, net, socketpair): rsock, wsock = socketpair t = KafkaTCPTransport(net, wsock) @@ -250,50 +233,6 @@ async def reader(): net.poll(timeout_ms=1000, future=f) assert received == [b'hello world'] - def test_write_eof_empty_buffer_shuts_down_immediately(self, net): - # Fast path: nothing buffered, so write_eof half-closes right away. - sock = _make_mock_sock() - t = KafkaTCPTransport(net, sock) - t.write_eof() - sock.shutdown.assert_called_once_with(socket.SHUT_WR) - - def test_write_eof_defers_shutdown_until_buffer_flushed(self, net): - # With data still buffered, write_eof must NOT shut down the write side - # yet -- doing so would discard the unflushed bytes (the latent bug). - sock = _make_mock_sock() - t = KafkaTCPTransport(net, sock) - t.write(b'pending') - t.write_eof() - assert not t._write - assert t._write_buffer # still buffered - sock.shutdown.assert_not_called() - - def test_write_eof_flushes_buffered_data_then_shuts_write(self, net, socketpair): - # Regression: data buffered before write_eof() must be fully delivered - # to the peer, and only then is the write side shut down (peer sees EOF). - rsock, wsock = socketpair - t = KafkaTCPTransport(net, wsock) - t.write(b'hello world') # buffered; _write_to_sock scheduled - t.write_eof() # half-close requested -- flush must win - assert t._write_buffer - - f = Future() - received = [] - async def reader(): - while True: - await net.wait_read(rsock) - data = rsock.recv(1024) - received.append(data) - if data == b'': # EOF => peer shut down its write side - break - f.success(True) - net.call_soon(reader) - net.poll(timeout_ms=1000, future=f) - - assert b''.join(received) == b'hello world' - assert received[-1] == b'' # shutdown(SHUT_WR) happened after the flush - assert not t._write_buffer - def test_close_empty_buffer_closes_immediately(self, net): # Fast path: nothing buffered, so close() tears down synchronously. sock = _make_mock_sock() @@ -347,12 +286,6 @@ async def reader(): assert t._sock is None # _write_to_sock flushed then _close()d proto.connection_lost.assert_called_once_with(None) - def test_writeSequence(self, net): - sock = _make_mock_sock() - t = KafkaTCPTransport(net, sock) - t.writeSequence([b'a', b'b']) - assert len(t._write_buffer) == 2 - def test_str_with_tcp_socket(self, net): sock = _make_mock_sock() t = KafkaTCPTransport(net, sock) diff --git a/test/test_future.py b/test/test_future.py index d87e4c906..b8cbe3c86 100644 --- a/test/test_future.py +++ b/test/test_future.py @@ -39,7 +39,7 @@ class TestFutureNotAwaitable: __await__ lives on the backend future (e.g. SelectorFuture from create_future()), not the base, so awaiting a handoff Future fails loudly. The backend futures' await behavior is covered by test_selector.py and the - BackendFuture conformance suite.""" + NetBackendFuture conformance suite.""" def test_no_dunder_await(self): assert not hasattr(Future(), '__await__')