def test_filter_fields(self):
        docman = self.opman.doc_managers[0]
        conn = self.opman.main_connection

        include_fields = ["a", "b", "c"]
        exclude_fields = ["d", "e", "f"]

        # Set fields to care about
        self.opman.fields = include_fields
        # Documents have more than just these fields
        doc = {
            "a": 1, "b": 2, "c": 3,
            "d": 4, "e": 5, "f": 6,
            "_id": 1
        }
        db = conn['test']['test']
        db.insert(doc)
        assert_soon(lambda: db.count() == 1)
        self.opman.dump_collection()

        result = docman._search()[0]
        keys = result.keys()
        for inc, exc in zip(include_fields, exclude_fields):
            self.assertIn(inc, keys)
            self.assertNotIn(exc, keys)

    def test_update(self):
        """Test update operations."""
        # Insert
        self.conn.test.test.insert({"a": 0})
        assert_soon(lambda: sum(1 for _ in self.mongo_doc._search()) == 1)

        def check_update(update_spec):
            updated = self.conn.test.test.find_and_modify(
                {"a": 0},
                update_spec,
                new=True
            )
            # Allow some time for update to propagate
            time.sleep(2)
            replicated = self.mongo_doc.mongo.test.test.find_one({"a": 0})
            self.assertEqual(replicated, updated)

        # Update by adding a field
        check_update({"$set": {"b": [{"c": 10}, {"d": 11}]}})

        # Update by changing a value within a sub-document (contains array)
        check_update({"$inc": {"b.0.c": 1}})

        # Update by changing the value within an array
        check_update({"$inc": {"b.1.f": 12}})

        # Update by changing an entire sub-document
        check_update({"$set": {"b.0": {"e": 4}}})

        # Update by adding a sub-document
        check_update({"$set": {"b": {"0": {"c": 100}}}})

        # Update whole document
        check_update({"a": 0, "b": {"1": {"d": 10000}}})

 def test_drop_database(self):
     self.initOplogThread()
     pymongo.collection.Collection(
         self.primary_conn['test'], 'test', create=True)
     self.primary_conn.drop_database('test')
     assert_soon(lambda: len(self.docman.commands) == 2)
     self.assertEqual(self.docman.commands[1], {'dropDatabase': 1})

 def test_remove(self):
     """Tests remove operations."""
     self.conn['test']['test'].insert({'name': 'paulie'})
     assert_soon(lambda: self._count() == 1)
     self.conn['test']['test'].remove({'name': 'paulie'})
     assert_soon(lambda: self._count() != 1)
     self.assertEqual(self._count(), 0)

    def test_many_targets(self):
        """Test that one OplogThread is capable of replicating to more than
        one target.
        """
        doc_managers = [DocManager(), DocManager(), DocManager()]
        self.opman.doc_managers = doc_managers

        # start replicating
        self.opman.start()
        self.primary_conn["test"]["test"].insert({"name": "kermit", "color": "green"})
        self.primary_conn["test"]["test"].insert({"name": "elmo", "color": "firetruck red"})

        assert_soon(
            lambda: sum(len(d._search()) for d in doc_managers) == 6,
            "OplogThread should be able to replicate to multiple targets",
        )

        self.primary_conn["test"]["test"].remove({"name": "elmo"})

        assert_soon(
            lambda: sum(len(d._search()) for d in doc_managers) == 3,
            "OplogThread should be able to replicate to multiple targets",
        )
        for d in doc_managers:
            self.assertEqual(d._search()[0]["name"], "kermit")

    def test_with_orphan_documents(self):
        """Test that DocManagers have proper state after a chunk migration
        that resuts in orphaned documents.
        """
        # Start replicating to dummy doc managers
        self.opman1.start()
        self.opman2.start()

        collection = self.mongos_conn["test"]["mcsharded"]
        collection.insert({"i": i + 500} for i in range(1000))
        # Assert current state of the mongoverse
        self.assertEqual(self.shard1_conn["test"]["mcsharded"].find().count(),
                         500)
        self.assertEqual(self.shard2_conn["test"]["mcsharded"].find().count(),
                         500)
        assert_soon(lambda: len(self.opman1.doc_managers[0]._search()) == 1000)

        # Stop replication using the 'rsSyncApplyStop' failpoint
        self.shard1_conn.admin.command(
            "configureFailPoint", "rsSyncApplyStop",
            mode="alwaysOn"
        )

        # Move a chunk from shard2 to shard1
        def move_chunk():
            try:
                self.mongos_conn["admin"].command(
                    "moveChunk",
                    "test.mcsharded",
                    find={"i": 1000},
                    to="demo-set-0"
                )
            except pymongo.errors.OperationFailure:
                pass

        # moveChunk will never complete, so use another thread to continue
        mover = threading.Thread(target=move_chunk)
        mover.start()

        # wait for documents to start moving to shard 1
        assert_soon(lambda: self.shard1_conn.test.mcsharded.count() > 500)

        # Get opid for moveChunk command
        operations = self.mongos_conn.test.current_op()
        opid = None
        for op in operations["inprog"]:
            if op.get("query", {}).get("moveChunk"):
                opid = op["opid"]
        self.assertNotEqual(opid, None, "could not find moveChunk operation")
        # Kill moveChunk with the opid
        self.mongos_conn["test"]["$cmd.sys.killop"].find_one({"op": opid})

        # Mongo Connector should not become confused by unsuccessful chunk move
        docs = self.opman1.doc_managers[0]._search()
        self.assertEqual(len(docs), 1000)
        self.assertEqual(sorted(d["i"] for d in docs),
                         list(range(500, 1500)))

        # cleanup
        mover.join()

 def test_drop_collection(self):
     self.initOplogThread()
     coll = pymongo.collection.Collection(
         self.primary_conn['test'], 'test', create=True)
     coll.drop()
     assert_soon(lambda: len(self.docman.commands) == 2)
     self.assertEqual(self.docman.commands[1], {'drop': 'test'})

 def test_remove(self):
     """Tests remove
     """
     self.conn["test"]["test"].insert({"name": "paulie"})
     assert_soon(lambda: sum(1 for _ in self.solr_conn.search("*:*")) == 1)
     self.conn["test"]["test"].remove({"name": "paulie"})
     assert_soon(lambda: sum(1 for _ in self.solr_conn.search("*:*")) == 0)

    def setUp(self):
        """ Starts a new connector for every test
        """
        try:
            os.unlink("config.txt")
        except OSError:
            pass
        open("config.txt", "w").close()
        self.connector = Connector(
            address='%s:%s' % (mongo_host, self.primary_p),
            oplog_checkpoint='config.txt',
            target_url=elastic_pair,
            ns_set=['test.test'],
            u_key='_id',
            auth_key=None,
            doc_manager='mongo_connector/doc_managers/elastic_doc_manager.py',
            auto_commit_interval=0
        )
        # Clean out test databases
        try:
            self.elastic_doc._remove()
        except OperationFailed:
            try:
                # Create test.test index if necessary
                client = Elasticsearch(hosts=[elastic_pair])
                idx_client = IndicesClient(client)
                idx_client.create(index='test.test')
            except es_exceptions.TransportError:
                pass

        self.conn.test.test.drop()
        self.connector.start()
        assert_soon(lambda: len(self.connector.shard_set) > 0)
        assert_soon(lambda: sum(1 for _ in self.elastic_doc._search()) == 0)

    def test_subdocument_with_id(self):
      """
      Test inserting a document with a subdocument containing an _id property.
      In the current version of translating from document data model to property graph, the root level _id
      field is included in all subdocument nodes in the graph. This test verifies there is no error when
      inserting a document containing a subdocument with an _id property.

      See https://github.com/neo4j-contrib/neo4j_doc_manager/issues/56

      """

      doc = {
        '_id': 'root_level_object_id',
        'name': 'Bob',
        'location': {
          '_id': 'sub_document_object_id',
          'city': 'Missoula',
          'state': 'Montana'
        }
      }

      self.connector.doc_managers[0].upsert(doc, "test.test_id", 1)
      assert_soon(lambda: self._count() > 0)
      result_set = self.neo4j_conn.find_one("location")
      self.assertNotEqual(result_set, None)
      self.assertEqual(result_set['_id'], 'root_level_object_id') # expect subdocument _id to be ignored

    def test_update(self):
        """Test update operations on Solr.

        Need to have the following defined in schema.xml:

        <field name="a" type="int" indexed="true" stored="true" />
        <field name="b.0.c" type="int" indexed="true" stored="true" />
        <field name="b.10.c" type="int" indexed="true" stored="true" />
        <field name="b.0.e" type="int" indexed="true" stored="true" />
        <field name="b.1.d" type="int" indexed="true" stored="true" />
        <field name="b.1.f" type="int" indexed="true" stored="true" />
        <field name="b.2.e" type="int" indexed="true" stored="true" />
        """
        docman = self.connector.doc_managers[0]

        # Insert
        self.conn.test.test.insert({"a": 0})
        assert_soon(lambda: sum(1 for _ in docman._search("*:*")) == 1)

        def check_update(update_spec):
            updated = self.conn.test.test.find_and_modify({"a": 0}, update_spec, new=True)
            # Stringify _id to match what will be retrieved from Solr
            updated["_id"] = str(updated["_id"])
            # Flatten the MongoDB document to match Solr
            updated = docman._clean_doc(updated)
            # Allow some time for update to propagate
            time.sleep(1)
            replicated = list(docman._search("a:0"))[0]
            # Remove add'l fields until these are stored in a separate Solr core
            replicated.pop("_ts")
            replicated.pop("ns")
            # Remove field added by Solr
            replicated.pop("_version_")
            self.assertEqual(replicated, docman._clean_doc(updated))

        # Update by adding a field.
        # Note that Solr can't mix types within an array
        check_update({"$set": {"b": [{"c": 10}, {"d": 11}]}})

        # Update by setting an attribute of a sub-document beyond end of array.
        check_update({"$set": {"b.10.c": 42}})

        # Update by changing a value within a sub-document (contains array)
        check_update({"$inc": {"b.0.c": 1}})

        # Update by changing the value within an array
        check_update({"$inc": {"b.1.f": 12}})

        # Update by adding new bucket to list
        check_update({"$push": {"b": {"e": 12}}})

        # Update by replacing an entire sub-document
        check_update({"$set": {"b.0": {"e": 4}}})

        # Update by adding a sub-document
        check_update({"$set": {"b": {"0": {"c": 100}}}})

        # Update whole document
        check_update({"a": 0, "b": {"1": {"d": 10000}}})

 def test_remove_file(self):
     """Tests removing a gridfs file
     """
     fs = GridFS(self.conn['test'], 'test')
     id = fs.put("test file", filename="test.txt", encoding='utf8')
     assert_soon(lambda: sum(1 for _ in self.solr_conn.search("*:*")) == 1)
     fs.delete(id)
     assert_soon(lambda: sum(1 for _ in self.solr_conn.search("*:*")) == 0)

 def test_rename_collection(self):
     self.initOplogThread()
     coll = pymongo.collection.Collection(
         self.primary_conn['test'], 'test', create=True)
     coll.rename('test2')
     assert_soon(lambda: len(self.docman.commands) == 2)
     self.assertEqual(
         self.docman.commands[1],
         {'renameCollection': 'test.test', 'to': 'test.test2'})

    def test_remove(self):
        """Tests remove
        """

        self.conn['test']['test'].insert({'name': 'paulie'})
        assert_soon(lambda: sum(1 for _ in self.mongo_doc._search()) == 1)
        self.conn['test']['test'].remove({'name': 'paulie'})
        assert_soon(lambda: sum(1 for _ in self.mongo_doc._search()) != 1)
        self.assertEqual(sum(1 for _ in self.mongo_doc._search()), 0)

 def check_update(update_spec):
     updated = self.conn.test.command(
         SON([('findAndModify', 'test'),
              ('query', {"a": 0}),
              ('update', update_spec),
              ('new', True)]))['value']
     # Stringify _id to match what will be retrieved from ES
     updated['_id'] = str(updated['_id'])
     assert_soon(lambda: next(self._search()) == updated)

 def test_bad_int_value(self):
     self.conn.test.test.insert_one({
         'inf': float('inf'), 'nan': float('nan'),
         'still_exists': True})
     assert_soon(lambda: self._count() > 0)
     for doc in self._search():
         self.assertNotIn('inf', doc)
         self.assertNotIn('nan', doc)
         self.assertTrue(doc['still_exists'])

    def test_single_target(self):
        """Test with a single replication target"""

        self.opman.start()

        # Insert first document with primary up
        self.main_conn["test"]["mc"].insert({"i": 0})
        self.assertEqual(self.primary_conn["test"]["mc"].find().count(), 1)

        # Make sure the insert is replicated
        secondary = self.secondary_conn
        assert_soon(lambda: secondary["test"]["mc"].count() == 1,
                    "first write didn't replicate to secondary")

        # Kill the primary
        kill_mongo_proc(self.primary_p, destroy=False)

        # Wait for the secondary to be promoted
        while not secondary["admin"].command("isMaster")["ismaster"]:
            time.sleep(1)

        # Insert another document. This will be rolled back later
        retry_until_ok(self.main_conn["test"]["mc"].insert, {"i": 1})
        self.assertEqual(secondary["test"]["mc"].count(), 2)

        # Wait for replication to doc manager
        assert_soon(lambda: len(self.opman.doc_managers[0]._search()) == 2,
                    "not all writes were replicated to doc manager")

        # Kill the new primary
        kill_mongo_proc(self.secondary_p, destroy=False)

        # Start both servers back up
        restart_mongo_proc(self.primary_p)
        primary_admin = self.primary_conn["admin"]
        assert_soon(lambda: primary_admin.command("isMaster")["ismaster"],
                    "restarted primary never resumed primary status")
        restart_mongo_proc(self.secondary_p)
        assert_soon(lambda: retry_until_ok(secondary.admin.command,
                                           'replSetGetStatus')['myState'] == 2,
                    "restarted secondary never resumed secondary status")
        assert_soon(lambda:
                    retry_until_ok(self.main_conn.test.mc.find().count) > 0,
                    "documents not found after primary/secondary restarted")

        # Only first document should exist in MongoDB
        self.assertEqual(self.main_conn["test"]["mc"].count(), 1)
        self.assertEqual(self.main_conn["test"]["mc"].find_one()["i"], 0)

        # Same case should hold for the doc manager
        doc_manager = self.opman.doc_managers[0]
        self.assertEqual(len(doc_manager._search()), 1)
        self.assertEqual(doc_manager._search()[0]["i"], 0)

        # cleanup
        self.opman.join()

 def test_insert(self):
     """Test insert operations."""
     self.conn['test']['test'].insert({'name': 'paulie'})
     assert_soon(lambda: self._count() > 0)
     result_set_1 = list(self._search())
     self.assertEqual(len(result_set_1), 1)
     result_set_2 = self.conn['test']['test'].find_one()
     for item in result_set_1:
         self.assertEqual(item['_id'], str(result_set_2['_id']))
         self.assertEqual(item['name'], result_set_2['name'])

    def test_stressed_rollback(self):
        """Test stressed rollback with number of documents equal to specified
            in global variable. Strategy for rollback is the same as before.
        """

        for i in range(0, STRESS_COUNT):
            self.conn['test']['test'].insert({'name': 'Paul ' + str(i)})

        search = self.mongo_doc._search
        condition = lambda: sum(1 for _ in search()) == STRESS_COUNT
        assert_soon(condition)
        primary_conn = MongoClient(mongo_host, self.primary_p)

        kill_mongo_proc(self.primary_p, destroy=False)

        new_primary_conn = MongoClient(mongo_host, self.secondary_p)

        admin = new_primary_conn['admin']
        assert_soon(lambda: admin.command("isMaster")['ismaster'])

        time.sleep(5)
        count = -1
        while count + 1 < STRESS_COUNT:
            try:
                count += 1
                self.conn['test']['test'].insert(
                    {'name': 'Pauline ' + str(count)})
            except (OperationFailure, AutoReconnect):
                time.sleep(1)
        assert_soon(lambda: sum(1 for _ in self.mongo_doc._search())
                    == self.conn['test']['test'].find().count())
        result_set_1 = self.mongo_doc._search()
        for item in result_set_1:
            if 'Pauline' in item['name']:
                result_set_2 = self.conn['test']['test'].find_one(
                    {'name': item['name']})
                self.assertEqual(item['_id'], result_set_2['_id'])

        kill_mongo_proc(self.secondary_p, destroy=False)

        restart_mongo_proc(self.primary_p)
        db_admin = primary_conn['admin']
        assert_soon(lambda: db_admin.command("isMaster")['ismaster'])
        restart_mongo_proc(self.secondary_p)

        search = self.mongo_doc._search
        condition = lambda: sum(1 for _ in search()) == STRESS_COUNT
        assert_soon(condition)

        result_set_1 = list(self.mongo_doc._search())
        self.assertEqual(len(result_set_1), STRESS_COUNT)
        for item in result_set_1:
            self.assertTrue('Paul' in item['name'])
        find_cursor = retry_until_ok(self.conn['test']['test'].find)
        self.assertEqual(retry_until_ok(find_cursor.count), STRESS_COUNT)

 def test_remove(self):
   """Tests remove operations."""
   self.conn['test']['test'].insert({'name': 'paulie'})
   result_set = self.conn['test']['test'].find_one()
   # self.connector.doc_managers[0].upsert({'_id': str(result_set['_id']),'name': 'paulie'}, "test.test", 1)
   assert_soon(lambda: self._count() == 1)
   self.conn['test']['test'].remove({'name': 'paulie'})
   assert_soon(lambda: self._count() != 1)
   self.connector.doc_managers[0].remove(str(result_set['_id']), 'test.test', 1)
   self.assertEqual(self._count(), 0)
   self.connector.doc_managers[0].graph.delete_all()