Skip to content

Commit 92a8230

Browse files
committed
PYTHON-5272 Add session reuse unit test for _configured_socket_interface
Verify that a pre-populated _SSLSessionCache passes the cached session as session= to wrap_socket on the next connection, using mocks so no live server is required.
1 parent 150eb37 commit 92a8230

2 files changed

Lines changed: 64 additions & 0 deletions

File tree

test/asynchronous/test_ssl.py

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,38 @@ def test_ssl_session_cache(self):
138138
cache.set("new_session")
139139
self.assertEqual(cache.get(), "new_session")
140140

141+
@unittest.skipUnless(_IS_SYNC, "Session reuse only applies to sync path")
142+
def test_tls_session_reused_on_second_connection(self):
143+
"""Cached TLS session is passed to wrap_socket on subsequent connections."""
144+
import unittest.mock as mock
145+
146+
from pymongo.pool_shared import _configured_socket_interface, _SSLSessionCache
147+
148+
fake_session = object()
149+
cache = _SSLSessionCache()
150+
cache.set(fake_session)
151+
152+
fake_ssl_sock = mock.MagicMock()
153+
fake_ssl_sock.getpeercert.return_value = {}
154+
155+
mock_ssl_context = mock.MagicMock()
156+
mock_ssl_context.wrap_socket.return_value = fake_ssl_sock
157+
mock_ssl_context.verify_mode = False
158+
mock_ssl_context.check_hostname = False
159+
160+
mock_opts = mock.MagicMock()
161+
mock_opts._ssl_context = mock_ssl_context
162+
mock_opts.socket_timeout = None
163+
mock_opts.tls_allow_invalid_hostnames = True
164+
165+
with mock.patch("pymongo.pool_shared._create_connection") as mock_create:
166+
mock_create.return_value = mock.MagicMock()
167+
_configured_socket_interface(("localhost", 27017), mock_opts, cache)
168+
169+
mock_ssl_context.wrap_socket.assert_called_once()
170+
_, kwargs = mock_ssl_context.wrap_socket.call_args
171+
self.assertIs(kwargs.get("session"), fake_session)
172+
141173

142174
class TestSSL(AsyncIntegrationTest):
143175
saved_port: int

test/test_ssl.py

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,38 @@ def test_ssl_session_cache(self):
138138
cache.set("new_session")
139139
self.assertEqual(cache.get(), "new_session")
140140

141+
@unittest.skipUnless(_IS_SYNC, "Session reuse only applies to sync path")
142+
def test_tls_session_reused_on_second_connection(self):
143+
"""Cached TLS session is passed to wrap_socket on subsequent connections."""
144+
import unittest.mock as mock
145+
146+
from pymongo.pool_shared import _configured_socket_interface, _SSLSessionCache
147+
148+
fake_session = object()
149+
cache = _SSLSessionCache()
150+
cache.set(fake_session)
151+
152+
fake_ssl_sock = mock.MagicMock()
153+
fake_ssl_sock.getpeercert.return_value = {}
154+
155+
mock_ssl_context = mock.MagicMock()
156+
mock_ssl_context.wrap_socket.return_value = fake_ssl_sock
157+
mock_ssl_context.verify_mode = False
158+
mock_ssl_context.check_hostname = False
159+
160+
mock_opts = mock.MagicMock()
161+
mock_opts._ssl_context = mock_ssl_context
162+
mock_opts.socket_timeout = None
163+
mock_opts.tls_allow_invalid_hostnames = True
164+
165+
with mock.patch("pymongo.pool_shared._create_connection") as mock_create:
166+
mock_create.return_value = mock.MagicMock()
167+
_configured_socket_interface(("localhost", 27017), mock_opts, cache)
168+
169+
mock_ssl_context.wrap_socket.assert_called_once()
170+
_, kwargs = mock_ssl_context.wrap_socket.call_args
171+
self.assertIs(kwargs.get("session"), fake_session)
172+
141173

142174
class TestSSL(IntegrationTest):
143175
saved_port: int

0 commit comments

Comments
 (0)