mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PYTHON-5399 Add a prose test for OIDC reauthentication when a session is involved (#2351)

Browse Source
This commit is contained in:
Steven Silvester 2025-06-05 09:21:10 -05:00 committed by GitHub
parent 454c163788
commit 6d33d4fb34
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 38 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
test/asynchronous/test_auth_oidc.py
View File

@ -1085,6 +1085,25 @@ class TestAuthOIDCMachine(OIDCTestBase):
# Assert there were `SaslStart` commands executed.
assert any(event.command_name.lower() == "saslstart" for event in listener.started_events)
async def test_4_5_reauthentication_succeeds_when_a_session_is_involved(self):
# Create an OIDC configured client.
client = await self.create_client()
# Set a fail point for `find` commands of the form:
async with self.fail_point(
{
"mode": {"times": 1},
"data": {"failCommands": ["find"], "errorCode": 391},
}
):
# Start a new session.
async with client.start_session() as session:
# In the started session perform a `find` operation that succeeds.
await client.test.test.find_one({}, session=session)
# Assert that the callback was called 2 times (once during the connection handshake, and again during reauthentication).
self.assertEqual(self.request_called, 2)
async def test_5_1_azure_with_no_username(self):
if ENVIRON != "azure":
raise unittest.SkipTest("Test is only supported on Azure")

19
test/test_auth_oidc.py
View File

@ -1083,6 +1083,25 @@ class TestAuthOIDCMachine(OIDCTestBase):
# Assert there were `SaslStart` commands executed.
assert any(event.command_name.lower() == "saslstart" for event in listener.started_events)
def test_4_5_reauthentication_succeeds_when_a_session_is_involved(self):
# Create an OIDC configured client.
client = self.create_client()
# Set a fail point for `find` commands of the form:
with self.fail_point(
{
"mode": {"times": 1},
"data": {"failCommands": ["find"], "errorCode": 391},
}
):
# Start a new session.
with client.start_session() as session:
# In the started session perform a `find` operation that succeeds.
client.test.test.find_one({}, session=session)
# Assert that the callback was called 2 times (once during the connection handshake, and again during reauthentication).
self.assertEqual(self.request_called, 2)
def test_5_1_azure_with_no_username(self):
if ENVIRON != "azure":
raise unittest.SkipTest("Test is only supported on Azure")