def test_init_disconnected(self):
        c = MongoClient(host, port, _connect=False)

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertIsInstance(c.is_primary, bool)
            self.assertIsInstance(c.is_mongos, bool)
            self.assertIsInstance(c.max_pool_size, int)
            self.assertIsInstance(c.use_greenlets, bool)
            self.assertIsInstance(c.nodes, frozenset)
            self.assertIsInstance(c.auto_start_request, bool)
            self.assertEqual(dict, c.get_document_class())
            self.assertIsInstance(c.tz_aware, bool)
            self.assertIsInstance(c.max_bson_size, int)
            self.assertIsInstance(c.min_wire_version, int)
            self.assertIsInstance(c.max_wire_version, int)
            self.assertIsInstance(c.max_write_batch_size, int)
            self.assertEqual(None, c.host)
            self.assertEqual(None, c.port)
        finally:
            ctx.exit()

        c.pymongo_test.test.find_one()  # Auto-connect.
        self.assertEqual((host, port), c.address)

        if version.at_least(c, (2, 5, 4, -1)):
            self.assertTrue(c.max_wire_version > 0)
        else:
            self.assertEqual(c.max_wire_version, 0)
        self.assertTrue(c.min_wire_version >= 0)

        bad_host = "somedomainthatdoesntexist.org"
        c = MongoClient(bad_host, port, connectTimeoutMS=1, _connect=False)
        self.assertRaises(ConnectionFailure, c.pymongo_test.test.find_one)

    def test_unix_socket(self):
        if not hasattr(socket, "AF_UNIX"):
            raise SkipTest("UNIX-sockets are not supported on this system")

        mongodb_socket = '/tmp/mongodb-27017.sock'
        encoded_socket = '%2Ftmp%2Fmongodb-27017.sock'
        if not os.access(mongodb_socket, os.R_OK):
            raise SkipTest("Socket file is not accessible")

        if client_context.auth_enabled:
            uri = "mongodb://%s:%[email protected]%s" % (db_user, db_pwd, encoded_socket)
        else:
            uri = "mongodb://%s" % encoded_socket

        # Confirm we can do operations via the socket.
        client = MongoClient(uri)
        client.pymongo_test.test.insert_one({"dummy": "object"})
        dbs = client.database_names()
        self.assertTrue("pymongo_test" in dbs)

        # Confirm it fails with a missing socket.
        self.assertRaises(
            ConnectionFailure,
            connected, MongoClient("mongodb://%2Ftmp%2Fnon-existent.sock",
                                   serverSelectionTimeoutMS=100))

    def test_ipv6(self):
        c = MongoClient("mongodb://[::1]:%d" % (port,), replicaSet=self.name)

        # Client switches to IPv4 once it has first ismaster response.
        msg = 'discovered primary with IPv4 address "%r"' % (self.primary,)
        wait_until(lambda: c.primary == self.primary, msg)

        # Same outcome with both IPv4 and IPv6 seeds.
        c = MongoClient("[::1]:%d,localhost:%d" % (port, port),
                        replicaSet=self.name)

        wait_until(lambda: c.primary == self.primary, msg)

        if client_context.auth_enabled:
            auth_str = "%s:%[email protected]" % (db_user, db_pwd)
        else:
            auth_str = ""

        uri = "mongodb://%slocalhost:%d,[::1]:%d" % (auth_str, port, port)
        client = MongoClient(uri, replicaSet=self.name)
        client.pymongo_test.test.insert_one({"dummy": u("object")})
        client.pymongo_test_bernie.test.insert_one({"dummy": u("object")})

        dbs = client.database_names()
        self.assertTrue("pymongo_test" in dbs)
        self.assertTrue("pymongo_test_bernie" in dbs)
        client.close()

    def test_max_idle_time_checkout(self):
        # Use high frequency to test _get_socket_no_auth.
        with client_knobs(kill_cursor_frequency=99999999):
            client = MongoClient(host, port, maxIdleTimeMS=.5)
            server = client._get_topology().select_server(any_server_selector)
            with server._pool.get_socket({}) as sock_info:
                pass
            self.assertEqual(1, len(server._pool.sockets))
            time.sleep(1) #  Sleep so that the socket becomes stale.

            with server._pool.get_socket({}) as new_sock_info:
                self.assertNotEqual(sock_info, new_sock_info)
            self.assertEqual(1, len(server._pool.sockets))
            self.assertFalse(sock_info in server._pool.sockets)
            self.assertTrue(new_sock_info in server._pool.sockets)

            # Test that sockets are reused if maxIdleTimeMS is not set.
            client = MongoClient(host, port)
            server = client._get_topology().select_server(any_server_selector)
            with server._pool.get_socket({}) as sock_info:
                pass
            self.assertEqual(1, len(server._pool.sockets))
            time.sleep(1)
            with server._pool.get_socket({}) as new_sock_info:
                self.assertEqual(sock_info, new_sock_info)
            self.assertEqual(1, len(server._pool.sockets))

