diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 58cba7f2c..7f4c57b47 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -1,11 +1,52 @@ -"""Pluggable async-backend contracts. +"""The pluggable async-backend contract. -For now this holds the :class:`BackendFuture` contract -- the surface of the +``NetBackend`` is the interface the rest of kafka-python depends on for its +event loop: the surface that ``KafkaProducer`` / ``KafkaConsumer`` / +``KafkaAdminClient`` (and the manager, cluster, connection, coordinator, +fetcher, sender) reach for through ``self._net`` / ``manager._net``. + +``NetworkSelector`` (``kafka/net/selector.py``) is the reference +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 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. + +Networking is a **connection seam**, not fd-readiness. asyncio and Twisted own +the socket (DNS, connect, TLS, buffering) and drive protocol callbacks; they do +not expose portable fd-readiness (asyncio's Proactor loop has no ``add_reader``, +Twisted never exposes arbitrary-fd readiness). So the backend provides +``create_connection`` -- given a ``KafkaConnection`` (which already implements +the ``asyncio.Protocol`` surface) and an endpoint, it establishes the transport +and wires the two together. The returned object satisfies the small +:class:`Transport` protocol (the subset ``KafkaConnection`` drives). + +Three things are intentionally **not** part of the contract: + +* ``wait_read`` / ``wait_write`` / ``unregister_event`` -- the low-level + fd-readiness primitives. They are the *selector's* private mechanism (used + only inside ``kafka/net/transport.py`` + ``inet.py``, zero core callers) and + do not port to asyncio/Twisted. The connection seam replaces them. +* ``poll(timeout_ms, future=...)`` -- the legacy single-tick driver. Its only + remaining caller is the ``KafkaNetClient`` compat shim + (``kafka/net/compat.py``), which is selector-bound; asyncio has no bounded + single-tick equivalent. +* ``register_event`` / selector internals -- backend-private plumbing. + +Method families: + +* **Lifecycle** -- ``start`` / ``stop`` / ``close`` / ``on_io_thread``. +* **Scheduling** -- ``call_soon`` / ``call_soon_threadsafe`` / + ``call_soon_with_future`` / ``call_at`` / ``call_later`` / ``cancel``. +* **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``). +* **Cross-thread wake** -- ``wakeup``. """ -from typing import Any, Callable, Protocol, runtime_checkable +from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable @runtime_checkable @@ -59,3 +100,106 @@ def add_both(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': . def chain(self, future: 'BackendFuture') -> 'BackendFuture': ... def succeeded(self) -> bool: ... def failed(self) -> bool: ... + + +@runtime_checkable +class Transport(Protocol): + """The transport surface a backend's ``create_connection`` returns. + + The subset of the ``asyncio.Transport`` / Twisted ``ITransport`` surface + that ``KafkaConnection`` actually drives. The selector's + ``KafkaTCPTransport`` and an asyncio-transport adapter both satisfy it. + """ + + def write(self, data: bytes) -> None: ... + def close(self) -> None: ... + def abort(self, error: Any = None) -> None: ... + def is_closing(self) -> bool: ... + def pause_reading(self) -> None: ... + def resume_reading(self) -> None: ... + def host_port(self) -> Tuple[str, int]: ... + + +@runtime_checkable +class NetBackend(Protocol): + """Structural contract for a pluggable async event-loop backend. + + ``runtime_checkable`` so conformance can be asserted with ``isinstance``; + note that only checks member *presence*, not signatures. ``NetworkSelector`` + satisfies this structurally (no explicit inheritance needed). + """ + + # --- lifecycle -------------------------------------------------------- + def start(self) -> None: + """Spawn/attach the IO thread that runs the loop. Idempotent.""" + + def stop(self, timeout_ms: Optional[float] = None) -> None: + """Stop the loop and join the IO thread. Idempotent.""" + + def close(self) -> None: + """Stop (if running) and release loop resources. Idempotent.""" + + def on_io_thread(self) -> bool: + """True if the caller is running on this backend's IO thread.""" + + # --- scheduling ------------------------------------------------------- + def call_soon(self, task: Any) -> Any: + """Enqueue a coroutine/callable to run on the next loop iteration.""" + + 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: + """Schedule ``coro`` and return a future that resolves with its result.""" + + def call_at(self, when: float, task: Any) -> Any: + """Schedule ``task`` to run at absolute monotonic time ``when``.""" + + def call_later(self, delay: float, task: Any) -> Any: + """Schedule ``task`` to run after ``delay`` seconds.""" + + def cancel(self, task: Any) -> None: + """Cancel a scheduled task/timer previously returned by call_*.""" + + # --- timing (core coroutines await this) ------------------------------ + def sleep(self, delay: float) -> Any: + """Awaitable that resolves after ``delay`` seconds.""" + + # --- connection seam -------------------------------------------------- + async def create_connection( + self, + protocol: Any, + host: str, + port: int, + *, + ssl: Any = None, + proxy_url: Optional[str] = None, + socket_options: Sequence[Any] = (), + timeout_at: Optional[float] = None, + ) -> Transport: + """Establish and return a connected :class:`Transport` to ``host:port``. + + The backend owns DNS, connect, TLS and (where supported) proxying. The + *caller* wires the transport to the protocol afterwards via + ``protocol.connection_made(transport)`` (so manager-level policy such as + the "closed during connect" check runs first). ``protocol`` (a + ``KafkaConnection``) is passed because backends that own the socket + (asyncio/Twisted) need it at connect time to receive transport events; + they buffer inbound data until ``connection_made`` is called. Backends + without native proxy support raise when ``proxy_url`` is set. + """ + + # --- cross-thread bridge --------------------------------------------- + def run(self, coro: Any, *args: Any) -> Any: + """Schedule ``coro`` on the loop, block the calling thread, return/raise. + + Raises ``RuntimeError`` if called from the IO thread itself. + """ + + # --- future factory --------------------------------------------------- + def create_future(self) -> BackendFuture: + """Create a loop-awaitable future (see ``BackendFuture``).""" + + # --- misc ------------------------------------------------------------- + def wakeup(self) -> None: + """Interrupt the loop's select() from another thread.""" diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 55501a2ea..1e1d2d44a 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -5,12 +5,11 @@ import socket import time -from .inet import create_connection from .connection import KafkaConnection from .metrics import KafkaManagerMetrics -from .transport import KafkaSSLTransport, KafkaTCPTransport from kafka.cluster import ClusterMetadata import kafka.errors as Errors +from kafka.net.transport import KafkaSSLTransport from kafka.net.wakeup_notifier import WakeupNotifier from kafka.protocol.broker_version_data import BrokerVersionData from kafka.version import __version__ @@ -85,6 +84,7 @@ def __init__(self, net, **configs): self.cluster.attach(self) self._conns = {} self._backoff = dict() # node_id => (failures, backoff_until, socket_connect_setup_timeout_ms) + self.ssl_context = KafkaSSLTransport.build_ssl_context(self.config) if self.ssl_enabled else None # Cache the most recent SASL / SSL / auth failure per node so we can # surface it to the user instead of silently retrying forever. # Cleared on successful connect. @@ -203,33 +203,18 @@ def close_idle_connections(self): def ssl_enabled(self): return self.config['security_protocol'] in ('SSL', 'SASL_SSL') - async def _build_transport(self, node, timeout_at=None): - sock = await create_connection(self._net, node.host, node.port, - self.config['socket_options'], - proxy_url=self.config['proxy_url'], - timeout_at=timeout_at) - if self.ssl_enabled: - ssl_configs = {key: value - for key, value in self.config.items() - if key.startswith('ssl_')} - transport = KafkaSSLTransport(self._net, sock, host=node.host, **ssl_configs) - else: - transport = KafkaTCPTransport(self._net, sock, host=node.host) - - try: - await transport.handshake() - except Exception as e: - raise Errors.KafkaConnectionError('Handshake failed: %s' % e) - else: - return transport - async def _connect(self, node, conn, reset_backoff_on_connect=True, timeout_at=None): # Tracks ownership of the freshly built transport: while non-None it is # ours to clean up (the connection hasn't taken it over yet), so the # finally clause closes it. Cleared once connection_made() succeeds. transport = None try: - transport = await self._build_transport(node, timeout_at=timeout_at) + transport = await self._net.create_connection( + conn, node.host, node.port, + ssl=self.ssl_context, + proxy_url=self.config['proxy_url'], + socket_options=self.config['socket_options'], + timeout_at=timeout_at) # The connection (or the whole manager) may have been closed while # we were building the transport. Handing it to connection_made() # would flip the conn back to `initializing` and resurrect a diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 7726a37b9..7f11daa4f 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -11,6 +11,8 @@ import kafka.errors as Errors from kafka.future import Future +from kafka.net.inet import create_connection as _inet_create_connection +from kafka.net.transport import KafkaSSLTransport, KafkaTCPTransport from kafka.version import __version__ @@ -277,6 +279,16 @@ def start(self): self._io_thread = t t.start() + def on_io_thread(self): + """True if the caller is running on this backend's IO thread. + + The clean form of the ``current_thread() is _io_thread`` identity + check; callers use it to avoid blocking the loop on itself (e.g. a + producer ``close()`` invoked from a produce callback). Part of the + NetBackend contract so alternate backends can answer it their own way. + """ + return self._io_thread is not None and threading.current_thread() is self._io_thread + def stop(self, timeout_ms=None): """Signal run_forever() to exit and join the IO thread. @@ -475,6 +487,31 @@ def create_future(self): """ return SelectorFuture() + async def create_connection(self, protocol, host, port, *, ssl=None, + proxy_url=None, socket_options=(), + timeout_at=None): + """Establish and return a connected transport to host:port. + + The selector owns the raw socket: DNS + non-blocking connect (with + optional SOCKS5/HTTP-CONNECT proxy via KafkaNetSocket), then wraps it + in a TCP or SSL transport and runs the TLS handshake. ``protocol`` (the + KafkaConnection) is not used here -- the caller wires it via + ``connection_made()`` after its own "closed during connect" check; it + is part of the contract because socket-owning backends (asyncio, + Twisted) need it at connect time. + """ + sock = await _inet_create_connection(self, host, port, socket_options, + proxy_url=proxy_url, timeout_at=timeout_at) + if ssl is not None: + transport = KafkaSSLTransport(self, sock, ssl, host=host) + else: + transport = KafkaTCPTransport(self, sock, host=host) + try: + await transport.handshake() + except Exception as e: + raise Errors.KafkaConnectionError('Handshake failed: %s' % e) + return transport + def sleep(self, delay): return KernelEvent('_sleep', delay) diff --git a/kafka/net/transport.py b/kafka/net/transport.py index 5a5365753..f44d0ff20 100644 --- a/kafka/net/transport.py +++ b/kafka/net/transport.py @@ -381,20 +381,21 @@ class KafkaSSLTransport(KafkaTCPTransport): 'ssl_password': None, 'ssl_crlfile': None, } - def __init__(self, net, sock, host=None, **configs): - self.ssl_config = copy.copy(self.DEFAULT_CONFIG) - for key in self.ssl_config: - if key in configs: - self.ssl_config[key] = configs[key] - self._ssl_context = self._build_ssl_context(self.ssl_config) + def __init__(self, net, sock, ssl_context, host=None): + self._ssl_context = ssl_context server_hostname = host.rstrip('.') if host is not None else None sock = self._ssl_context.wrap_socket( sock, server_hostname=server_hostname, do_handshake_on_connect=False) super().__init__(net, sock, host=host) - @staticmethod - def _build_ssl_context(config): + @classmethod + def build_ssl_context(cls, configs): + config = copy.copy(cls.DEFAULT_CONFIG) + for key in config: + if key in configs: + config[key] = configs[key] + if config['ssl_context'] is not None: return config['ssl_context'] ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 20df36f00..2552b8246 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -695,8 +695,7 @@ def __getattr__(self, name): log.info("%s: Closing the Kafka producer with %s secs timeout.", str(self), timeout) self.flush(timeout) - on_io_thread = bool(self._net._io_thread is not None - and threading.current_thread() is self._net._io_thread) + on_io_thread = self._net.on_io_thread() if timeout > 0: if on_io_thread: log.warning("%s: Overriding close timeout %s secs to 0 in order to" diff --git a/test/mock_broker.py b/test/mock_broker.py index cd61d355d..7c5b7f3d2 100644 --- a/test/mock_broker.py +++ b/test/mock_broker.py @@ -351,7 +351,7 @@ def stop(self, error=None): """Take the broker down, as if the process was killed. Aborts all live transports with ``error`` and refuses new connections - (an attached manager's ``_build_transport`` raises + (an attached manager's ``net.create_connection`` raises ``KafkaConnectionError``) until :meth:`start` is called. This exercises the client's real connection-lost, connect-failure, and reconnect-backoff paths. @@ -501,16 +501,16 @@ def attach(self, manager): """ broker = self - async def _mock_build_transport(node, timeout_at=None): + async def _mock_create_connection(protocol, host, port, **kwargs): if not broker.online: raise Errors.KafkaConnectionError( 'connect to %s:%s refused (MockBroker stopped)' - % (node.host, node.port)) + % (host, port)) return MockTransport( manager._net, broker, - node_id=node.node_id, host=node.host, port=node.port) + node_id=protocol.node_id, host=host, port=port) - manager._build_transport = _mock_build_transport + manager._net.create_connection = _mock_create_connection def client_factory(self): """Return a callable suitable for passing as ``kafka_client=...`` @@ -699,16 +699,16 @@ def attach(self, manager): """ cluster = self - async def _mock_build_transport(node, timeout_at=None): - broker = cluster._by_addr.get((node.host, node.port)) + async def _mock_create_connection(protocol, host, port, **kwargs): + broker = cluster._by_addr.get((host, port)) if broker is None or not broker.online: raise Errors.KafkaConnectionError( - 'connect to %s:%s refused' % (node.host, node.port)) + 'connect to %s:%s refused' % (host, port)) return MockTransport( manager._net, broker, - node_id=node.node_id, host=node.host, port=node.port) + node_id=protocol.node_id, host=host, port=port) - manager._build_transport = _mock_build_transport + manager._net.create_connection = _mock_create_connection def client_factory(self): """Return a callable suitable for passing as ``kafka_client=...`` to diff --git a/test/net/test_backend.py b/test/net/test_backend.py new file mode 100644 index 000000000..2a2d50452 --- /dev/null +++ b/test/net/test_backend.py @@ -0,0 +1,84 @@ +"""Conformance tests for the NetBackend contract (kafka/net/backend.py). + +NetworkSelector is the reference implementation; these pin that it satisfies +the NetBackend Protocol structurally and that the shared lifecycle helper +``on_io_thread()`` behaves correctly. Step 4's AsyncioBackend will be held to +the same isinstance/method-presence checks. +""" +import threading + +from kafka.net.backend import NetBackend, Transport +from kafka.net.selector import NetworkSelector +from kafka.net.transport import KafkaTCPTransport + + +# The full contract surface, kept here so a missing/renamed method fails loudly. +CONTRACT_METHODS = ( + 'start', 'stop', 'close', 'on_io_thread', + 'call_soon', 'call_soon_threadsafe', 'call_soon_with_future', + 'call_at', 'call_later', 'cancel', + 'sleep', 'create_connection', + 'run', 'create_future', 'wakeup', +) + +# Removed from the contract by the connection-seam revision (selector-private). +NON_CONTRACT_METHODS = ('wait_read', 'wait_write', 'unregister_event', 'poll') + + +class TestNetBackendContract: + def test_networkselector_satisfies_protocol(self): + assert isinstance(NetworkSelector(), NetBackend) + + def test_plain_object_is_not_netbackend(self): + assert not isinstance(object(), NetBackend) + + def test_partial_impl_is_not_netbackend(self): + class Partial: + def start(self): + pass + # missing everything else + assert not isinstance(Partial(), NetBackend) + + def test_all_contract_methods_present_and_callable(self): + net = NetworkSelector() + for name in CONTRACT_METHODS: + assert callable(getattr(net, name)), name + + def test_readiness_primitives_and_poll_excluded(self): + # wait_read/wait_write/unregister_event (selector-private, replaced by + # the connection seam) and legacy poll() are intentionally NOT in the + # contract, though they still exist on NetworkSelector. + net = NetworkSelector() + for name in NON_CONTRACT_METHODS: + assert name not in CONTRACT_METHODS + assert hasattr(net, name), name # still present on the selector impl + + +class TestTransportContract: + def test_kafkatcptransport_satisfies_transport(self): + # Method-presence check against the Transport 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) + + +class TestOnIoThread: + def test_false_before_start(self): + net = NetworkSelector() + assert net.on_io_thread() is False + + def test_false_from_other_thread_true_on_loop(self): + net = NetworkSelector() + net.start() + try: + async def where(): + return net.on_io_thread() + # Runs on the IO thread -> True; the calling test thread -> False. + assert net.run(where) is True + assert net.on_io_thread() is False + assert threading.current_thread() is not net._io_thread + finally: + net.close() diff --git a/test/net/test_manager.py b/test/net/test_manager.py index 036b6fa6a..fcc45f041 100644 --- a/test/net/test_manager.py +++ b/test/net/test_manager.py @@ -86,23 +86,23 @@ def test_proxy_url_takes_precedence_over_legacy(self, net): ) assert m.config['proxy_url'] == 'socks5://new:1080' - def test_build_transport_passes_proxy_url(self, net): - """_build_transport must forward the configured proxy_url to - create_connection. Regression guard against the kwarg name drifting - from the create_connection signature.""" - import asyncio + def test_connect_passes_proxy_url(self, net): + """_connect must forward the configured proxy_url to + net.create_connection. Regression guard against the kwarg name + drifting from the create_connection signature.""" m = KafkaConnectionManager(net, proxy_url='socks5://proxy:1080') - node = MagicMock(host='broker', port=9092) - async def fake_create_connection(*args, **kwargs): + node = MagicMock(host='broker', port=9092, node_id='bootstrap-0') + conn = KafkaConnection(net, node_id='bootstrap-0', **m.config) + + async def fake_create_connection(protocol, host, port, **kwargs): + # Close mid-connect so _connect short-circuits before + # connection_made()/initialize() -- we only assert the forwarded kwarg. + conn.close() return MagicMock() - with patch('kafka.net.manager.create_connection', - side_effect=fake_create_connection) as mc, \ - patch('kafka.net.manager.KafkaTCPTransport') as mock_transport_cls: - mock_transport = MagicMock() - mock_transport.handshake = MagicMock( - side_effect=lambda: asyncio.sleep(0)) - mock_transport_cls.return_value = mock_transport - net.run(m._build_transport(node)) + + with patch.object(net, 'create_connection', + side_effect=fake_create_connection) as mc: + net.run(m._connect(node, conn)) assert mc.call_args.kwargs.get('proxy_url') == 'socks5://proxy:1080' @@ -412,8 +412,8 @@ def test_close_no_connections(self, manager): class TestKafkaConnectionManagerConnectRace: """A connection can be closed (by manager.close() / bootstrap teardown) - while its _connect() coroutine is still awaiting _build_transport. When - the transport finally arrives, _connect must not resurrect the dead + while its _connect() coroutine is still awaiting net.create_connection. + When the transport finally arrives, _connect must not resurrect the dead connection via connection_made() -- doing so flips it back to `initializing`.""" @@ -423,13 +423,13 @@ def test_connect_discards_transport_when_closed_during_build(self, net): conn = KafkaConnection(net, node_id='bootstrap-0', **manager.config) transport = MagicMock() - async def fake_build_transport(n, timeout_at=None): + async def fake_create_connection(protocol, host, port, **kwargs): # Simulate a concurrent close landing mid-connect. conn.close() return transport - with patch.object(manager, '_build_transport', - side_effect=fake_build_transport): + with patch.object(net, 'create_connection', + side_effect=fake_create_connection): net.run(manager._connect(node, conn)) # Dead connection stays dead -- not resurrected to `initializing`. diff --git a/test/net/test_transport.py b/test/net/test_transport.py index d1ed0ba90..36af1abc0 100644 --- a/test/net/test_transport.py +++ b/test/net/test_transport.py @@ -1,4 +1,5 @@ import socket +import ssl import time from unittest.mock import MagicMock @@ -385,72 +386,75 @@ def _make_ssl_sock(self): def test_sni_sent_when_check_hostname_true(self, net): sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com', - ssl_context=ctx, ssl_check_hostname=True) + ctx.check_hostname = True + KafkaSSLTransport(net, sock, ctx, host='broker.example.com') _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] == 'broker.example.com' def test_sni_sent_when_check_hostname_false(self, net): # The bug: SNI used to be suppressed when verification was disabled. sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com', - ssl_context=ctx, ssl_check_hostname=False) + ctx.check_hostname = False + KafkaSSLTransport(net, sock, ctx, host='broker.example.com') _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] == 'broker.example.com' def test_sni_strips_trailing_dot(self, net): # A trailing dot is a valid FQDN but illegal in the SNI extension. sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com.', - ssl_context=ctx, ssl_check_hostname=False) + ctx.check_hostname = False + KafkaSSLTransport(net, sock, ctx, host='broker.example.com.') _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] == 'broker.example.com' def test_sni_none_when_host_missing(self, net): sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host=None, ssl_context=ctx) + KafkaSSLTransport(net, sock, ctx, host=None) _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] is None def test_handshake_not_done_on_connect(self, net): sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com', ssl_context=ctx) + KafkaSSLTransport(net, sock, ctx, host='broker.example.com') _, kwargs = ctx.wrap_socket.call_args assert kwargs['do_handshake_on_connect'] is False def test_provided_ssl_context_is_used(self, net): sock, ctx, wrapped = self._make_ssl_sock() - t = KafkaSSLTransport(net, sock, host='broker.example.com', - ssl_context=ctx) + t = KafkaSSLTransport(net, sock, ctx, host='broker.example.com') assert t._ssl_context is ctx assert t._sock is wrapped - def test_config_defaults_and_overrides(self, net): - sock, ctx, _ = self._make_ssl_sock() - t = KafkaSSLTransport(net, sock, host='h', ssl_context=ctx, - ssl_check_hostname=False) - # Explicitly-passed ssl_* keys land in ssl_config... - assert t.ssl_config['ssl_check_hostname'] is False - assert t.ssl_config['ssl_context'] is ctx - # ...unspecified keys keep their defaults. - assert t.ssl_config['ssl_cafile'] is None - assert t.ssl_config['ssl_check_hostname'] is not None + def test_ssl_context_is_required(self, net): + # The transport no longer builds a context itself; callers must pass + # a pre-built one (via build_ssl_context). Omitting it is a TypeError, + # not a silently-default context. + sock, _, _ = self._make_ssl_sock() + with pytest.raises(TypeError): + KafkaSSLTransport(net, sock, host='broker.example.com') # pylint: disable=E1120 class TestBuildSSLContext: def test_returns_provided_context(self): ctx = MagicMock() - config = dict(KafkaSSLTransport.DEFAULT_CONFIG, ssl_context=ctx) - assert KafkaSSLTransport._build_ssl_context(config) is ctx + config = dict(ssl_context=ctx) + assert KafkaSSLTransport.build_ssl_context(config) is ctx def test_check_hostname_propagates_to_context(self): - config = dict(KafkaSSLTransport.DEFAULT_CONFIG, ssl_check_hostname=False) - ctx = KafkaSSLTransport._build_ssl_context(config) + config = dict(ssl_check_hostname=False) + ctx = KafkaSSLTransport.build_ssl_context(config) assert ctx.check_hostname is False def test_check_hostname_true_requires_verification(self): - config = dict(KafkaSSLTransport.DEFAULT_CONFIG, ssl_check_hostname=True) - ctx = KafkaSSLTransport._build_ssl_context(config) + config = dict(ssl_check_hostname=True) + ctx = KafkaSSLTransport.build_ssl_context(config) + assert ctx.check_hostname is True + + def test_empty_config_builds_default_context(self): + # Missing keys fall back to DEFAULT_CONFIG: a real TLS-client context + # with hostname checking on. This is the default-build path. + ctx = KafkaSSLTransport.build_ssl_context({}) + assert isinstance(ctx, ssl.SSLContext) assert ctx.check_hostname is True