559 lines
23 KiB
Python
559 lines
23 KiB
Python
# Copyright 2017 MongoDB, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Test retryable writes."""
|
|
|
|
import copy
|
|
import os
|
|
import sys
|
|
|
|
sys.path[0:0] = [""]
|
|
|
|
from bson.codec_options import DEFAULT_CODEC_OPTIONS
|
|
from bson.int64 import Int64
|
|
from bson.objectid import ObjectId
|
|
from bson.raw_bson import RawBSONDocument
|
|
from bson.son import SON
|
|
|
|
|
|
from pymongo.errors import (ConnectionFailure,
|
|
OperationFailure,
|
|
ServerSelectionTimeoutError,
|
|
WriteConcernError)
|
|
from pymongo.mongo_client import MongoClient
|
|
from pymongo.operations import (InsertOne,
|
|
DeleteMany,
|
|
DeleteOne,
|
|
ReplaceOne,
|
|
UpdateMany,
|
|
UpdateOne)
|
|
from pymongo.write_concern import WriteConcern
|
|
|
|
from test import unittest, client_context, IntegrationTest, SkipTest, client_knobs
|
|
from test.utils import (rs_or_single_client,
|
|
DeprecationFilter,
|
|
OvertCommandListener,
|
|
TestCreator)
|
|
from test.utils_spec_runner import SpecRunner
|
|
from test.version import Version
|
|
|
|
# Location of JSON test specifications.
|
|
_TEST_PATH = os.path.join(
|
|
os.path.dirname(os.path.realpath(__file__)), 'retryable_writes')
|
|
|
|
|
|
class TestAllScenarios(SpecRunner):
|
|
|
|
def get_object_name(self, op):
|
|
return op.get('object', 'collection')
|
|
|
|
def get_scenario_db_name(self, scenario_def):
|
|
return scenario_def.get('database_name', 'pymongo_test')
|
|
|
|
def get_scenario_coll_name(self, scenario_def):
|
|
return scenario_def.get('collection_name', 'test')
|
|
|
|
def run_test_ops(self, sessions, collection, test):
|
|
# Transform retryable writes spec format into transactions.
|
|
operation = test['operation']
|
|
outcome = test['outcome']
|
|
if 'error' in outcome:
|
|
operation['error'] = outcome['error']
|
|
if 'result' in outcome:
|
|
operation['result'] = outcome['result']
|
|
test['operations'] = [operation]
|
|
super(TestAllScenarios, self).run_test_ops(sessions, collection, test)
|
|
|
|
|
|
def create_test(scenario_def, test, name):
|
|
@client_context.require_test_commands
|
|
@client_context.require_no_mmap
|
|
def run_scenario(self):
|
|
self.run_scenario(scenario_def, test)
|
|
|
|
return run_scenario
|
|
|
|
test_creator = TestCreator(create_test, TestAllScenarios, _TEST_PATH)
|
|
test_creator.create_tests()
|
|
|
|
|
|
def _retryable_single_statement_ops(coll):
|
|
return [
|
|
(coll.bulk_write, [[InsertOne({}), InsertOne({})]], {}),
|
|
(coll.bulk_write, [[InsertOne({}),
|
|
InsertOne({})]], {'ordered': False}),
|
|
(coll.bulk_write, [[ReplaceOne({}, {})]], {}),
|
|
(coll.bulk_write, [[ReplaceOne({}, {}), ReplaceOne({}, {})]], {}),
|
|
(coll.bulk_write, [[UpdateOne({}, {'$set': {'a': 1}}),
|
|
UpdateOne({}, {'$set': {'a': 1}})]], {}),
|
|
(coll.bulk_write, [[DeleteOne({})]], {}),
|
|
(coll.bulk_write, [[DeleteOne({}), DeleteOne({})]], {}),
|
|
(coll.insert_one, [{}], {}),
|
|
(coll.insert_many, [[{}, {}]], {}),
|
|
(coll.replace_one, [{}, {}], {}),
|
|
(coll.update_one, [{}, {'$set': {'a': 1}}], {}),
|
|
(coll.delete_one, [{}], {}),
|
|
(coll.find_one_and_replace, [{}, {'a': 3}], {}),
|
|
(coll.find_one_and_update, [{}, {'$set': {'a': 1}}], {}),
|
|
(coll.find_one_and_delete, [{}, {}], {}),
|
|
]
|
|
|
|
|
|
def retryable_single_statement_ops(coll):
|
|
return _retryable_single_statement_ops(coll) + [
|
|
# Deprecated methods.
|
|
# Insert with single or multiple documents.
|
|
(coll.insert, [{}], {}),
|
|
(coll.insert, [[{}]], {}),
|
|
(coll.insert, [[{}, {}]], {}),
|
|
# Save with and without an _id.
|
|
(coll.save, [{}], {}),
|
|
(coll.save, [{'_id': ObjectId()}], {}),
|
|
# Non-multi update.
|
|
(coll.update, [{}, {'$set': {'a': 1}}], {}),
|
|
# Non-multi remove.
|
|
(coll.remove, [{}], {'multi': False}),
|
|
# Replace.
|
|
(coll.find_and_modify, [{}, {'a': 3}], {}),
|
|
# Update.
|
|
(coll.find_and_modify, [{}, {'$set': {'a': 1}}], {}),
|
|
# Delete.
|
|
(coll.find_and_modify, [{}, {}], {'remove': True}),
|
|
]
|
|
|
|
|
|
def non_retryable_single_statement_ops(coll):
|
|
return [
|
|
(coll.bulk_write, [[UpdateOne({}, {'$set': {'a': 1}}),
|
|
UpdateMany({}, {'$set': {'a': 1}})]], {}),
|
|
(coll.bulk_write, [[DeleteOne({}), DeleteMany({})]], {}),
|
|
(coll.update_many, [{}, {'$set': {'a': 1}}], {}),
|
|
(coll.delete_many, [{}], {}),
|
|
# Deprecated methods.
|
|
# Multi remove.
|
|
(coll.remove, [{}], {}),
|
|
# Multi update.
|
|
(coll.update, [{}, {'$set': {'a': 1}}], {'multi': True}),
|
|
# Unacknowledged deprecated methods.
|
|
(coll.insert, [{}], {'w': 0}),
|
|
# Unacknowledged Non-multi update.
|
|
(coll.update, [{}, {'$set': {'a': 1}}], {'w': 0}),
|
|
# Unacknowledged Non-multi remove.
|
|
(coll.remove, [{}], {'multi': False, 'w': 0}),
|
|
# Unacknowledged Replace.
|
|
(coll.find_and_modify, [{}, {'a': 3}], {'writeConcern': {'w': 0}}),
|
|
# Unacknowledged Update.
|
|
(coll.find_and_modify, [{}, {'$set': {'a': 1}}],
|
|
{'writeConcern': {'w': 0}}),
|
|
# Unacknowledged Delete.
|
|
(coll.find_and_modify, [{}, {}],
|
|
{'remove': True, 'writeConcern': {'w': 0}}),
|
|
]
|
|
|
|
|
|
class IgnoreDeprecationsTest(IntegrationTest):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(IgnoreDeprecationsTest, cls).setUpClass()
|
|
cls.deprecation_filter = DeprecationFilter()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.deprecation_filter.stop()
|
|
super(IgnoreDeprecationsTest, cls).tearDownClass()
|
|
|
|
|
|
class TestRetryableWritesMMAPv1(IgnoreDeprecationsTest):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestRetryableWritesMMAPv1, cls).setUpClass()
|
|
# Speed up the tests by decreasing the heartbeat frequency.
|
|
cls.knobs = client_knobs(heartbeat_frequency=0.1,
|
|
min_heartbeat_interval=0.1)
|
|
cls.knobs.enable()
|
|
cls.client = rs_or_single_client(retryWrites=True)
|
|
cls.db = cls.client.pymongo_test
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.knobs.disable()
|
|
cls.client.close()
|
|
|
|
@client_context.require_version_min(3, 5)
|
|
@client_context.require_no_standalone
|
|
def test_actionable_error_message(self):
|
|
if client_context.storage_engine != 'mmapv1':
|
|
raise SkipTest('This cluster is not running MMAPv1')
|
|
|
|
expected_msg = ("This MongoDB deployment does not support retryable "
|
|
"writes. Please add retryWrites=false to your "
|
|
"connection string.")
|
|
for method, args, kwargs in retryable_single_statement_ops(
|
|
self.db.retryable_write_test):
|
|
with self.assertRaisesRegex(OperationFailure, expected_msg):
|
|
method(*args, **kwargs)
|
|
|
|
|
|
class TestRetryableWrites(IgnoreDeprecationsTest):
|
|
|
|
@classmethod
|
|
@client_context.require_no_mmap
|
|
def setUpClass(cls):
|
|
super(TestRetryableWrites, cls).setUpClass()
|
|
# Speed up the tests by decreasing the heartbeat frequency.
|
|
cls.knobs = client_knobs(heartbeat_frequency=0.1,
|
|
min_heartbeat_interval=0.1)
|
|
cls.knobs.enable()
|
|
cls.listener = OvertCommandListener()
|
|
cls.client = rs_or_single_client(
|
|
retryWrites=True, event_listeners=[cls.listener])
|
|
cls.db = cls.client.pymongo_test
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.knobs.disable()
|
|
cls.client.close()
|
|
super(TestRetryableWrites, cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
if (client_context.version.at_least(3, 5) and client_context.is_rs
|
|
and client_context.test_commands_enabled):
|
|
self.client.admin.command(SON([
|
|
('configureFailPoint', 'onPrimaryTransactionalWrite'),
|
|
('mode', 'alwaysOn')]))
|
|
|
|
def tearDown(self):
|
|
if (client_context.version.at_least(3, 5) and client_context.is_rs
|
|
and client_context.test_commands_enabled):
|
|
self.client.admin.command(SON([
|
|
('configureFailPoint', 'onPrimaryTransactionalWrite'),
|
|
('mode', 'off')]))
|
|
|
|
def test_supported_single_statement_no_retry(self):
|
|
listener = OvertCommandListener()
|
|
client = rs_or_single_client(
|
|
retryWrites=False, event_listeners=[listener])
|
|
self.addCleanup(client.close)
|
|
for method, args, kwargs in retryable_single_statement_ops(
|
|
client.db.retryable_write_test):
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
listener.results.clear()
|
|
method(*args, **kwargs)
|
|
for event in listener.results['started']:
|
|
self.assertNotIn(
|
|
'txnNumber', event.command,
|
|
'%s sent txnNumber with %s' % (msg, event.command_name))
|
|
|
|
@client_context.require_version_min(3, 5)
|
|
@client_context.require_no_standalone
|
|
def test_supported_single_statement_supported_cluster(self):
|
|
for method, args, kwargs in retryable_single_statement_ops(
|
|
self.db.retryable_write_test):
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
self.listener.results.clear()
|
|
method(*args, **kwargs)
|
|
commands_started = self.listener.results['started']
|
|
self.assertEqual(len(self.listener.results['succeeded']), 1, msg)
|
|
first_attempt = commands_started[0]
|
|
self.assertIn(
|
|
'lsid', first_attempt.command,
|
|
'%s sent no lsid with %s' % (msg, first_attempt.command_name))
|
|
initial_session_id = first_attempt.command['lsid']
|
|
self.assertIn(
|
|
'txnNumber', first_attempt.command,
|
|
'%s sent no txnNumber with %s' % (
|
|
msg, first_attempt.command_name))
|
|
|
|
# There should be no retry when the failpoint is not active.
|
|
if (client_context.is_mongos or
|
|
not client_context.test_commands_enabled):
|
|
self.assertEqual(len(commands_started), 1)
|
|
continue
|
|
|
|
initial_transaction_id = first_attempt.command['txnNumber']
|
|
retry_attempt = commands_started[1]
|
|
self.assertIn(
|
|
'lsid', retry_attempt.command,
|
|
'%s sent no lsid with %s' % (msg, first_attempt.command_name))
|
|
self.assertEqual(
|
|
retry_attempt.command['lsid'], initial_session_id, msg)
|
|
self.assertIn(
|
|
'txnNumber', retry_attempt.command,
|
|
'%s sent no txnNumber with %s' % (
|
|
msg, first_attempt.command_name))
|
|
self.assertEqual(retry_attempt.command['txnNumber'],
|
|
initial_transaction_id, msg)
|
|
|
|
def test_supported_single_statement_unsupported_cluster(self):
|
|
if client_context.version.at_least(3, 5) and (
|
|
client_context.is_rs or client_context.is_mongos):
|
|
raise SkipTest('This cluster supports retryable writes')
|
|
|
|
for method, args, kwargs in retryable_single_statement_ops(
|
|
self.db.retryable_write_test):
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
self.listener.results.clear()
|
|
method(*args, **kwargs)
|
|
|
|
for event in self.listener.results['started']:
|
|
self.assertNotIn(
|
|
'txnNumber', event.command,
|
|
'%s sent txnNumber with %s' % (msg, event.command_name))
|
|
|
|
def test_unsupported_single_statement(self):
|
|
coll = self.db.retryable_write_test
|
|
coll.insert_many([{}, {}])
|
|
coll_w0 = coll.with_options(write_concern=WriteConcern(w=0))
|
|
for method, args, kwargs in (non_retryable_single_statement_ops(coll) +
|
|
retryable_single_statement_ops(coll_w0)):
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
self.listener.results.clear()
|
|
method(*args, **kwargs)
|
|
started_events = self.listener.results['started']
|
|
self.assertEqual(len(self.listener.results['succeeded']),
|
|
len(started_events), msg)
|
|
self.assertEqual(len(self.listener.results['failed']), 0, msg)
|
|
for event in started_events:
|
|
self.assertNotIn(
|
|
'txnNumber', event.command,
|
|
'%s sent txnNumber with %s' % (msg, event.command_name))
|
|
|
|
def test_server_selection_timeout_not_retried(self):
|
|
"""A ServerSelectionTimeoutError is not retried."""
|
|
listener = OvertCommandListener()
|
|
client = MongoClient(
|
|
'somedomainthatdoesntexist.org',
|
|
serverSelectionTimeoutMS=1,
|
|
retryWrites=True, event_listeners=[listener])
|
|
for method, args, kwargs in retryable_single_statement_ops(
|
|
client.db.retryable_write_test):
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
listener.results.clear()
|
|
with self.assertRaises(ServerSelectionTimeoutError, msg=msg):
|
|
method(*args, **kwargs)
|
|
self.assertEqual(len(listener.results['started']), 0, msg)
|
|
|
|
@client_context.require_version_min(3, 5)
|
|
@client_context.require_replica_set
|
|
@client_context.require_test_commands
|
|
def test_retry_timeout_raises_original_error(self):
|
|
"""A ServerSelectionTimeoutError on the retry attempt raises the
|
|
original error.
|
|
"""
|
|
listener = OvertCommandListener()
|
|
client = rs_or_single_client(
|
|
retryWrites=True, event_listeners=[listener])
|
|
self.addCleanup(client.close)
|
|
topology = client._topology
|
|
select_server = topology.select_server
|
|
|
|
def mock_select_server(*args, **kwargs):
|
|
server = select_server(*args, **kwargs)
|
|
|
|
def raise_error(*args, **kwargs):
|
|
raise ServerSelectionTimeoutError(
|
|
'No primary available for writes')
|
|
# Raise ServerSelectionTimeout on the retry attempt.
|
|
topology.select_server = raise_error
|
|
return server
|
|
|
|
for method, args, kwargs in retryable_single_statement_ops(
|
|
client.db.retryable_write_test):
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
listener.results.clear()
|
|
topology.select_server = mock_select_server
|
|
with self.assertRaises(ConnectionFailure, msg=msg):
|
|
method(*args, **kwargs)
|
|
self.assertEqual(len(listener.results['started']), 1, msg)
|
|
|
|
@client_context.require_version_min(3, 5)
|
|
@client_context.require_replica_set
|
|
@client_context.require_test_commands
|
|
def test_batch_splitting(self):
|
|
"""Test retry succeeds after failures during batch splitting."""
|
|
large = 's' * 1024 * 1024 * 15
|
|
coll = self.db.retryable_write_test
|
|
coll.delete_many({})
|
|
self.listener.results.clear()
|
|
bulk_result = coll.bulk_write([
|
|
InsertOne({'_id': 1, 'l': large}),
|
|
InsertOne({'_id': 2, 'l': large}),
|
|
InsertOne({'_id': 3, 'l': large}),
|
|
UpdateOne({'_id': 1, 'l': large},
|
|
{'$unset': {'l': 1}, '$inc': {'count': 1}}),
|
|
UpdateOne({'_id': 2, 'l': large}, {'$set': {'foo': 'bar'}}),
|
|
DeleteOne({'l': large}),
|
|
DeleteOne({'l': large})])
|
|
# Each command should fail and be retried.
|
|
# With OP_MSG 3 inserts are one batch. 2 updates another.
|
|
# 2 deletes a third.
|
|
self.assertEqual(len(self.listener.results['started']), 6)
|
|
self.assertEqual(coll.find_one(), {'_id': 1, 'count': 1})
|
|
# Assert the final result
|
|
expected_result = {
|
|
"writeErrors": [],
|
|
"writeConcernErrors": [],
|
|
"nInserted": 3,
|
|
"nUpserted": 0,
|
|
"nMatched": 2,
|
|
"nModified": 2,
|
|
"nRemoved": 2,
|
|
"upserted": [],
|
|
}
|
|
self.assertEqual(bulk_result.bulk_api_result, expected_result)
|
|
|
|
@client_context.require_version_min(3, 5)
|
|
@client_context.require_replica_set
|
|
@client_context.require_test_commands
|
|
def test_batch_splitting_retry_fails(self):
|
|
"""Test retry fails during batch splitting."""
|
|
large = 's' * 1024 * 1024 * 15
|
|
coll = self.db.retryable_write_test
|
|
coll.delete_many({})
|
|
self.client.admin.command(SON([
|
|
('configureFailPoint', 'onPrimaryTransactionalWrite'),
|
|
('mode', {'skip': 3}), # The number of _documents_ to skip.
|
|
('data', {'failBeforeCommitExceptionCode': 1})]))
|
|
self.listener.results.clear()
|
|
with self.client.start_session() as session:
|
|
initial_txn = session._server_session._transaction_id
|
|
try:
|
|
coll.bulk_write([InsertOne({'_id': 1, 'l': large}),
|
|
InsertOne({'_id': 2, 'l': large}),
|
|
InsertOne({'_id': 3, 'l': large}),
|
|
InsertOne({'_id': 4, 'l': large})],
|
|
session=session)
|
|
except ConnectionFailure:
|
|
pass
|
|
else:
|
|
self.fail("bulk_write should have failed")
|
|
|
|
started = self.listener.results['started']
|
|
self.assertEqual(len(started), 3)
|
|
self.assertEqual(len(self.listener.results['succeeded']), 1)
|
|
expected_txn = Int64(initial_txn + 1)
|
|
self.assertEqual(started[0].command['txnNumber'], expected_txn)
|
|
self.assertEqual(started[0].command['lsid'], session.session_id)
|
|
expected_txn = Int64(initial_txn + 2)
|
|
self.assertEqual(started[1].command['txnNumber'], expected_txn)
|
|
self.assertEqual(started[1].command['lsid'], session.session_id)
|
|
started[1].command.pop('$clusterTime')
|
|
started[2].command.pop('$clusterTime')
|
|
self.assertEqual(started[1].command, started[2].command)
|
|
final_txn = session._server_session._transaction_id
|
|
self.assertEqual(final_txn, expected_txn)
|
|
self.assertEqual(coll.find_one(projection={'_id': True}), {'_id': 1})
|
|
|
|
|
|
class TestWriteConcernError(IntegrationTest):
|
|
@classmethod
|
|
@client_context.require_replica_set
|
|
@client_context.require_no_mmap
|
|
@client_context.require_failCommand_fail_point
|
|
def setUpClass(cls):
|
|
super(TestWriteConcernError, cls).setUpClass()
|
|
cls.fail_insert = {
|
|
'configureFailPoint': 'failCommand',
|
|
'mode': {'times': 2},
|
|
'data': {
|
|
'failCommands': ['insert'],
|
|
'writeConcernError': {
|
|
'code': 91,
|
|
'errmsg': 'Replication is being shut down'},
|
|
}}
|
|
|
|
@client_context.require_version_min(4, 0)
|
|
def test_RetryableWriteError_error_label(self):
|
|
listener = OvertCommandListener()
|
|
client = rs_or_single_client(
|
|
retryWrites=True, event_listeners=[listener])
|
|
|
|
# Ensure collection exists.
|
|
client.pymongo_test.testcoll.insert_one({})
|
|
|
|
with self.fail_point(self.fail_insert):
|
|
with self.assertRaises(WriteConcernError) as cm:
|
|
client.pymongo_test.testcoll.insert_one({})
|
|
self.assertTrue(cm.exception.has_error_label(
|
|
'RetryableWriteError'))
|
|
|
|
if client_context.version >= Version(4, 4):
|
|
# In MongoDB 4.4+ we rely on the server returning the error label.
|
|
self.assertIn(
|
|
'RetryableWriteError',
|
|
listener.results['succeeded'][-1].reply['errorLabels'])
|
|
|
|
@client_context.require_version_min(4, 4)
|
|
def test_RetryableWriteError_error_label_RawBSONDocument(self):
|
|
# using RawBSONDocument should not cause errorLabel parsing to fail
|
|
with self.fail_point(self.fail_insert):
|
|
with self.client.start_session() as s:
|
|
s._start_retryable_write()
|
|
result = self.client.pymongo_test.command(
|
|
'insert', 'testcoll', documents=[{'_id': 1}],
|
|
txnNumber=s._server_session.transaction_id, session=s,
|
|
codec_options=DEFAULT_CODEC_OPTIONS.with_options(
|
|
document_class=RawBSONDocument))
|
|
|
|
self.assertIn('writeConcernError', result)
|
|
self.assertIn('RetryableWriteError', result['errorLabels'])
|
|
|
|
|
|
# TODO: Make this a real integration test where we stepdown the primary.
|
|
class TestRetryableWritesTxnNumber(IgnoreDeprecationsTest):
|
|
@client_context.require_version_min(3, 6)
|
|
@client_context.require_replica_set
|
|
@client_context.require_no_mmap
|
|
def test_increment_transaction_id_without_sending_command(self):
|
|
"""Test that the txnNumber field is properly incremented, even when
|
|
the first attempt fails before sending the command.
|
|
"""
|
|
listener = OvertCommandListener()
|
|
client = rs_or_single_client(
|
|
retryWrites=True, event_listeners=[listener])
|
|
topology = client._topology
|
|
select_server = topology.select_server
|
|
|
|
def raise_connection_err_select_server(*args, **kwargs):
|
|
# Raise ConnectionFailure on the first attempt and perform
|
|
# normal selection on the retry attempt.
|
|
topology.select_server = select_server
|
|
raise ConnectionFailure('Connection refused')
|
|
|
|
for method, args, kwargs in _retryable_single_statement_ops(
|
|
client.db.retryable_write_test):
|
|
listener.results.clear()
|
|
topology.select_server = raise_connection_err_select_server
|
|
with client.start_session() as session:
|
|
kwargs = copy.deepcopy(kwargs)
|
|
kwargs['session'] = session
|
|
msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
|
|
initial_txn_id = session._server_session.transaction_id
|
|
|
|
# Each operation should fail on the first attempt and succeed
|
|
# on the second.
|
|
method(*args, **kwargs)
|
|
self.assertEqual(len(listener.results['started']), 1, msg)
|
|
retry_cmd = listener.results['started'][0].command
|
|
sent_txn_id = retry_cmd['txnNumber']
|
|
final_txn_id = session._server_session.transaction_id
|
|
self.assertEqual(Int64(initial_txn_id + 1), sent_txn_id, msg)
|
|
self.assertEqual(sent_txn_id, final_txn_id, msg)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|