def stash(results):
    """
    暂存到mongo数据库中。
    """
    summary = {}
    mongo = MongoClient(**config.mongo)
    try:
        for item_model, objs in results:
            collection_name = item_model['name']
            db = mongo.get_database('theforce')
            collection = db.get_collection(collection_name)
            collection.insert_many(objs)
            summary[collection_name] = len(
                objs) if collection_name not in summary else len(
                    objs) + summary[collection_name]

        print
        print "=" * 40
        print ' ' * 15, u'Stash'
        print "=" * 40
        print
        print u"数据已成功保存到MongoDB的theforce库中，其中新增数据:"
        for name, length in summary.items():
            print name, length
    finally:
        mongo.close()

    def test_unix_socket(self):
        if not hasattr(socket, "AF_UNIX"):
            raise SkipTest("UNIX-sockets are not supported on this system")
        client = MongoClient(host, port)
        if (sys.platform == 'darwin' and
                server_started_with_auth(client) and
                not version.at_least(client, (2, 7, 1))):
            raise SkipTest("SERVER-8492")

        mongodb_socket = '/tmp/mongodb-27017.sock'
        if not os.access(mongodb_socket, os.R_OK):
            raise SkipTest("Socket file is not accessable")

        self.assertTrue(MongoClient("mongodb://%s" % mongodb_socket))

        client = MongoClient("mongodb://%s" % mongodb_socket)
        client.pymongo_test.test.save({"dummy": "object"})

        # Confirm we can read via the socket
        dbs = client.database_names()
        self.assertTrue("pymongo_test" in dbs)

        # Confirm it fails with a missing socket
        self.assertRaises(ConnectionFailure, MongoClient,
                          "mongodb:///tmp/none-existent.sock")

    def test_document_class(self):
        c = MongoClient(host, port)
        db = c.pymongo_test
        db.test.insert({"x": 1})

        self.assertEqual(dict, c.document_class)
        self.assertTrue(isinstance(db.test.find_one(), dict))
        self.assertFalse(isinstance(db.test.find_one(), SON))

        c.document_class = SON

        self.assertEqual(SON, c.document_class)
        self.assertTrue(isinstance(db.test.find_one(), SON))
        self.assertFalse(isinstance(db.test.find_one(as_class=dict), SON))

        c = MongoClient(host, port, document_class=SON)
        db = c.pymongo_test

        self.assertEqual(SON, c.document_class)
        self.assertTrue(isinstance(db.test.find_one(), SON))
        self.assertFalse(isinstance(db.test.find_one(as_class=dict), SON))

        c.document_class = dict

        self.assertEqual(dict, c.document_class)
        self.assertTrue(isinstance(db.test.find_one(), dict))
        self.assertFalse(isinstance(db.test.find_one(), SON))

    def test_document_class(self):
        c = MongoClient(host, port)
        db = c.pymongo_test
        db.test.insert({"x": 1})

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertEqual(dict, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), dict))
            self.assertFalse(isinstance(db.test.find_one(), SON))

            c.document_class = SON
            db = c.pymongo_test

            self.assertEqual(SON, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), SON))
            self.assertFalse(isinstance(db.test.find_one(as_class=dict), SON))

            c = MongoClient(host, port, document_class=SON)
            db = c.pymongo_test

            self.assertEqual(SON, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), SON))
            self.assertFalse(isinstance(db.test.find_one(as_class=dict), SON))

            c.document_class = dict
            db = c.pymongo_test

            self.assertEqual(dict, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), dict))
            self.assertFalse(isinstance(db.test.find_one(), SON))
        finally:
            ctx.exit()

 def __init__(self):
     connection = MongoClient(
         settings['MONGO_SERVER'],
         settings['MONGO_PORT']
     )
     db = connection.get_database(settings['MONGO_DB'])
     self.collection = db[settings['MONGO_COLLECTION']]

	def open_spider(self, spider):
		self._build_unique_key()

		if self._replica_set is not None:
			self.connection = MongoReplicaSetClient(
				self._uri,
				replicaSet=self._replica_set,
				w=self._write_concern,
				fsync=self._fsync,
				read_preference=ReadPreference.PRIMARY_PREFERRED)
		else:
			self.connection = MongoClient(
				self._uri,
				fsync=self._fsync,
				read_preference=ReadPreference.PRIMARY)

		self.database = self.connection[self._database]
		self.collection = self.database[self._collection]

		log.msg('Connected to MongoDB "%s", using "%s/%s"' %
			self._uri, self._database, self._collection)

		# ensure index
		if self._unique_key:
			log.msg('Creating index for key %s' % self._unique_key)
			self.collection.ensure_index(self._unique_key.items(), unique=True, sparse=True)

    def test_init_disconnected(self):
        c = MongoClient(host, port, _connect=False)

        self.assertIsInstance(c.is_primary, bool)
        self.assertIsInstance(c.is_mongos, bool)
        self.assertIsInstance(c.max_pool_size, int)
        self.assertIsInstance(c.use_greenlets, bool)
        self.assertIsInstance(c.nodes, frozenset)
        self.assertIsInstance(c.auto_start_request, bool)
        self.assertEqual(dict, c.get_document_class())
        self.assertIsInstance(c.tz_aware, bool)
        self.assertIsInstance(c.max_bson_size, int)
        self.assertIsInstance(c.min_wire_version, int)
        self.assertIsInstance(c.max_wire_version, int)
        self.assertEqual(None, c.host)
        self.assertEqual(None, c.port)

        c.pymongo_test.test.find_one()  # Auto-connect.
        self.assertEqual(host, c.host)
        self.assertEqual(port, c.port)

        if version.at_least(c, (2, 5, 4, -1)):
            self.assertTrue(c.max_wire_version > 0)
        else:
            self.assertEqual(c.max_wire_version, 0)
        self.assertTrue(c.min_wire_version >= 0)

        bad_host = "somedomainthatdoesntexist.org"
        c = MongoClient(bad_host, port, connectTimeoutMS=1, _connect=False)
        self.assertRaises(ConnectionFailure, c.pymongo_test.test.find_one)

    def test_read_with_failover(self):
        c = MongoClient(
            self.seed,
            replicaSet=self.name,
            serverSelectionTimeoutMS=self.server_selection_timeout)
        wait_until(lambda: c.primary, "discover primary")
        wait_until(lambda: len(c.secondaries) == 2, "discover secondaries")

        def iter_cursor(cursor):
            for _ in cursor:
                pass
            return True

        w = len(c.secondaries) + 1
        db = c.get_database("pymongo_test",
                            write_concern=WriteConcern(w=w))
        db.test.delete_many({})
        # Force replication
        db.test.insert_many([{'foo': i} for i in xrange(10)])
        self.assertEqual(10, db.test.count())

        db.read_preference = SECONDARY_PREFERRED
        cursor = db.test.find().batch_size(5)
        next(cursor)
        self.assertEqual(5, cursor._Cursor__retrieved)
        self.assertTrue(cursor.address in c.secondaries)
        ha_tools.kill_primary()
        # Primary failure shouldn't interrupt the cursor
        self.assertTrue(iter_cursor(cursor))
        self.assertEqual(10, cursor._Cursor__retrieved)

