def test_connection_kwargs(self):
        """Ensure that connection kwargs get passed to pymongo."""
        connect('mongoenginetest', alias='t1', tz_aware=True)
        conn = get_connection('t1')

        self.assertTrue(get_tz_awareness(conn))

        connect('mongoenginetest2', alias='t2')
        conn = get_connection('t2')
        self.assertFalse(get_tz_awareness(conn))

    def test_sharing_connections(self):
        """Ensure that connections are shared when the connection settings are exactly the same
        """
        connect('mongoenginetest', alias='testdb1')

        expected_connection = get_connection('testdb1')

        connect('mongoenginetest', alias='testdb2')
        actual_connection = get_connection('testdb2')
        self.assertEqual(expected_connection, actual_connection)

    def test_connection_kwargs(self):
        """Ensure that connection kwargs get passed to pymongo.
        """
        connect("mongoenginetest", alias="t1", tz_aware=True)
        conn = get_connection("t1")

        self.assertTrue(conn.tz_aware)

        connect("mongoenginetest2", alias="t2")
        conn = get_connection("t2")
        self.assertFalse(conn.tz_aware)

    def test_connect(self):
        """Ensure that the connect() method works properly."""
        connect('mongoenginetest')

        conn = get_connection()
        self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)

        db = get_db()
        self.assertIsInstance(db, pymongo.database.Database)
        self.assertEqual(db.name, 'mongoenginetest')

        connect('mongoenginetest2', alias='testdb')
        conn = get_connection('testdb')
        self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)

def disconnect(alias=DEFAULT_CONNECTION_NAME):
    """ To disconnect pymongo connection.

    Copied from mongoengine/connection.py to fix a bug in mongoengine source code,
    ('disconnect' method is removed from pymongo MongoClient in latest version.)
    """
    global _connections
    global _dbs

    if alias in _connections:
        get_connection(alias=alias).close()
        del _connections[alias]
    if alias in _dbs:
        del _dbs[alias]

    def test_connect_in_mocking(self):
        """Ensure that the connect() method works properly in mocking.
        """
        try:
            import mongomock
        except ImportError:
            raise SkipTest('you need mongomock installed to run this testcase')

        connect('mongoenginetest', host='mongomock://localhost')
        conn = get_connection()
        self.assertTrue(isinstance(conn, mongomock.MongoClient))

        connect('mongoenginetest2', host='mongomock://localhost', alias='testdb')
        conn = get_connection('testdb')
        self.assertTrue(isinstance(conn, mongomock.MongoClient))

    def test_connect(self):
        """Ensure that the connect() method works properly.
        """
        connect('mongoenginetest')

        conn = get_connection()
        self.assertTrue(isinstance(conn, pymongo.connection.Connection))

        db = get_db()
        self.assertTrue(isinstance(db, pymongo.database.Database))
        self.assertEqual(db.name, 'mongoenginetest')

        connect('mongoenginetest2', alias='testdb')
        conn = get_connection('testdb')
        self.assertTrue(isinstance(conn, pymongo.connection.Connection))

    def test_sharing_connections(self):
        """Ensure that connections are shared when the connection settings are exactly the same
        """
        connect('mongoenginetests', alias='testdb1')
        expected_connection = get_connection('testdb1')

        connect('mongoenginetests', alias='testdb2')
        actual_connection = get_connection('testdb2')

        # Handle PyMongo 3+ Async Connection
        if IS_PYMONGO_3:
            # Ensure we are connected, throws ServerSelectionTimeoutError otherwise.
            # Purposely not catching exception to fail test if thrown.
            expected_connection.server_info()

        self.assertEqual(expected_connection, actual_connection)

 def teardown_databases(self, *args, **kwargs):
     for alias, params in self._iter_test_databases():
         connection = get_connection(alias)
         print("Dropping test database for alias '%s': %s" % (alias, params['name']))
         connection.drop_database(params['name'])
         disconnect(alias)
     return super(TestRunner, self).teardown_databases(*args, **kwargs)

    def test_register_connection_defaults(self):
        """Ensure that defaults are used when the host and port are None.
        """
        register_connection('testdb', 'mongoenginetest', host=None, port=None)

        conn = get_connection('testdb')
        self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))

 def teardown_databases(self, old_config, **kwargs):
     from mongoengine.connection import get_connection, disconnect
     connection = get_connection()
     connection.drop_database(self.mongodb_name)
     print 'Dropping mongo test database: ' + self.mongodb_name
     disconnect()
     super(MongoTestRunner, self).teardown_databases(old_config, **kwargs)

 def setUp(self):
     # データベースに接続
     addr = '127.0.0.1'
     port = 27017
     connect('test', host=addr, port=port)
     self.conn = get_connection()
     self.db = get_db()

    def test_connect_uri(self):
        """Ensure that the connect() method works properly with URIs."""
        c = connect(db='mongoenginetest', alias='admin')
        c.admin.system.users.remove({})
        c.mongoenginetest.system.users.remove({})

        c.admin.add_user("admin", "password")
        c.admin.authenticate("admin", "password")
        c.mongoenginetest.add_user("username", "password")

        if not IS_PYMONGO_3:
            self.assertRaises(
                MongoEngineConnectionError, connect, 'testdb_uri_bad',
                host='mongodb://test:[email protected]'
            )

        connect("testdb_uri", host='mongodb://username:[email protected]/mongoenginetest')

        conn = get_connection()
        self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))

        db = get_db()
        self.assertTrue(isinstance(db, pymongo.database.Database))
        self.assertEqual(db.name, 'mongoenginetest')

        c.admin.system.users.remove({})
        c.mongoenginetest.system.users.remove({})

