summaryrefslogtreecommitdiff
path: root/t
diff options
context:
space:
mode:
authorMatus Valo <matusvalo@gmail.com>2021-08-05 14:47:05 +0200
committerAsif Saif Uddin <auvipy@gmail.com>2021-08-05 20:05:25 +0600
commitc0f2b847f05aa960f0c0d25ce5fcdd0078ec45b8 (patch)
treedbf8843a3e8ff2f5e53084cf857249bd7d6a6517 /t
parent200f183d9634cdef9ee50d2408ef279fbf769b1a (diff)
downloadkombu-c0f2b847f05aa960f0c0d25ce5fcdd0078ec45b8.tar.gz
Use hostname from URI when server_host is None
Diffstat (limited to 't')
-rw-r--r--t/unit/transport/test_pyamqp.py44
1 files changed, 43 insertions, 1 deletions
diff --git a/t/unit/transport/test_pyamqp.py b/t/unit/transport/test_pyamqp.py
index c7243e8e..84aa1fc2 100644
--- a/t/unit/transport/test_pyamqp.py
+++ b/t/unit/transport/test_pyamqp.py
@@ -1,9 +1,10 @@
import sys
from itertools import count
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, MagicMock
from case import mock
+import pytest
from kombu import Connection
from kombu.transport import pyamqp
@@ -161,6 +162,47 @@ class test_pyamqp:
c = Connection(port=1337, transport=Transport).connect()
assert c['host'] == '127.0.0.1:1337'
+ def test_ssl(self):
+ # Test setting TLS by ssl=True.
+ class Transport(pyamqp.Transport):
+ Connection = MagicMock()
+
+ Connection(transport=Transport, ssl=True).connect()
+ Transport.Connection.assert_called_once()
+ _, kwargs = Transport.Connection.call_args
+ assert kwargs['ssl'] is True
+
+ def test_ssl_dict(self):
+ # Test setting TLS by setting ssl as dict.
+ class Transport(pyamqp.Transport):
+ Connection = MagicMock()
+
+ Connection(transport=Transport, ssl={'a': 1, 'b': 2}).connect()
+ Transport.Connection.assert_called_once()
+ _, kwargs = Transport.Connection.call_args
+ assert kwargs['ssl'] == {'a': 1, 'b': 2}
+
+ @pytest.mark.parametrize(
+ 'hostname',
+ [
+ 'broker.example.com',
+ 'amqp://broker.example.com/0',
+ 'amqps://broker.example.com/0',
+ 'amqp://guest:guest@broker.example.com/0',
+ 'amqp://broker.example.com;broker2.example.com'
+ ])
+ def test_ssl_server_hostname(self, hostname):
+ # Test setting server_hostname from URI.
+ class Transport(pyamqp.Transport):
+ Connection = MagicMock()
+
+ Connection(
+ hostname, transport=Transport, ssl={'server_hostname': None}
+ ).connect()
+ Transport.Connection.assert_called_once()
+ _, kwargs = Transport.Connection.call_args
+ assert kwargs['ssl'] == {'server_hostname': 'broker.example.com'}
+
def test_register_with_event_loop(self):
t = pyamqp.Transport(Mock())
conn = Mock(name='conn')