def mongo_text_test():
    cli = MongoClient()
    test = cli.get_database("test").get_collection("test_search")
    test.create_index(
          [("super_type", ASCENDING), ("resource_state", ASCENDING),
           ("uuid", TEXT), ("name", TEXT), ("description", TEXT)],
          background=True
      )
    docs = [{
        "uuid": str(uuid.uuid4()),
        "name": str(i),
        "description": "Nature, \"time, and patience are "
        "the three great -physicians.",
        "super_type": "super_vol",
        "type": "vol",
        "create_time": int(
        time.time()),
        "resource_state": "inUse"
    } for i in range(1, 11)]
    test.insert_many(docs)

    text = [" -physicians"]
    print test.find({
                "description": {
                    "$regex":  "|".join([re.sub(
                        r"(\*|\.|\?|\+|\$|\^|\[|\]|\(|\)|\{|\}|\||\\|/)",
                        r"\\\1",
                        g
                    ) for g in text]),
                    "$options": "i"
                }
            }).count()

    def test_database_names(self):
        client = MongoClient(host, port)

        client.pymongo_test.test.save({"dummy": u"object"})
        client.pymongo_test_mike.test.save({"dummy": u"object"})

        dbs = client.database_names()
        self.assertTrue("pymongo_test" in dbs)
        self.assertTrue("pymongo_test_mike" in dbs)

