diff --git a/pyproject.toml b/pyproject.toml index 91a064d..d1b6a0f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,14 +86,11 @@ disallow_untyped_calls = false check_untyped_defs = true [[tool.mypy.overrides]] -# Legacy test files: test type-validation (intentionally wrong types), -# use pre-rewrite channel API, or have Python 2 compat cruft +# Legacy/integration test files: test type-validation (intentionally wrong +# types), use pre-rewrite channel API, or have Python 2 compat cruft module = [ - "tests.integration_tests", - "tests.utils_tests", - "tests.base_tests", - "tests.channel_test", "tests.helpers", + "tests.test_integration", "tests.test_queue", "tests.test_message", ] @@ -129,6 +126,7 @@ exclude = [ "tests/channel_test.py", "tests/helpers.py", "tests/integration_tests.py", + "tests/test_integration.py", "tests/test_message.py", "tests/test_queue.py", "tests/utils_tests.py", diff --git a/rabbitpy/base.py b/rabbitpy/base.py index 0c8ea32..17f0cea 100644 --- a/rabbitpy/base.py +++ b/rabbitpy/base.py @@ -292,7 +292,7 @@ def write_frames(self, frames: list[base.Frame]) -> None: def _build_close_frame(self) -> commands.Channel.Close: """Return the proper close frame for this object.""" return self.CLOSE_REQUEST_FRAME( - self.DEFAULT_CLOSE_CODE, self.DEFAULT_CLOSE_REASON + self.DEFAULT_CLOSE_CODE, self.DEFAULT_CLOSE_REASON, 0, 0 ) def _can_write(self) -> bool: @@ -441,7 +441,7 @@ def _validate_frame_type( if result: return True return False - elif isinstance(frame_value, commands.Frame): + elif isinstance(frame_value, base.Frame): return frame_value.name == frame_type.name return False diff --git a/rabbitpy/channel.py b/rabbitpy/channel.py index ffe3a13..7c3a59d 100644 --- a/rabbitpy/channel.py +++ b/rabbitpy/channel.py @@ -376,7 +376,15 @@ def _create_message( return None if not header_frame: LOGGER.debug('Malformed header frame: %r', header_frame) - props = header_frame.properties.to_dict() if header_frame else {} + if header_frame: + p = header_frame.properties + props = { + k: getattr(p, k) + for k in p.attributes() + if getattr(p, k) is not None + } + else: + props = {} msg = message.Message(self, body, props) msg.method = method_frame msg.name = method_frame.name diff --git a/rabbitpy/channel0.py b/rabbitpy/channel0.py index 564f5a1..9a748e6 100644 --- a/rabbitpy/channel0.py +++ b/rabbitpy/channel0.py @@ -148,6 +148,7 @@ def _negotiate(self) -> None: # Step 2: wait for Connection.Start frame = self._wait_for_frame(timeout=self._args['timeout']) + self._raise_if_server_close(frame) if not isinstance(frame, commands.Connection.Start): raise exceptions.ConnectionException( f'Expected Connection.Start, got {type(frame).__name__}' @@ -156,6 +157,7 @@ def _negotiate(self) -> None: # Step 3: wait for Connection.Tune frame = self._wait_for_frame(timeout=self._args['timeout']) + self._raise_if_server_close(frame) if not isinstance(frame, commands.Connection.Tune): raise exceptions.ConnectionException( f'Expected Connection.Tune, got {type(frame).__name__}' @@ -164,6 +166,7 @@ def _negotiate(self) -> None: # Step 4: send Connection.Open (TuneOk + Open sent in _on_connection_tune) frame = self._wait_for_frame(timeout=self._args['timeout']) + self._raise_if_server_close(frame) if not isinstance(frame, commands.Connection.OpenOk): raise exceptions.ConnectionException( f'Expected Connection.OpenOk, got {type(frame).__name__}' @@ -208,6 +211,25 @@ def _handle_runtime_frame(self, frame: pamqp.frame.FrameTypes) -> None: self._exceptions.put(exc_cls(frame.reply_code, frame.reply_text)) self._events.set(ev_module.EXCEPTION_RAISED) + def _raise_if_server_close(self, frame: pamqp.frame.FrameTypes) -> None: + """Raise the appropriate exception if the server sent Connection.Close. + + Called after each ``_wait_for_frame`` during negotiation so that an + access-refused or forced-close from the broker surfaces as the correct + exception type rather than a generic ConnectionException. + + """ + if isinstance(frame, commands.Connection.Close): + LOGGER.error( + 'Server closed the connection during negotiation (%s): %s', + frame.reply_code, + frame.reply_text, + ) + exc_cls = exceptions.AMQP.get( + frame.reply_code, exceptions.RemoteClosedException + ) + raise exc_cls(frame.reply_code, frame.reply_text) + def _wait_for_frame(self, timeout: float) -> pamqp.frame.FrameTypes: """Block until a frame arrives on pending_frames or timeout expires. diff --git a/rabbitpy/connection.py b/rabbitpy/connection.py index 910418c..ba32e97 100644 --- a/rabbitpy/connection.py +++ b/rabbitpy/connection.py @@ -97,7 +97,7 @@ def __init__( connection_name: str | None = None, client_properties: dict[str, typing.Any] | None = None, ) -> None: - """Create a new instance of the Connection object""" + """Create a new instance of the Connection object and connect.""" super().__init__() self._args: url_parser.ConnectionArgs = url_parser.parse(url) self._channel_lock = threading.Lock() @@ -110,13 +110,15 @@ def __init__( self._heartbeat: heartbeat_mod.Heartbeat | None = None self._io: io.IO | None = None self._name = connection_name or '0x%x' % id(self) # noqa: UP031 + self.connect() def __enter__(self) -> typing.Self: """For use as a context manager, return a handle to this object instance. """ - self.connect() + if not self.is_open: + self.connect() return self def __exit__( diff --git a/rabbitpy/message.py b/rabbitpy/message.py index 1a414d6..a0d699a 100644 --- a/rabbitpy/message.py +++ b/rabbitpy/message.py @@ -363,10 +363,13 @@ def _coerce_properties(self) -> None: continue python_type = self._AMQP_TYPE_MAP.get(amqp_type) if python_type == 'str': - if not isinstance(value, (bytes, str)): - LOGGER.warning('Coercing property %s to bytes', key) - value = str(value) - self.properties[key] = utils.maybe_utf8_encode(value) + if isinstance(value, bytes): + self.properties[key] = value.decode( + 'utf-8', errors='replace' + ) + elif not isinstance(value, str): + LOGGER.warning('Coercing property %s to str', key) + self.properties[key] = str(value) elif python_type == 'int': LOGGER.warning('Coercing property %s to int', key) try: diff --git a/tests/base_tests.py b/tests/test_base.py similarity index 81% rename from tests/base_tests.py rename to tests/test_base.py index 516b7ea..bb487bf 100644 --- a/tests/base_tests.py +++ b/tests/test_base.py @@ -3,6 +3,8 @@ """ +import unittest + from rabbitpy import base, utils from tests import helpers @@ -16,14 +18,14 @@ def test_channel_invalid(self): self.assertRaises(ValueError, base.AMQPClass, 'Foo', 'Bar') def test_name_bytes(self): - obj = base.AMQPClass(self.channel, b'Foo') - self.assertIsInstance(obj.name, bytes) + # Python 3 only: bytes names are not accepted + self.assertRaises(ValueError, base.AMQPClass, self.channel, b'Foo') def test_name_str(self): obj = base.AMQPClass(self.channel, 'Foo') self.assertIsInstance(obj.name, str) - @helpers.unittest.skipIf(utils.PYTHON3, 'No unicode in Python 3') + @unittest.skipIf(utils.PYTHON3, 'No unicode in Python 3') def test_name_py2_unicode(self): obj = base.AMQPClass(self.channel, 'Foo') self.assertIsInstance(obj.name, str) diff --git a/tests/channel_test.py b/tests/test_channel.py similarity index 98% rename from tests/channel_test.py rename to tests/test_channel.py index d7bc348..e768af7 100644 --- a/tests/channel_test.py +++ b/tests/test_channel.py @@ -52,7 +52,7 @@ def test_invoking_consume_raises(self): self.channel._server_capabilities['consumer_priorities'] = False self.assertRaises( exceptions.NotSupportedError, - self.channel._consume, + self.channel.register_consumer, self, True, 100, diff --git a/tests/test_channel0.py b/tests/test_channel0.py new file mode 100644 index 0000000..5bde103 --- /dev/null +++ b/tests/test_channel0.py @@ -0,0 +1,424 @@ +"""Unit tests for rabbitpy.channel0 AMQP connection-negotiation thread.""" + +import queue +import unittest +from unittest import mock + +import pamqp.heartbeat +from pamqp import commands, header + +from rabbitpy import channel0, events, exceptions +from rabbitpy.url_parser import ConnectionArgs, SslOptions + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_DEFAULT_SSL_OPTIONS: SslOptions = { + 'check_hostname': False, + 'cafile': None, + 'capath': None, + 'certfile': None, + 'keyfile': None, + 'verify': None, +} + + +def _make_args(**overrides: object) -> ConnectionArgs: + base: ConnectionArgs = { + 'username': 'guest', + 'password': 'guest', + 'virtual_host': '/', + 'timeout': 3, + 'heartbeat': 60, + 'frame_max': 131072, + 'channel_max': 65535, + 'locale': 'en_US', + 'host': 'localhost', + 'port': 5672, + 'ssl': False, + 'ssl_options': _DEFAULT_SSL_OPTIONS, + } + base.update(overrides) # type: ignore[typeddict-item] + return base + + +def _make(**arg_overrides): + """Return (Channel0, Events, exc_queue).""" + args = _make_args(**arg_overrides) + ev = events.Events() + exc: queue.Queue[Exception] = queue.Queue() + ch0 = channel0.Channel0(args=args, events=ev, exceptions_queue=exc) + return ch0, ev, exc + + +def _mock_io(): + return mock.MagicMock(name='io.IO') + + +# Pre-built frames for the happy-path handshake +_START = commands.Connection.Start( + version_major=0, + version_minor=9, + server_properties={'capabilities': {'publisher_confirms': True}}, + mechanisms='PLAIN', + locales='en_US', +) +_TUNE = commands.Connection.Tune(channel_max=0, frame_max=131072, heartbeat=60) +_OPEN_OK = commands.Connection.OpenOk() + + +def _load_happy_path(ch0_obj): + """Pre-load negotiation frames and a None terminator for _run_loop.""" + for frame in (_START, _TUNE, _OPEN_OK, None): + ch0_obj.pending_frames.put(frame) + + +def _run(ch0_obj, mock_io=None, timeout=5): + """Start ch0 thread and wait for it to finish.""" + ch0_obj.start(mock_io or _mock_io()) + ch0_obj.join(timeout=timeout) + + +# --------------------------------------------------------------------------- +# Properties +# --------------------------------------------------------------------------- + + +class TestChannel0Properties(unittest.TestCase): + def setUp(self): + self.ch0, self.ev, self.exc = _make() + + def test_initial_open_is_false(self): + self.assertFalse(self.ch0.open) + + def test_initial_properties_empty(self): + self.assertEqual(self.ch0.properties, {}) + + def test_maximum_channels_from_args(self): + self.assertEqual(self.ch0.maximum_channels, 65535) + + def test_maximum_frame_size_from_args(self): + self.assertEqual(self.ch0.maximum_frame_size, 131072) + + def test_heartbeat_interval_from_args(self): + self.assertEqual(self.ch0.heartbeat_interval, 60) + + def test_daemon_thread(self): + self.assertTrue(self.ch0.daemon) + + +# --------------------------------------------------------------------------- +# Static helpers +# --------------------------------------------------------------------------- + + +class TestNegotiateValue(unittest.TestCase): + def _n(self, s, c): + return channel0.Channel0._negotiate_value(s, c) + + def test_both_nonzero_returns_minimum(self): + self.assertEqual(self._n(100, 50), 50) + self.assertEqual(self._n(50, 100), 50) + + def test_server_zero_returns_client(self): + self.assertEqual(self._n(0, 50), 50) + + def test_client_zero_returns_server(self): + self.assertEqual(self._n(100, 0), 100) + + def test_both_zero_returns_zero(self): + self.assertEqual(self._n(0, 0), 0) + + def test_equal_values_returns_same(self): + self.assertEqual(self._n(60, 60), 60) + + +class TestBuildClientProperties(unittest.TestCase): + def setUp(self): + self.props = channel0.Channel0._build_client_properties() + + def test_product_is_rabbitpy(self): + self.assertEqual(self.props['product'], 'rabbitpy') + + def test_capabilities_present(self): + self.assertIn('capabilities', self.props) + + def test_publisher_confirms_capability(self): + self.assertTrue(self.props['capabilities']['publisher_confirms']) + + def test_version_present(self): + self.assertIn('version', self.props) + + def test_platform_present(self): + self.assertIn('platform', self.props) + + +# --------------------------------------------------------------------------- +# Happy-path negotiation +# --------------------------------------------------------------------------- + + +class TestHappyPathNegotiation(unittest.TestCase): + def setUp(self): + self.ch0, self.ev, self.exc = _make() + self.mock_io = _mock_io() + _load_happy_path(self.ch0) + _run(self.ch0, self.mock_io) + + def _frames_written(self): + return [c[0][1] for c in self.mock_io.write_frame.call_args_list] + + def test_open_is_true(self): + self.assertTrue(self.ch0.open) + + def test_channel0_opened_event_set(self): + self.assertTrue(self.ev.is_set(events.CHANNEL0_OPENED)) + + def test_no_exceptions(self): + self.assertTrue(self.exc.empty()) + + def test_protocol_header_sent_first(self): + self.assertIsInstance(self._frames_written()[0], header.ProtocolHeader) + + def test_start_ok_sent(self): + frames = self._frames_written() + self.assertTrue( + any(isinstance(f, commands.Connection.StartOk) for f in frames) + ) + + def test_credentials_in_start_ok(self): + frames = self._frames_written() + start_ok = next( + f for f in frames if isinstance(f, commands.Connection.StartOk) + ) + self.assertIn('guest', start_ok.response) + + def test_tune_ok_sent(self): + frames = self._frames_written() + self.assertTrue( + any(isinstance(f, commands.Connection.TuneOk) for f in frames) + ) + + def test_connection_open_sent(self): + frames = self._frames_written() + self.assertTrue( + any(isinstance(f, commands.Connection.Open) for f in frames) + ) + + def test_virtual_host_in_open(self): + frames = self._frames_written() + open_frame = next( + f for f in frames if isinstance(f, commands.Connection.Open) + ) + self.assertEqual(open_frame.virtual_host, '/') + + def test_server_properties_stored(self): + self.assertIn('capabilities', self.ch0.properties) + + def test_heartbeat_negotiated(self): + # server=60, client=60 → min(60,60) = 60 + self.assertEqual(self.ch0.heartbeat_interval, 60) + + +# --------------------------------------------------------------------------- +# Heartbeat negotiation edge cases +# --------------------------------------------------------------------------- + + +class TestHeartbeatNegotiation(unittest.TestCase): + def _negotiate(self, server_hb, client_hb, **extra_tune): + ch0, _, _ = _make(heartbeat=client_hb) + tune = commands.Connection.Tune( + channel_max=extra_tune.get('channel_max', 0), + frame_max=extra_tune.get('frame_max', 131072), + heartbeat=server_hb, + ) + for frame in (_START, tune, _OPEN_OK, None): + ch0.pending_frames.put(frame) + _run(ch0) + return ch0 + + def test_both_nonzero_takes_minimum(self): + ch0 = self._negotiate(30, 60) + self.assertEqual(ch0.heartbeat_interval, 30) + + def test_server_zero_disables(self): + ch0 = self._negotiate(0, 60) + self.assertEqual(ch0.heartbeat_interval, 0) + + def test_client_zero_disables(self): + ch0 = self._negotiate(60, 0) + self.assertEqual(ch0.heartbeat_interval, 0) + + def test_both_zero_disables(self): + ch0 = self._negotiate(0, 0) + self.assertEqual(ch0.heartbeat_interval, 0) + + def test_server_zero_frame_max_uses_client(self): + ch0 = self._negotiate(60, 60, frame_max=0, channel_max=0) + self.assertEqual(ch0.maximum_frame_size, 131072) + self.assertEqual(ch0.maximum_channels, 65535) + + def test_server_smaller_frame_max_used(self): + ch0 = self._negotiate(60, 60, frame_max=65536) + self.assertEqual(ch0.maximum_frame_size, 65536) + + +# --------------------------------------------------------------------------- +# Negotiation error paths +# --------------------------------------------------------------------------- + + +class TestNegotiationErrors(unittest.TestCase): + def _run_and_join(self, ch0_obj): + _run(ch0_obj) + + def test_wrong_first_frame_raises_connection_exception(self): + ch0, _ev, exc = _make() + ch0.pending_frames.put(commands.Connection.Tune(0, 131072, 60)) + self._run_and_join(ch0) + self.assertFalse(ch0.open) + self.assertFalse(exc.empty()) + self.assertIsInstance(exc.get_nowait(), exceptions.ConnectionException) + + def test_wrong_tune_frame_raises_connection_exception(self): + ch0, _ev, exc = _make() + ch0.pending_frames.put(_START) + ch0.pending_frames.put(commands.Connection.OpenOk()) + self._run_and_join(ch0) + self.assertFalse(ch0.open) + self.assertFalse(exc.empty()) + self.assertIsInstance(exc.get_nowait(), exceptions.ConnectionException) + + def test_wrong_open_ok_frame_raises_connection_exception(self): + ch0, _ev, exc = _make() + ch0.pending_frames.put(_START) + ch0.pending_frames.put(_TUNE) + ch0.pending_frames.put(_TUNE) # wrong type + self._run_and_join(ch0) + self.assertFalse(ch0.open) + self.assertFalse(exc.empty()) + + def test_pre_injected_exception_propagates(self): + ch0, _ev, exc = _make() + exc.put(exceptions.ConnectionException('pre-injected')) + self._run_and_join(ch0) + self.assertFalse(ch0.open) + self.assertFalse(exc.empty()) + + def test_exception_raised_event_set_on_error(self): + ch0, ev, _exc = _make() + ch0.pending_frames.put(commands.Connection.Tune(0, 0, 0)) + self._run_and_join(ch0) + self.assertTrue(ev.is_set(events.EXCEPTION_RAISED)) + + def test_channel0_opened_not_set_on_error(self): + ch0, ev, _exc = _make() + ch0.pending_frames.put(commands.Connection.Tune(0, 0, 0)) + self._run_and_join(ch0) + self.assertFalse(ev.is_set(events.CHANNEL0_OPENED)) + + +# --------------------------------------------------------------------------- +# Runtime frames (post-negotiation) +# --------------------------------------------------------------------------- + + +class TestRuntimeFrames(unittest.TestCase): + def _open(self, extra_frames): + ch0, ev, exc = _make() + for frame in (_START, _TUNE, _OPEN_OK, *extra_frames, None): + ch0.pending_frames.put(frame) + _run(ch0) + return ch0, ev, exc + + def test_connection_blocked_sets_event(self): + _, ev, _ = self._open( + [commands.Connection.Blocked(reason='memory alarm')] + ) + self.assertTrue(ev.is_set(events.CONNECTION_BLOCKED)) + + def test_connection_unblocked_clears_event(self): + _, ev, _ = self._open( + [ + commands.Connection.Blocked(reason='memory alarm'), + commands.Connection.Unblocked(), + ] + ) + self.assertFalse(ev.is_set(events.CONNECTION_BLOCKED)) + + def test_server_close_queues_exception(self): + ch0, _, exc = self._open( + [ + commands.Connection.Close( + reply_code=320, + reply_text='CONNECTION_FORCED', + class_id=0, + method_id=0, + ) + ] + ) + self.assertFalse(ch0.open) + self.assertFalse(exc.empty()) + + def test_server_close_maps_known_reply_code(self): + _ch0, _, exc = self._open( + [ + commands.Connection.Close( + reply_code=403, + reply_text='ACCESS_REFUSED', + class_id=0, + method_id=0, + ) + ] + ) + err = exc.get_nowait() + self.assertIsInstance(err, exceptions.AMQPAccessRefused) + + +# --------------------------------------------------------------------------- +# Heartbeat and close +# --------------------------------------------------------------------------- + + +class TestSendHeartbeat(unittest.TestCase): + def test_sends_heartbeat_frame(self): + ch0, _, _ = _make() + mock_io = _mock_io() + ch0._io = mock_io + ch0.send_heartbeat() + mock_io.write_frame.assert_called_once() + args = mock_io.write_frame.call_args[0] + self.assertEqual(args[0], 0) + self.assertIsInstance(args[1], pamqp.heartbeat.Heartbeat) + + def test_no_io_does_not_raise(self): + ch0, _, _ = _make() + # _io is None — should succeed silently + ch0.send_heartbeat() + + def test_close_when_open_sends_frame(self): + ch0, _, _ = _make() + mock_io = _mock_io() + ch0._io = mock_io + ch0._open = True + ch0.close() + mock_io.write_frame.assert_called_once() + close_frame = mock_io.write_frame.call_args[0][1] + self.assertIsInstance(close_frame, commands.Connection.Close) + + def test_close_marks_not_open(self): + ch0, _, _ = _make() + ch0._io = _mock_io() + ch0._open = True + ch0.close() + self.assertFalse(ch0.open) + + def test_close_when_not_open_does_not_write(self): + ch0, _, _ = _make() + mock_io = _mock_io() + ch0._io = mock_io + ch0._open = False + ch0.close() + mock_io.write_frame.assert_not_called() diff --git a/tests/test_connection.py b/tests/test_connection.py new file mode 100644 index 0000000..613d2b5 --- /dev/null +++ b/tests/test_connection.py @@ -0,0 +1,371 @@ +"""Unit tests for rabbitpy.connection.Connection.""" + +import queue +import unittest +from unittest import mock + +from rabbitpy import connection, events, exceptions + +_DEFAULT_URL = 'amqp://guest:guest@localhost:5672/%2F' + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_conn(url=_DEFAULT_URL, **kwargs): + with mock.patch('rabbitpy.connection.Connection.connect'): + return connection.Connection(url, **kwargs) + + +def _make_mock_io(events_obj, *, signal_opened=True): + m = mock.MagicMock(name='io.IO') + m.write_trigger = mock.MagicMock() + m.write_queue = queue.Queue() + if signal_opened: + + def _start(): + events_obj.set(events.SOCKET_OPENED) + + m.start.side_effect = _start + return m + + +def _make_mock_ch0( + events_obj, *, signal_opened=True, exc_queue=None, exception=None +): + m = mock.MagicMock(name='channel0.Channel0') + m.pending_frames = queue.Queue() + m.open = True + m.heartbeat_interval = 0 + m.maximum_frame_size = 131072 + m.maximum_channels = 65535 + m.properties = {} + if signal_opened and exception is None: + + def _start(io): + events_obj.set(events.CHANNEL0_OPENED) + + m.start.side_effect = _start + elif exception is not None and exc_queue is not None: + + def _start_exc(io): + exc_queue.put(exception) + + m.start.side_effect = _start_exc + return m + + +# --------------------------------------------------------------------------- +# Construction +# --------------------------------------------------------------------------- + + +class TestConnectionInit(unittest.TestCase): + def test_default_url_host(self): + conn = _make_conn() + self.assertEqual(conn._args['host'], 'localhost') + + def test_default_url_port(self): + conn = _make_conn() + self.assertEqual(conn._args['port'], 5672) + + def test_custom_url_parsed(self): + conn = _make_conn('amqp://user:pass@myhost:1234/vhost') + self.assertEqual(conn._args['host'], 'myhost') + self.assertEqual(conn._args['username'], 'user') + self.assertEqual(conn._args['password'], 'pass') + + def test_initial_state_is_closed(self): + conn = _make_conn() + self.assertTrue(conn.is_closed) + + def test_custom_connection_name(self): + conn = _make_conn(connection_name='my-conn') + self.assertEqual(conn._name, 'my-conn') + + def test_channels_dict_empty(self): + conn = _make_conn() + self.assertEqual(conn._channels, {}) + + def test_freed_ids_set_empty(self): + conn = _make_conn() + self.assertEqual(conn._freed_channel_ids, set()) + + +# --------------------------------------------------------------------------- +# connect() lifecycle +# --------------------------------------------------------------------------- + + +class TestConnectionConnect(unittest.TestCase): + def setUp(self): + self.conn = _make_conn() + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_happy_path_state_is_open(self, mock_ch0_cls, mock_io_cls): + mock_io_cls.return_value = _make_mock_io(self.conn._events) + mock_ch0_cls.return_value = _make_mock_ch0(self.conn._events) + self.conn.connect() + self.assertTrue(self.conn.is_open) + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_context_manager_opens_and_closes(self, mock_ch0_cls, mock_io_cls): + mock_io = _make_mock_io(self.conn._events) + mock_io_cls.return_value = mock_io + mock_ch0_cls.return_value = _make_mock_ch0(self.conn._events) + with self.conn: + self.assertTrue(self.conn.is_open) + self.assertTrue(self.conn.is_closed) + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_socket_timeout_raises_runtime_error( + self, mock_ch0_cls, mock_io_cls + ): + mock_io_cls.return_value = _make_mock_io( + self.conn._events, signal_opened=False + ) + mock_ch0_cls.return_value = mock.MagicMock() + with self.assertRaises(RuntimeError): + self.conn.connect() + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_socket_exception_propagates(self, mock_ch0_cls, mock_io_cls): + exc = exceptions.ConnectionException('refused') + + def _start(): + self.conn._exceptions.put(exc) + + mock_io = mock.MagicMock() + mock_io.start.side_effect = _start + mock_io_cls.return_value = mock_io + mock_ch0_cls.return_value = mock.MagicMock() + with self.assertRaises(exceptions.ConnectionException): + self.conn.connect() + + @mock.patch('rabbitpy.connection.time') + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_negotiation_timeout_raises_runtime_error( + self, mock_ch0_cls, mock_io_cls, mock_time + ): + mock_io_cls.return_value = _make_mock_io(self.conn._events) + mock_ch0 = mock.MagicMock() + mock_ch0.pending_frames = queue.Queue() + mock_ch0_cls.return_value = mock_ch0 + # Deadline expires immediately on second monotonic() call + mock_time.monotonic.side_effect = [0.0, 9999.0] + mock_time.sleep = mock.MagicMock() + with self.assertRaises(RuntimeError): + self.conn.connect() + self.assertTrue(self.conn.is_closed) + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_negotiation_exception_propagates(self, mock_ch0_cls, mock_io_cls): + mock_io_cls.return_value = _make_mock_io(self.conn._events) + exc = exceptions.AMQPAccessRefused('access refused') + mock_ch0_cls.return_value = _make_mock_ch0( + self.conn._events, + signal_opened=False, + exc_queue=self.conn._exceptions, + exception=exc, + ) + with self.assertRaises(exceptions.AMQPAccessRefused): + self.conn.connect() + self.assertTrue(self.conn.is_closed) + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_heartbeat_started_when_interval_nonzero( + self, mock_ch0_cls, mock_io_cls + ): + mock_io_cls.return_value = _make_mock_io(self.conn._events) + mock_ch0 = _make_mock_ch0(self.conn._events) + mock_ch0.heartbeat_interval = 60 + mock_ch0_cls.return_value = mock_ch0 + with mock.patch( + 'rabbitpy.connection.heartbeat_mod.Heartbeat' + ) as mock_hb_cls: + mock_hb_cls.return_value = mock.MagicMock() + self.conn.connect() + mock_hb_cls.return_value.start.assert_called_once() + + @mock.patch('rabbitpy.connection.io.IO') + @mock.patch('rabbitpy.connection.channel0.Channel0') + def test_heartbeat_not_started_when_interval_zero( + self, mock_ch0_cls, mock_io_cls + ): + mock_io_cls.return_value = _make_mock_io(self.conn._events) + mock_ch0_cls.return_value = _make_mock_ch0(self.conn._events) + with mock.patch( + 'rabbitpy.connection.heartbeat_mod.Heartbeat' + ) as mock_hb_cls: + self.conn.connect() + mock_hb_cls.assert_not_called() + + +# --------------------------------------------------------------------------- +# channel() method +# --------------------------------------------------------------------------- + + +class TestConnectionChannelMethod(unittest.TestCase): + """Tests for Connection.channel() using a pre-opened mock connection. + + Channel.open() is overridden to set state=OPEN without doing any I/O, + so that the channel lifecycle tracking in the connection works correctly. + """ + + def setUp(self): + from rabbitpy import channel as channel_mod + + self.conn = _make_conn() + self.conn._set_state(self.conn.OPEN) + mock_io = mock.MagicMock() + mock_io.write_trigger = mock.MagicMock() + mock_io.write_queue = queue.Queue() + self.conn._io = mock_io + mock_ch0 = mock.MagicMock() + mock_ch0.maximum_frame_size = 131072 + mock_ch0.maximum_channels = 65535 + mock_ch0.properties = {} + self.conn._channel0 = mock_ch0 + # Override open() so channels reach OPEN state without real I/O + self._open_patcher = mock.patch.object( + channel_mod.Channel, + 'open', + lambda self: self._set_state(self.OPEN), + ) + self._open_patcher.start() + + def tearDown(self): + self._open_patcher.stop() + + def test_channel_returns_channel_instance(self): + from rabbitpy import channel as channel_mod + + ch = self.conn.channel() + self.assertIsInstance(ch, channel_mod.Channel) + + def test_first_channel_id_is_one(self): + ch = self.conn.channel() + self.assertEqual(ch.id, 1) + + def test_second_channel_id_is_two(self): + self.conn.channel() + ch2 = self.conn.channel() + self.assertEqual(ch2.id, 2) + + def test_channel_registered_in_channels(self): + ch = self.conn.channel() + self.assertIn(ch.id, self.conn._channels) + + def test_channel_registered_in_io(self): + self.conn.channel() + self.conn._io.add_channel.assert_called() # pyright: ignore[reportOptionalMemberAccess,reportAttributeAccessIssue] + + def test_channel_raises_when_connection_closed(self): + self.conn._set_state(self.conn.CLOSED) + with self.assertRaises(exceptions.ConnectionClosed): + self.conn.channel() + + +# --------------------------------------------------------------------------- +# Channel ID reuse (#121) — tested via _get_next_channel_id directly so that +# we can inject mock channels with controlled closed/open states. +# --------------------------------------------------------------------------- + + +class TestChannelIdReuse(unittest.TestCase): + def setUp(self): + self.conn = _make_conn() + self.conn._set_state(self.conn.OPEN) + mock_ch0 = mock.MagicMock() + mock_ch0.maximum_channels = 65535 + self.conn._channel0 = mock_ch0 + + def _mock_chan(self, closed=False): + m = mock.MagicMock() + m.closed = closed + return m + + def _next_id(self): + with self.conn._channel_lock: + return self.conn._get_next_channel_id() + + def test_first_id_is_one_when_no_channels(self): + self.assertEqual(self._next_id(), 1) + + def test_id_increments_when_channels_open(self): + self.conn._channels[1] = self._mock_chan(closed=False) + self.conn._channels[2] = self._mock_chan(closed=False) + self.assertEqual(self._next_id(), 3) + + def test_closed_channel_id_is_reclaimed(self): + self.conn._channels[1] = self._mock_chan(closed=True) + self.conn._channels[2] = self._mock_chan(closed=False) + self.assertEqual(self._next_id(), 1) + + def test_all_closed_ids_reclaimed_before_new_allocation(self): + self.conn._channels[1] = self._mock_chan(closed=True) + self.conn._channels[2] = self._mock_chan(closed=True) + first = self._next_id() + self.assertIn(first, {1, 2}) + + def test_closed_channel_removed_from_dict_during_cleanup(self): + self.conn._channels[1] = self._mock_chan(closed=True) + self._next_id() + self.assertNotIn(1, self.conn._channels) + + def test_too_many_channels_raises(self): + self.conn._channel0.maximum_channels = 2 # pyright: ignore[reportOptionalMemberAccess,reportAttributeAccessIssue] + self.conn._channels[1] = self._mock_chan(closed=False) + self.conn._channels[2] = self._mock_chan(closed=False) + with self.assertRaises(exceptions.TooManyChannelsError): + self._next_id() + + +# --------------------------------------------------------------------------- +# Properties +# --------------------------------------------------------------------------- + + +class TestConnectionProperties(unittest.TestCase): + def test_capabilities_empty_without_channel0(self): + conn = _make_conn() + conn._channel0 = None + self.assertEqual(conn.capabilities, {}) + + def test_server_properties_empty_without_channel0(self): + conn = _make_conn() + conn._channel0 = None + self.assertEqual(conn.server_properties, {}) + + def test_capabilities_from_channel0(self): + conn = _make_conn() + m = mock.MagicMock() + m.properties = {'capabilities': {'basic.nack': True}} + conn._channel0 = m + self.assertEqual(conn.capabilities, {'basic.nack': True}) + + def test_server_properties_from_channel0(self): + conn = _make_conn() + m = mock.MagicMock() + m.properties = {'version': '3.12'} + conn._channel0 = m + self.assertEqual(conn.server_properties, {'version': '3.12'}) + + def test_blocked_false_when_event_not_set(self): + conn = _make_conn() + self.assertFalse(conn.blocked) + + def test_blocked_true_when_event_set(self): + conn = _make_conn() + conn._events.set(events.CONNECTION_BLOCKED) + self.assertTrue(conn.blocked) diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py new file mode 100644 index 0000000..c5739a4 --- /dev/null +++ b/tests/test_heartbeat.py @@ -0,0 +1,116 @@ +"""Unit tests for rabbitpy.heartbeat.Heartbeat.""" + +import unittest +from unittest import mock + +from rabbitpy import heartbeat + + +def _make(interval=60): + mock_io = mock.MagicMock(name='io.IO') + mock_io.bytes_written = 0 + mock_ch0 = mock.MagicMock(name='channel0.Channel0') + hb = heartbeat.Heartbeat(mock_io, mock_ch0, interval) + return hb, mock_io, mock_ch0 + + +class TestHeartbeatInit(unittest.TestCase): + def test_interval_is_half_of_configured(self): + hb, _, _ = _make(60) + self.assertEqual(hb._interval, 30.0) + + def test_float_interval(self): + hb, _, _ = _make(10) + self.assertEqual(hb._interval, 5.0) + + def test_initial_stopped_false(self): + hb, _, _ = _make() + self.assertFalse(hb._stopped) + + def test_initial_timer_none(self): + hb, _, _ = _make() + self.assertIsNone(hb._timer) + + +class TestHeartbeatStartStop(unittest.TestCase): + def test_start_creates_timer(self): + hb, _, _ = _make() + hb.start() + try: + self.assertIsNotNone(hb._timer) + self.assertFalse(hb._stopped) + finally: + hb.stop() + + def test_stop_cancels_timer(self): + hb, _, _ = _make() + hb.start() + hb.stop() + self.assertIsNone(hb._timer) + self.assertTrue(hb._stopped) + + def test_stop_without_start_is_safe(self): + hb, _, _ = _make() + hb.stop() # should not raise + self.assertTrue(hb._stopped) + + def test_double_stop_is_safe(self): + hb, _, _ = _make() + hb.start() + hb.stop() + hb.stop() + + +class TestHeartbeatDisabled(unittest.TestCase): + def test_zero_interval_does_not_create_timer(self): + hb, _, _ = _make(0) + hb.start() + self.assertIsNone(hb._timer) + + def test_zero_interval_start_does_not_raise(self): + hb, _, _ = _make(0) + hb.start() # should not raise + + +class TestMaybeSend(unittest.TestCase): + def setUp(self): + self.hb, self.mock_io, self.mock_ch0 = _make(2) + self.hb._stopped = False + self.mock_io.bytes_written = 100 + self.hb._last_written = 100 + + def test_sends_heartbeat_when_no_data_written(self): + self.hb._maybe_send() + self.mock_ch0.send_heartbeat.assert_called_once() + + def test_no_send_when_data_was_written(self): + self.mock_io.bytes_written = 200 + self.hb._maybe_send() + self.mock_ch0.send_heartbeat.assert_not_called() + + def test_stopped_skips_send(self): + self.hb._stopped = True + self.hb._maybe_send() + self.mock_ch0.send_heartbeat.assert_not_called() + + def test_last_written_updated_after_check(self): + self.mock_io.bytes_written = 200 + self.hb._maybe_send() + self.assertEqual(self.hb._last_written, 200) + + def test_restarts_timer_after_send(self): + with mock.patch.object(self.hb, '_start_timer') as mock_start: + self.hb._maybe_send() + mock_start.assert_called_once() + + def test_no_timer_restart_when_stopped_after_send(self): + """If stopped is set after the first lock check, timer is not restarted.""" + original_send = self.mock_ch0.send_heartbeat + + def _stop_during_send(): + self.hb._stopped = True + + original_send.side_effect = _stop_during_send + with mock.patch.object(self.hb, '_start_timer') as mock_start: + self.hb._maybe_send() + mock_start.assert_not_called() diff --git a/tests/integration_tests.py b/tests/test_integration.py similarity index 73% rename from tests/integration_tests.py rename to tests/test_integration.py index 861beef..c12c419 100644 --- a/tests/integration_tests.py +++ b/tests/test_integration.py @@ -1,5 +1,6 @@ import logging import os +import pathlib import re import threading import time @@ -7,9 +8,13 @@ import uuid from urllib import parse +import dotenv + import rabbitpy from rabbitpy import exceptions, utils +dotenv.load_dotenv(pathlib.Path(__file__).parent.parent / '.env') + LOGGER = logging.getLogger(__name__) logging.basicConfig(level=logging.DEBUG) @@ -82,7 +87,7 @@ def test_get_returns_expected_message(self): self.assertEqual(msg.properties['app_id'], self.app_id) self.assertEqual( msg.properties['message_id'], - self.msg.properties['message_id'].decode('utf-8'), + self.msg.properties['message_id'], ) self.assertEqual( msg.properties['timestamp'], self.msg.properties['timestamp'] @@ -128,7 +133,7 @@ def test_get_returns_expected_message(self): self.assertEqual(msg.properties['app_id'], self.app_id) self.assertEqual( msg.properties['message_id'], - self.msg.properties['message_id'].decode('utf-8'), + self.msg.properties['message_id'], ) self.assertEqual( msg.properties['timestamp'], self.msg.properties['timestamp'] @@ -177,7 +182,7 @@ def test_iterator_returns_expected_message(self): self.assertEqual(msg.properties['app_id'], self.app_id) self.assertEqual( msg.properties['message_id'], - self.msg.properties['message_id'].decode('utf-8'), + self.msg.properties['message_id'], ) self.assertEqual( msg.properties['timestamp'], self.msg.properties['timestamp'] @@ -289,7 +294,7 @@ def test_iterator_returns_expected_message(self): self.assertEqual(msg.properties['app_id'], self.app_id) self.assertEqual( msg.properties['message_id'], - self.msg.properties['message_id'].decode('utf-8'), + self.msg.properties['message_id'], ) self.assertEqual( msg.properties['timestamp'], self.msg.properties['timestamp'] @@ -550,3 +555,211 @@ def test_exception_is_raised(self): ) with self.assertRaises(exceptions.AMQPAccessRefused): rabbitpy.Connection(url) + + +# --------------------------------------------------------------------------- +# Multiple channels on one connection +# --------------------------------------------------------------------------- + + +class MultipleChannelsTest(unittest.TestCase): + def setUp(self): + self.connection = rabbitpy.Connection(os.environ['RABBITMQ_URL']) + + def tearDown(self): + self.connection.close() + + def test_two_channels_have_different_ids(self): + ch1 = self.connection.channel() + ch2 = self.connection.channel() + self.assertNotEqual(ch1.id, ch2.id) + ch1.close() + ch2.close() + + def test_channel_id_reused_after_close(self): + ch1 = self.connection.channel() + first_id = ch1.id + ch1.close() + ch2 = self.connection.channel() + # The freed id should be reused + self.assertEqual(ch2.id, first_id) + ch2.close() + + def test_independent_queues_on_separate_channels(self): + ch1 = self.connection.channel() + ch2 = self.connection.channel() + q1 = rabbitpy.Queue(ch1, 'multi-ch-q1', auto_delete=True) + q2 = rabbitpy.Queue(ch2, 'multi-ch-q2', auto_delete=True) + q1.declare() + q2.declare() + msg1 = rabbitpy.Message(ch1, b'from-ch1') + msg2 = rabbitpy.Message(ch2, b'from-ch2') + msg1.publish('', routing_key='multi-ch-q1') + msg2.publish('', routing_key='multi-ch-q2') + self.assertEqual(q1.get(True).body, b'from-ch1') + self.assertEqual(q2.get(True).body, b'from-ch2') + q1.delete() + q2.delete() + ch1.close() + ch2.close() + + +# --------------------------------------------------------------------------- +# Nack and requeue=False +# --------------------------------------------------------------------------- + + +class NackWithoutRequeueTest(unittest.TestCase): + def setUp(self): + self.connection = rabbitpy.Connection(os.environ['RABBITMQ_URL']) + self.channel = self.connection.channel() + self.queue = rabbitpy.Queue( + self.channel, 'nack-no-requeue', auto_delete=True + ) + self.queue.declare() + msg = rabbitpy.Message(self.channel, b'nack-me') + msg.publish('', routing_key='nack-no-requeue') + + def tearDown(self): + self.queue.delete() + self.channel.close() + self.connection.close() + + def test_nacked_message_not_requeued(self): + msg = self.queue.get(True) + self.assertIsNotNone(msg) + msg.nack(requeue=False) + # Queue should be empty — message discarded + self.assertIsNone(self.queue.get(True)) + + +# --------------------------------------------------------------------------- +# Queue purge +# --------------------------------------------------------------------------- + + +class QueuePurgeTest(unittest.TestCase): + def setUp(self): + self.connection = rabbitpy.Connection(os.environ['RABBITMQ_URL']) + self.channel = self.connection.channel() + self.channel.enable_publisher_confirms() + self.queue = rabbitpy.Queue( + self.channel, 'purge-test-queue', auto_delete=True + ) + self.queue.declare() + for _ in range(5): + rabbitpy.Message(self.channel, b'x').publish( + '', routing_key='purge-test-queue' + ) + + def tearDown(self): + self.queue.delete() + self.channel.close() + self.connection.close() + + def test_purge_empties_queue(self): + self.assertEqual(len(self.queue), 5) + self.queue.purge() + self.assertEqual(len(self.queue), 0) + + +# --------------------------------------------------------------------------- +# Large message (forces multi-frame body) +# --------------------------------------------------------------------------- + + +class LargeMessageTest(unittest.TestCase): + # Default frame_max is 131072 bytes; send something bigger + BODY_SIZE = 300_000 + + def setUp(self): + self.connection = rabbitpy.Connection(os.environ['RABBITMQ_URL']) + self.channel = self.connection.channel() + self.queue = rabbitpy.Queue( + self.channel, 'large-msg-queue', auto_delete=True + ) + self.queue.declare() + self.body = b'X' * self.BODY_SIZE + rabbitpy.Message(self.channel, self.body).publish( + '', routing_key='large-msg-queue' + ) + + def tearDown(self): + self.queue.delete() + self.channel.close() + self.connection.close() + + def test_large_message_body_round_trips(self): + msg = self.queue.get(True) + self.assertIsNotNone(msg) + self.assertEqual(len(msg.body), self.BODY_SIZE) + self.assertEqual(msg.body, self.body) + msg.ack() + + +# --------------------------------------------------------------------------- +# Channel context manager +# --------------------------------------------------------------------------- + + +class ChannelContextManagerTest(unittest.TestCase): + def setUp(self): + self.connection = rabbitpy.Connection(os.environ['RABBITMQ_URL']) + + def tearDown(self): + self.connection.close() + + def test_channel_closes_on_context_exit(self): + with self.connection.channel() as ch: + self.assertTrue(ch.open) + self.assertTrue(ch.closed) + + def test_channel_after_close_can_open_new(self): + with self.connection.channel(): + pass + ch2 = self.connection.channel() + self.assertTrue(ch2.open) + ch2.close() + + +# --------------------------------------------------------------------------- +# Transaction (Tx) commit and rollback +# --------------------------------------------------------------------------- + + +class TransactionTest(unittest.TestCase): + def setUp(self): + self.connection = rabbitpy.Connection(os.environ['RABBITMQ_URL']) + self.channel = self.connection.channel() + self.queue = rabbitpy.Queue( + self.channel, 'tx-test-queue', auto_delete=True + ) + self.queue.declare() + + def tearDown(self): + self.queue.delete() + self.channel.close() + self.connection.close() + + def test_committed_message_is_visible(self): + from rabbitpy import Tx + + with Tx(self.channel): + rabbitpy.Message(self.channel, b'committed').publish( + '', routing_key='tx-test-queue' + ) + msg = self.queue.get(True) + self.assertIsNotNone(msg) + self.assertEqual(msg.body, b'committed') + msg.ack() + + def test_rolled_back_message_not_visible(self): + from rabbitpy import Tx + + tx = Tx(self.channel) + tx.select() + rabbitpy.Message(self.channel, b'rolled-back').publish( + '', routing_key='tx-test-queue' + ) + tx.rollback() + self.assertIsNone(self.queue.get(True)) diff --git a/tests/test_message.py b/tests/test_message.py index 5aebc91..d531083 100644 --- a/tests/test_message.py +++ b/tests/test_message.py @@ -358,7 +358,8 @@ def test_prune_invalid_properties_removes_bogus_property(self): def test_coerce_property_int_to_str(self): self.msg.properties['expiration'] = 123 self.msg._coerce_properties() - self.assertIsInstance(self.msg.properties['expiration'], bytes) + # pamqp 4 expects str for shortstr properties, not bytes + self.assertIsInstance(self.msg.properties['expiration'], str) def test_coerce_property_str_to_int(self): self.msg.properties['priority'] = '9' diff --git a/tests/utils_tests.py b/tests/test_utils.py similarity index 100% rename from tests/utils_tests.py rename to tests/test_utils.py