def get_mongodb_version():
    """Return the version of the connected mongoDB (first 2 digits)

    :return: tuple(int, int)
    """
    version_list = get_connection().server_info()['versionArray'][:2]     # e.g: (3, 2)
    return tuple(version_list)

    def test_connect_uri(self):
        """Ensure that the connect() method works properly with URIs."""
        c = connect(db='mongoenginetest', alias='admin')
        c.admin.system.users.delete_many({})
        c.mongoenginetest.system.users.delete_many({})

        c.admin.command("createUser", "admin", pwd="password", roles=["root"])
        c.admin.authenticate("admin", "password")
        c.admin.command("createUser", "username", pwd="password", roles=["dbOwner"])

        if not IS_PYMONGO_3:
            self.assertRaises(
                MongoEngineConnectionError, connect, 'testdb_uri_bad',
                host='mongodb://test:[email protected]'
            )

        connect("testdb_uri", host='mongodb://username:[email protected]/mongoenginetest')

        conn = get_connection()
        self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)

        db = get_db()
        self.assertIsInstance(db, pymongo.database.Database)
        self.assertEqual(db.name, 'mongoenginetest')

        c.admin.system.users.delete_many({})
        c.mongoenginetest.system.users.delete_many({})

    def test_connect_uri_without_db(self):
        """Ensure that the connect() method works properly with uri's
        without database_name
        """
        c = connect(db='mongoenginetest', alias='admin')
        c.admin.system.users.remove({})
        c.mongoenginetest.system.users.remove({})

        c.admin.add_user("admin", "password")
        c.admin.authenticate("admin", "password")
        c.mongoenginetest.add_user("username", "password")

        self.assertRaises(ConnectionError, connect, "testdb_uri_bad", host='mongodb://test:[email protected]')

        connect("mongoenginetest", host='mongodb://localhost/')

        conn = get_connection()
        self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))

        db = get_db()
        self.assertTrue(isinstance(db, pymongo.database.Database))
        self.assertEqual(db.name, 'mongoenginetest')

        c.admin.system.users.remove({})
        c.mongoenginetest.system.users.remove({})

    def test_connect_uri(self):
        """Ensure that the connect() method works properly with uri's
        """
        c = connect(alias='admin')
        register_db('mongoenginetest', 'admin', 'admin')
        c.admin.system.users.remove({})
        c.mongoenginetest.system.users.remove({})

        c.admin.add_user("admin", "password")
        c.admin.authenticate("admin", "password")
        c.mongoenginetest.add_user("username", "password")

        self.assertRaises(
            ConnectionError, connect, "testdb_uri_bad",
            host='mongodb://test:[email protected]')

        # Whilst database names can be specified in the URI, they are ignored
        # in mongoengine since the DB/connection split
        connect(host='mongodb://username:[email protected]/mongoenginetest')
        register_db('testdb_uri')

        conn = get_connection()
        self.assertTrue(isinstance(conn, pymongo.connection.Connection))

        db = get_db()
        self.assertTrue(isinstance(db, pymongo.database.Database))
        self.assertEqual(db.name, 'testdb_uri')

    def test_ttl_indexes(self):

        class Log(Document):
            created = DateTimeField(default=datetime.now)
            meta = {
                'indexes': [
                    {'fields': ['created'], 'expireAfterSeconds': 3600}
                ]
            }

        Log.drop_collection()

        if pymongo.version_tuple[0] < 2 and pymongo.version_tuple[1] < 3:
            raise SkipTest('pymongo needs to be 2.3 or higher for this test')

        connection = get_connection()
        version_array = connection.server_info()['versionArray']
        if version_array[0] < 2 and version_array[1] < 2:
            raise SkipTest('MongoDB needs to be 2.2 or higher for this test')

        # Indexes are lazy so use list() to perform query
        list(Log.objects)
        info = Log.objects._collection.index_information()
        self.assertEqual(3600,
                         info['created_1']['expireAfterSeconds'])

    def test_connect_with_db_name_external(self):
        """Ensure that connect() works if db name is $external
        """
        """Ensure that the connect() method works properly."""
        connect('$external')

        conn = get_connection()
        self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)

        db = get_db()
        self.assertIsInstance(db, pymongo.database.Database)
        self.assertEqual(db.name, '$external')

        connect('$external', alias='testdb')
        conn = get_connection('testdb')
        self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)

 def _post_teardown(self):
     from mongoengine.connection import get_connection, disconnect
     for db_name, db_alias in settings.MONGO_DATABASES.items():
         connection = get_connection(db_alias)
         connection.drop_database(db_name)
         disconnect(db_alias)
     super(MongoTestCase, self)._post_teardown()