def doEverything():
#     certfile = '/home/bryan/Downloads/baratheon.pem'
    conn = MongoClient(url)
    db = conn[database]
    
    commands = []
    collectionName = "pythonMongo"
    commands.append("Creating collection " + collectionName)
    collection = db[collectionName]
    
    #insert 1
    commands.append("# 1 Inserts")
    commands.append("# 1.1 Insert a single document to a collection")
    collection.insert({"name": "test1", "value": 1})
    commands.append("Inserted {\"name\": \"test1\", \"value\": 1}")
    
    #insert many
    commands.append("#1.2 Inserting multiple entries into collection")
    multiPost = [{"name": "test1", "value": 1},{"name": "test2", "value": 2}, {"name": "test3", "value": 3}] 
    collection.insert(multiPost)
    commands.append("Inserted \n {\"name\": \"test1\", \"value\": 1} \n {\"name\": \"test2\", \"value\": 2} \n {\"name\": \"test3\", \"value\": 3}")
     
    # Find 
    commands.append("#2 Queries")
    commands.append("#2.1 Find one that matches a query condition")
    commands.append(collection.find_one({"name": "test1"}))
     
    # Find all 
    commands.append("#2.2 Find all that match a query condition")
    for doc in collection.find({"name": "test1"}):
        commands.append(doc)
    
    # Display all documents
    commands.append( "#2.3 Find all documents in collection")
    for doc in collection.find():
        commands.append(doc)   
    
    # update document
    commands.append("#3 Updating Documents")
    collection.update({"name": "test3"}, {"$set": { "value": 4}})
    commands.append("Updated test3 with value 4")
     
    # delete document
    commands.append("#4 Delete Documents")
    collection.remove({"name": "test2"})  
    commands.append("Deleted all with name test2")
    
    # Display all collection names
    commands.append("#5 Get a list of all of the collections")
    commands.append( db.collection_names())
    
    commands.append("#6 Drop a collection")
    db.drop_collection(collectionName)
    conn.close()
    commands.append("Connection to database has been closed")
    return commands

    def test_alive(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertTrue(get_client().alive())

            client = MongoClient('doesnt exist', _connect=False)
            self.assertFalse(client.alive())
        finally:
            ctx.exit()

def makeDBConnection(port, **kwargs):
    global _C
    if not _C:
        logging.info("establishing db connection at port %s ..." % port)
        import pymongo
        logging.info("using pymongo version %s" % pymongo.version)
        from pymongo.mongo_client import MongoClient
        _C = MongoClient(port = port,  **kwargs)
        mongo_info = _C.server_info()
        logging.info("mongodb version: %s" % mongo_info["version"])

 def test_stale_getmore(self):
     # A cursor is created, but its member goes down and is removed from
     # the topology before the getMore message is sent. Test that
     # MongoClient._send_message_with_response handles the error.
     with self.assertRaises(AutoReconnect):
         client = MongoClient(host, port, connect=False,
                              serverSelectionTimeoutMS=100,
                              replicaSet=client_context.replica_set_name)
         client._send_message_with_response(
             operation=message._GetMore('collection', 101, 1234),
             address=('not-a-member', 27017))

    def test_disconnect(self):
        c = MongoClient(host, port)
        coll = c.pymongo_test.bar

        c.disconnect()
        c.disconnect()

        coll.count()

        c.disconnect()
        c.disconnect()

        coll.count()

 def test_mongo(self):
     client = MongoClient()
     db = client.test_database
     collection = db.test_collection
     import datetime
     post = {"author": "Mike",
             "text": "My first blog post!",
             "tags": ["mongodb", "python", "pymongo"],
             "date": datetime.datetime.utcnow()}
     posts = db.posts
     post_id = posts.insert(post)
     print(post_id)
     print(client.server_info())