'Test the rollback method in a sharded environment
Cases:
1. Documents on both shards, rollback on one shard
2. Documents on both shards, rollback on both shards'

def test_rollback(self):

    self.opman1.start()
    self.opman2.start()
    db_main = self.mongos_conn['test']['mcsharded']
    db_main2 = db_main.with_options(write_concern=WriteConcern(w=2))
    db_main2.insert_one({'i': 0})
    db_main2.insert_one({'i': 1000})
    self.assertEqual(self.shard1_conn['test']['mcsharded'].count(), 1)
    self.assertEqual(self.shard2_conn['test']['mcsharded'].count(), 1)
    self.cluster.shards[0].primary.stop(destroy=False)
    shard1_secondary_admin = self.shard1_secondary_conn['admin']
    assert_soon((lambda : shard1_secondary_admin.command('isMaster')['ismaster']))
    def insert_one(doc):
        if (not db_main.find_one(doc)):
            return db_main.insert_one(doc)
        return True
    assert_soon((lambda : retry_until_ok(insert_one, {'i': 1})), 'could    not    insert    into    shard1    with    one    node    down')
    db_secondary1 = self.shard1_secondary_conn['test']['mcsharded']
    db_secondary2 = self.shard2_secondary_conn['test']['mcsharded']
    self.assertEqual(db_secondary1.count(), 2)
    c = (lambda : (len(self.opman1.doc_managers[0]._search()) == 3))
    assert_soon(c, 'not    all    writes    were    replicated    to    doc    manager', max_tries=120)
    self.cluster.shards[0].secondary.stop(destroy=False)
    self.cluster.shards[0].primary.start()
    primary_admin = self.shard1_conn['admin']
    c = (lambda : primary_admin.command('isMaster')['ismaster'])
    assert_soon((lambda : retry_until_ok(c)))
    self.cluster.shards[0].secondary.start()
    secondary_admin = self.shard1_secondary_conn['admin']
    c = (lambda : (secondary_admin.command('replSetGetStatus')['myState'] == 2))
    assert_soon(c)
    query = {'i': {'$lt': 1000}}
    assert_soon((lambda : (retry_until_ok(db_main.find(query).count) > 0)))
    self.assertEqual(db_main.find(query).count(), 1)
    self.assertEqual(db_main.find_one(query)['i'], 0)
    def check_docman_rollback():
        docman_docs = [d for d in self.opman1.doc_managers[0]._search() if (d['i'] < 1000)]
        return ((len(docman_docs) == 1) and (docman_docs[0]['i'] == 0))
    assert_soon(check_docman_rollback, 'doc    manager    did    not    roll    back')
    db_main.insert_one({'i': (-1)})
    db_main.delete_one({'i': (-1)})
    def rollback_done():
        secondary1_count = retry_until_ok(db_secondary1.count)
        secondary2_count = retry_until_ok(db_secondary2.count)
        return ((1, 1) == (secondary1_count, secondary2_count))
    assert_soon(rollback_done, 'rollback    never    replicated    to    one    or    more    secondaries')
    self.cluster.shards[0].primary.stop(destroy=False)
    self.cluster.shards[1].primary.stop(destroy=False)
    shard1_secondary_admin = self.shard1_secondary_conn['admin']
    shard2_secondary_admin = self.shard2_secondary_conn['admin']
    assert_soon((lambda : shard1_secondary_admin.command('isMaster')['ismaster']))
    assert_soon((lambda : shard2_secondary_admin.command('isMaster')['ismaster']))
    assert_soon((lambda : retry_until_ok(insert_one, {'i': 1})), 'could    not    insert    into    shard1    with    one    node    down')
    self.assertEqual(db_secondary1.count(), 2)
    assert_soon((lambda : retry_until_ok(insert_one, {'i': 1001})), 'could    not    insert    into    shard2    with    one    node    down')
    self.assertEqual(db_secondary2.count(), 2)
    c = (lambda : (len(self.opman1.doc_managers[0]._search()) == 4))
    assert_soon(c, 'not    all    writes    were    replicated    to    doc    manager')
    self.cluster.shards[0].secondary.stop(destroy=False)
    self.cluster.shards[1].secondary.stop(destroy=False)
    self.cluster.shards[0].primary.start()
    c = (lambda : self.shard1_conn['admin'].command('isMaster')['ismaster'])
    assert_soon((lambda : retry_until_ok(c)))
    self.cluster.shards[0].secondary.start()
    secondary_admin = self.shard1_secondary_conn['admin']
    c = (lambda : (secondary_admin.command('replSetGetStatus')['myState'] == 2))
    assert_soon(c)
    self.cluster.shards[1].primary.start()
    c = (lambda : self.shard2_conn['admin'].command('isMaster')['ismaster'])
    assert_soon((lambda : retry_until_ok(c)))
    self.cluster.shards[1].secondary.start()
    secondary_admin = self.shard2_secondary_conn['admin']
    c = (lambda : (secondary_admin.command('replSetGetStatus')['myState'] == 2))
    assert_soon(c)
    assert_soon((lambda : (retry_until_ok(db_main.find(query).count) > 0)))
    query2 = {'i': {'$gte': 1000}}
    assert_soon((lambda : (retry_until_ok(db_main.find(query2).count) > 0)))
    self.assertEqual(db_main.find(query).count(), 1)
    self.assertEqual(db_main.find_one(query)['i'], 0)
    self.assertEqual(db_main.find(query2).count(), 1)
    self.assertEqual(db_main.find_one(query2)['i'], 1000)
    assert_soon((lambda : (len(self.opman1.doc_managers[0]._search()) == 2)))
    i_values = [d['i'] for d in self.opman1.doc_managers[0]._search()]
    self.assertIn(0, i_values)
    self.assertIn(1000, i_values)
