diff --git a/bin/clone-dependencies.sh b/bin/clone-dependencies.sh index 10260a8..1949ecb 100755 --- a/bin/clone-dependencies.sh +++ b/bin/clone-dependencies.sh @@ -15,7 +15,7 @@ set -u BDE_TOOLS_TAG=4.38.0.0 BDE_TAG=4.38.0.0 NTF_CORE_TAG=2.6.12 -BLAZINGMQ_TAG=v0.95.14 +BLAZINGMQ_TAG=v0.95.20 if [ ! -d "${DIR_THIRDPARTY}/bde-tools" ]; then diff --git a/docs/api_reference.rst b/docs/api_reference.rst index b380ab8..438ff85 100644 --- a/docs/api_reference.rst +++ b/docs/api_reference.rst @@ -187,6 +187,8 @@ Testing Utilities Helper Types ============ +.. autoclass:: blazingmq.AuthnCredentialProvider + .. autoclass:: blazingmq.PropertyTypeDict .. autoclass:: blazingmq.PropertyValueDict diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 910d1ad..e630190 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -27,6 +27,7 @@ from ._session import Session from ._session import SessionOptions from ._timeouts import Timeouts +from ._typing import AuthnCredentialProvider from ._typing import PropertyTypeDict from ._typing import PropertyValueDict from .exceptions import Error @@ -34,6 +35,7 @@ __all__ = [ "Ack", "AckStatus", + "AuthnCredentialProvider", "BasicHealthMonitor", "CompressionAlgorithmType", "Error", diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index b455d49..a035f66 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -37,6 +37,9 @@ class FakeHostHealthMonitor: def set_healthy(self) -> None: ... def set_unhealthy(self) -> None: ... +class AuthnCredentialCbAdapter: + def __init__(self, callback: Callable[[], Optional[tuple[str, bytes]]]) -> None: ... + class Session: def __init__( self, @@ -53,6 +56,7 @@ class Session: timeouts: Timeouts = Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: Optional[FakeHostHealthMonitor] = None, + authn_credential_cb: Optional[AuthnCredentialCbAdapter] = None, user_agent_prefix: bytes = b"", ) -> None: ... def stop(self) -> None: ... diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 4d07e74..e0b2be7 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -154,6 +154,37 @@ cdef class FakeHostHealthMonitor: self._monitor.get().setState(HostHealthState.e_UNHEALTHY) +cdef class AuthnCredentialCbAdapter: + cdef object _callback + + def __cinit__(self, callback): + self._callback = callback + + def get_credential_data(self): + """Call the provider and marshal its result for the C++ session. + + Called by ``pybmq::AuthnCredentialCbFunctor``. Returns the mechanism + and data as a tuple of ``bytes``, or `None` if credentials could not + be obtained, in which case authentication fails. + """ + try: + result = self._callback() + if result is None: + return None + + mechanism, data = result + if not isinstance(mechanism, str) or not isinstance(data, bytes): + raise TypeError( + "authn_credential_provider must return (str, bytes) or None" + ) + + return mechanism.encode('utf-8'), data + + except Exception: + LOGGER.exception("Error in authentication credential callback") + return None + + cdef class Session: cdef object __weakref__ cdef NativeSession* _session @@ -175,6 +206,7 @@ cdef class Session: timeouts: _timeouts.Timeouts = _timeouts.Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, + authn_credential_cb: AuthnCredentialCbAdapter = None, _mock: Optional[object] = None, user_agent_prefix: bytes = b"", ) -> None: @@ -249,6 +281,7 @@ cdef class Session: session_cb, message_cb, ack_cb, + authn_credential_cb, config, fake_host_health_monitor_sp, Error, diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index c7e31f9..16afbb0 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -27,6 +27,7 @@ from ._about import __version__ from ._enums import CompressionAlgorithmType from ._enums import PropertyType +from ._ext import AuthnCredentialCbAdapter from ._ext import DEFAULT_CONSUMER_PRIORITY from ._ext import DEFAULT_MAX_UNCONFIRMED_BYTES from ._ext import DEFAULT_MAX_UNCONFIRMED_MESSAGES @@ -38,6 +39,7 @@ from ._messages import MessageHandle from ._monitors import BasicHealthMonitor from ._timeouts import Timeouts +from ._typing import AuthnCredentialProvider from ._typing import PropertyTypeDict from ._typing import PropertyValueDict from ._typing import PropertyValueType @@ -54,6 +56,10 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]: return None +def DefaultAuthnCredentialProvider() -> Optional[AuthnCredentialProvider]: + return None + + DEFAULT_TIMEOUT = DefaultTimeoutType() KNOWN_MONITORS = ("blazingmq.BasicHealthMonitor",) @@ -331,6 +337,15 @@ class SessionOptions: 96 bytes long. This is provided for libraries that are wrapping this SDK. Applications directly using the SDK are encouraged *NOT* to set this value. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + An optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``. It is called + each time the session authenticates with the broker, including on + reauthentication. If it returns ``None`` or raises, the + connection is closed: starting a session fails, while an + already-started session sees `.ConnectionLost` and then + reconnects, calling this callable again. If not provided, no + authentication credentials are sent to the broker. """ def __init__( @@ -344,6 +359,9 @@ def __init__( event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, user_agent_prefix: Optional[bytes] = None, + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() + ), ) -> None: self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts @@ -354,6 +372,7 @@ def __init__( self.event_queue_watermarks = event_queue_watermarks self.stats_dump_interval = stats_dump_interval self.user_agent_prefix = user_agent_prefix + self.authn_credential_provider = authn_credential_provider def __eq__(self, other: object) -> bool: if not isinstance(other, SessionOptions): @@ -368,6 +387,7 @@ def __eq__(self, other: object) -> bool: and self.event_queue_watermarks == other.event_queue_watermarks and self.stats_dump_interval == other.stats_dump_interval and self.user_agent_prefix == other.user_agent_prefix + and self.authn_credential_provider == other.authn_credential_provider ) def __ne__(self, other: object) -> bool: @@ -384,6 +404,7 @@ def __repr__(self) -> str: "event_queue_watermarks", "stats_dump_interval", "user_agent_prefix", + "authn_credential_provider", ) params = [] @@ -451,6 +472,15 @@ class Session: must be at most 96 bytes long. This is provided for libraries that are wrapping this SDK. Applications directly using the SDK are encouraged *NOT* to set this value. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + an optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``. It is called + each time the session authenticates with the broker, including on + reauthentication. If it returns ``None`` or raises, the + connection is closed: starting a session fails, while an + already-started session sees `.ConnectionLost` and then + reconnects, calling this callable again. If not provided, no + authentication credentials are sent to the broker. Raises: `~blazingmq.Error`: If the session start request was not successful. @@ -476,6 +506,9 @@ def __init__( event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, user_agent_prefix: Optional[bytes] = None, + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() + ), ) -> None: if host_health_monitor is not None: if not isinstance(host_health_monitor, BasicHealthMonitor): @@ -486,6 +519,11 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) + authn_credential_cb = ( + AuthnCredentialCbAdapter(authn_credential_provider) + if authn_credential_provider is not None + else None + ) self._has_no_on_message = on_message is None @@ -515,6 +553,7 @@ def __init__( timeouts=_validate_timeouts(timeout), monitor_host_health=monitor_host_health, fake_host_health_monitor=fake_host_health_monitor, + authn_credential_cb=authn_credential_cb, user_agent_prefix=_make_user_agent_prefix(user_agent_prefix), ) self._ext.set_owned_by_session() @@ -577,6 +616,7 @@ def with_options( event_queue_watermarks=session_options.event_queue_watermarks, stats_dump_interval=session_options.stats_dump_interval, user_agent_prefix=session_options.user_agent_prefix, + authn_credential_provider=session_options.authn_credential_provider, ) def open_queue( diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 39cf633..c6fe00a 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,7 +13,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +from typing import Callable from typing import Mapping +from typing import Optional from typing import Union from ._enums import PropertyType @@ -23,3 +25,10 @@ PropertyValueDict = Mapping[str, PropertyValueType] PropertyTypeDict = Mapping[str, PropertyType] + +AuthnCredentialProvider = Callable[[], Optional[tuple[str, bytes]]] +"""A callable that returns authentication credentials as a ``(mechanism, +data)`` tuple of ``(str, bytes)``, or ``None`` if an error occurs while +obtaining them, in which case authentication fails and the connection is +closed. +""" diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 7033987..5540e89 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -1,4 +1,4 @@ -// Copyright 2019-2023 Bloomberg Finance L.P. +// Copyright 2019-2026 Bloomberg Finance L.P. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -15,6 +15,7 @@ #include +#include #include #include #include @@ -25,12 +26,16 @@ #include #include #include +#include +#include #include #include +#include #include #include #include +#include #include #include #include @@ -71,12 +76,80 @@ class BrokerTimeoutError : public bsl::runtime_error } }; +// Invoke `get_credential_data()` on a Python object and convert the result to +// a `bmqt::AuthnCredential`. The object is held, not owned: the `Session` +// holds the reference and drops it only after destroying the `bmqa::Session` +// that owns every copy of this functor. +class AuthnCredentialCbFunctor +{ + private: + // DATA + PyObject* d_callback_p; + + public: + // CREATORS + explicit AuthnCredentialCbFunctor(PyObject* callback); + + // ACCESSORS + bsl::optional operator()() const; +}; + +AuthnCredentialCbFunctor::AuthnCredentialCbFunctor(PyObject* callback) +: d_callback_p(callback) +{ +} + +bsl::optional +AuthnCredentialCbFunctor::operator()() const +{ + pybmq::GilAcquireGuard guard; + + // The adapter validates the provider's result and reports any problem to + // the Python logger, so a failure here is just an empty credential. It + // hands back the mechanism and data already marshalled to `bytes`. + bslma::ManagedPtr result = RefUtils::toManagedPtr( + PyObject_CallMethod(d_callback_p, "get_credential_data", NULL)); + + if (!result) { + PyErr_WriteUnraisable(d_callback_p); + return bsl::optional(); + } + + if (result.get() == Py_None) { + return bsl::optional(); + } + + const char* mechanism_p; + Py_ssize_t mechanism_len; + const char* data_p; + Py_ssize_t data_len; + if (!PyArg_ParseTuple( + result.get(), + "y#y#", + &mechanism_p, + &mechanism_len, + &data_p, + &data_len)) + { + // The adapter broke its contract with us. + PyErr_WriteUnraisable(d_callback_p); + return bsl::optional(); + } + + bmqt::AuthnCredential credential( + bsl::string_view(mechanism_p, mechanism_len), + bsl::vector(data_p, data_p + data_len)); + return bsl::optional( + bslmf::MovableRefUtil::move(credential)); +} + } // namespace Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, @@ -88,6 +161,7 @@ Session::Session( , d_message_compression_type(bmqt::CompressionAlgorithmType::e_NONE) , d_error(error) , d_broker_timeout_error(broker_timeout_error) +, d_authn_credential_cb(NULL) , d_session_mp() { bsl::shared_ptr host_health_monitor_sp; @@ -107,6 +181,14 @@ Session::Session( } d_message_compression_type = config.message_compression_type; + + bmqt::SessionOptions::AuthnCredentialCb cpp_callback; + + if (authn_credential_cb != NULL && authn_credential_cb != Py_None) { + d_authn_credential_cb = authn_credential_cb; + cpp_callback = AuthnCredentialCbFunctor(d_authn_credential_cb); + } + { pybmq::GilReleaseGuard guard; bmqt::SessionOptions options; @@ -132,6 +214,10 @@ Session::Session( config.event_queue_watermarks.value().second); } + if (cpp_callback) { + options.setAuthnCredentialCb(cpp_callback); + } + if (config.stats_dump_interval != bsls::TimeInterval()) { options.setStatsDumpInterval(config.stats_dump_interval); } @@ -173,15 +259,22 @@ Session::Session( } Py_INCREF(d_error); Py_INCREF(d_broker_timeout_error); + Py_XINCREF(d_authn_credential_cb); } Session::~Session() { + BSLS_ASSERT(!d_started); + { + // Destroy the session first: it owns the copies of + // `AuthnCredentialCbFunctor`, which borrow `d_authn_credential_cb`. + pybmq::GilReleaseGuard gil_release_guard; + d_session_mp.reset(); + } + + Py_XDECREF(d_authn_credential_cb); Py_DECREF(d_broker_timeout_error); Py_DECREF(d_error); - BSLS_ASSERT(!d_started); - pybmq::GilReleaseGuard gil_release_guard; - d_session_mp.reset(); } PyObject* @@ -519,8 +612,8 @@ Session::post( oss << "Failed to post message to " << queue_uri << " queue: " << post_rc; throw GenericError(oss.str()); } - // We have a successful post and the SDK now owns the `on_ack` callback object - // so release our reference without a DECREF. + // We have a successful post and the SDK now owns the `on_ack` callback + // object so release our reference without a DECREF. managed_on_ack.release(); } catch (const GenericError& exc) { PyErr_SetString(d_error, exc.what()); diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 8be2b56..01e08ea 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -1,4 +1,4 @@ -// Copyright 2019-2023 Bloomberg Finance L.P. +// Copyright 2019-2026 Bloomberg Finance L.P. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -43,6 +43,7 @@ class Session bmqt::CompressionAlgorithmType::Enum d_message_compression_type; PyObject* d_error; PyObject* d_broker_timeout_error; + PyObject* d_authn_credential_cb; bslma::ManagedPtr d_session_mp; // NOT IMPLEMENTED @@ -53,6 +54,7 @@ class Session Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index 7d21f96..67bc0cb 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -58,6 +58,7 @@ cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: Session(object on_session_event, object on_message_event, object on_ack_event, + object authn_credential_cb, const SessionConfig& config, shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, diff --git a/tests/unit/test_authn_credential_cb_adapter.py b/tests/unit/test_authn_credential_cb_adapter.py new file mode 100644 index 0000000..54b2d11 --- /dev/null +++ b/tests/unit/test_authn_credential_cb_adapter.py @@ -0,0 +1,156 @@ +# Copyright 2026 Bloomberg Finance L.P. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from blazingmq._ext import AuthnCredentialCbAdapter + + +def test_valid_return(): + # GIVEN + def provider(): + return ("mechanism", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == (b"mechanism", b"data") + + +def test_mechanism_is_encoded_as_utf8(): + # GIVEN + def provider(): + return ("mécanisme", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == ("mécanisme".encode("utf-8"), b"data") + + +def test_mechanism_not_encodable(): + # GIVEN a mechanism holding a lone surrogate, which has no UTF-8 encoding + def provider(): + return ("\ud800", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_mechanism_with_embedded_nul_is_not_truncated(): + # GIVEN + def provider(): + return ("PLAIN\x00extra", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == (b"PLAIN\x00extra", b"data") + + +def test_none_return(): + # GIVEN + def provider(): + return None + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_not_a_tuple(): + # GIVEN + def provider(): + return "not a tuple" + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_tuple_wrong_length(): + # GIVEN + def provider(): + return ("mechanism", b"data", "extra") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_mechanism_not_str(): + # GIVEN + def provider(): + return (123, b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_data_not_bytes(): + # GIVEN + def provider(): + return ("mechanism", "not bytes") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_callback_raises(): + # GIVEN + def provider(): + raise RuntimeError("broken") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index be61bb7..349153b 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -78,10 +78,61 @@ def dummy2(): ), monitor_host_health=False, fake_host_health_monitor=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) +@mock.patch("blazingmq._session.ExtSession") +def test_session_constructed_with_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + # WHEN + Session( + dummy1, + on_message=dummy2, + broker="some_uri", + timeout=60.0, + host_health_monitor=None, + authn_credential_provider=my_provider, + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=Timeouts( + connect_timeout=None, + disconnect_timeout=None, + open_queue_timeout=60.0, + configure_queue_timeout=60.0, + close_queue_timeout=60.0, + ), + monitor_host_health=False, + fake_host_health_monitor=None, + authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["authn_credential_cb"] is not None + + @mock.patch("blazingmq._session.ExtSession") def test_session_constructed_with_timeouts(ext_cls): # GIVEN @@ -129,6 +180,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -174,6 +226,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -210,6 +263,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -263,6 +317,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -295,6 +350,99 @@ def dummy2(): ) +@mock.patch("blazingmq._session.ExtSession") +def test_session_default_with_options_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + session_options = SessionOptions(authn_credential_provider=my_provider) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=Timeouts(), + monitor_host_health=False, + fake_host_health_monitor=None, + authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["authn_credential_cb"] is not None + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_with_options_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + timeouts = Timeouts( + connect_timeout=60.0, + disconnect_timeout=70.0, + open_queue_timeout=80.0, + configure_queue_timeout=90.0, + close_queue_timeout=100.0, + ) + + session_options = SessionOptions( + timeouts=timeouts, + authn_credential_provider=my_provider, + ) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=timeouts, + monitor_host_health=False, + fake_host_health_monitor=None, + authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["authn_credential_cb"] is not None + + @mock.patch("blazingmq._session.ExtSession") def test_session_basic_monitor(ext_cls): # GIVEN @@ -337,6 +485,7 @@ def dummy2(): ), monitor_host_health=True, fake_host_health_monitor=monitor._monitor, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -369,6 +518,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) diff --git a/tests/unit/test_session_options.py b/tests/unit/test_session_options.py index ce14459..1c31cee 100644 --- a/tests/unit/test_session_options.py +++ b/tests/unit/test_session_options.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -60,6 +60,7 @@ def test_session_options_default_to_none(): assert options.message_compression_algorithm is None assert options.timeouts is None assert options.host_health_monitor is None + assert options.authn_credential_provider is None assert options.num_processing_threads is None assert options.blob_buffer_size is None assert options.channel_high_watermark is None @@ -95,6 +96,7 @@ def test_session_options_equality(): blazingmq.SessionOptions(channel_high_watermark=8000000), blazingmq.SessionOptions(event_queue_watermarks=(6000000, 7000000)), blazingmq.SessionOptions(stats_dump_interval=30.0), + blazingmq.SessionOptions(authn_credential_provider=lambda: None), blazingmq.SessionOptions(user_agent_prefix=b"mylib:1.0"), ], ) @@ -104,3 +106,15 @@ def test_queue_options_other_inequality(right): # THEN assert not left == right + + +def test_session_options_repr_with_authn_credential_provider(): + # GIVEN + def my_provider(): + return ("mechanism", b"data") + + # WHEN + options = blazingmq.SessionOptions(authn_credential_provider=my_provider) + + # THEN + assert "authn_credential_provider=" in repr(options)