Fix test infrastructure, pamqp 4 compatibility, and runtime bugs

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",

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
 

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

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.
 

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__(

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:

tests/base_tests.py → 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)

tests/channel_test.py → 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,