Added __repr__ to Connection() and *Transport() classes

5 files changed, 69 insertions, 0 deletions

diff --git a/amqp/connection.py b/amqp/connection.py
index 7b34321..0c662b1 100644
--- a/amqp/connection.py
+++ b/amqp/connection.py
@@ -277,6 +277,14 @@ class Connection(AbstractChannel):
 
         self.connect_timeout = connect_timeout
 
+    def __repr__(self):
+        if self._transport:
+            return f'<AMQP Connection: {self.host}/{self.virtual_host} '\
+                   f'using {self._transport} at {id(self):#x}>'
+        else:
+            return f'<AMQP Connection: {self.host}/{self.virtual_host} '\
+                   f'(disconnected) at {id(self):#x}>'
+
     def __enter__(self):
         self.connect()
         return self
diff --git a/amqp/transport.py b/amqp/transport.py
index dfe7cfa..0818709 100644
--- a/amqp/transport.py
+++ b/amqp/transport.py
@@ -68,6 +68,14 @@ class _AbstractTransport:
         self.write_timeout = write_timeout
         self.socket_settings = socket_settings
 
+    def __repr__(self):
+        if self.sock:
+            src = f'{self.sock.getsockname()[0]}:{self.sock.getsockname()[1]}'
+            dst = f'{self.sock.getpeername()[0]}:{self.sock.getpeername()[1]}'
+            return f'<{type(self).__name__}: {src} -> {dst} at {id(self):#x}>'
+        else:
+            return f'<{type(self).__name__}: (disconnected) at {id(self):#x}>'
+
     def connect(self):
         try:
             # are we already connected?
diff --git a/t/integration/test_rmq.py b/t/integration/test_rmq.py
index ff13484..f6b26d1 100644
--- a/t/integration/test_rmq.py
+++ b/t/integration/test_rmq.py
@@ -48,6 +48,7 @@ def connection(request):
 @pytest.mark.flaky(reruns=5, reruns_delay=2)
 def test_connect(connection):
     connection.connect()
+    repr(connection)
     connection.close()
 
 
diff --git a/t/unit/test_connection.py b/t/unit/test_connection.py
index 67dcb85..7685f05 100644
--- a/t/unit/test_connection.py
+++ b/t/unit/test_connection.py
@@ -1,3 +1,4 @@
+import re
 import socket
 import warnings
 from unittest.mock import Mock, call, patch
@@ -501,3 +502,19 @@ class test_Connection:
     def test_server_capabilities(self):
         self.conn.server_properties['capabilities'] = {'foo': 1}
         assert self.conn.server_capabilities == {'foo': 1}
+
+    def test_repr_disconnected(self):
+        assert re.fullmatch(
+            r'<AMQP Connection: broker.com:1234// \(disconnected\) at 0x.*>',
+            repr(Connection(host='broker.com:1234'))
+        )
+
+    def test_repr_connected(self):
+        c = Connection(host='broker.com:1234')
+        c._transport = Mock(name='transport')
+        assert re.fullmatch(
+            r'<AMQP Connection: broker.com:1234// using {} at 0x.*>'.format(
+                repr(c.transport)
+            ),
+            repr(c)
+        )
diff --git a/t/unit/test_transport.py b/t/unit/test_transport.py
index 7152a60..fdb07b6 100644
--- a/t/unit/test_transport.py
+++ b/t/unit/test_transport.py
@@ -1,5 +1,6 @@
 import errno
 import os
+import re
 import socket
 import struct
 from struct import pack
@@ -53,6 +54,12 @@ class MockSocket:
         self.connected = False
         self.sa = None
 
+    def getsockname(self):
+        return ('127.0.0.1', 1234)
+
+    def getpeername(self):
+        return ('1.2.3.4', 5671)
+
 
 TCP_KEEPIDLE = 4
 TCP_KEEPINTVL = 5
@@ -579,6 +586,20 @@ class test_SSLTransport:
         def _init_socket(self, *args):
             pass
 
+    def test_repr_disconnected(self):
+        assert re.fullmatch(
+            r'<SSLTransport: \(disconnected\) at 0x.*>',
+            repr(transport.SSLTransport('host', 3))
+        )
+
+    def test_repr_connected(self):
+        t = transport.SSLTransport('host', 3)
+        t.sock = MockSocket()
+        re.fullmatch(
+            '<SSLTransport: 127.0.0.1:1234 -> 1.2.3.4:5671 at 0x.*>',
+            repr(t)
+        )
+
     @pytest.fixture(autouse=True)
     def setup_transport(self):
         self.t = self.Transport(
@@ -686,6 +707,20 @@ class test_TCPTransport:
         def _init_socket(self, *args):
             pass
 
+    def test_repr_disconnected(self):
+        assert re.fullmatch(
+            r'<TCPTransport: \(disconnected\) at 0x.*>',
+            repr(transport.TCPTransport('host', 3))
+        )
+
+    def test_repr_connected(self):
+        t = transport.SSLTransport('host', 3)
+        t.sock = MockSocket()
+        re.fullmatch(
+            '<TCPTransport: 127.0.0.1:1234 -> 1.2.3.4:5671 at 0x.*>',
+            repr(t)
+        )
+
     @pytest.fixture(autouse=True)
     def setup_transport(self):
         self.t = self.Transport('host', 3)