Python servers.Server类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Python中mongo_orchestration.servers.Server类的典型用法代码示例。如果您正苦于以下问题：Python Server类的具体用法？Python Server怎么用？Python Server使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了Server类的17个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_start

    def test_start(self):
        self.assertNotIn('pid', self.server.info()['procInfo'])
        self.assertTrue(self.server.start(30))
        self.assertTrue(self.server.info()['procInfo']['pid'] > 0)

        fake_server = Server('fake_proc_', {})
        self.assertRaises(OSError, fake_server.start, 5)
        fake_server.cleanup()

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:8，代码来源:test_servers.py

示例2: test_mongos

 def test_mongos(self):
     self.server.cleanup()
     self.server = Server(self.mongod, {'configsvr': True})
     self.server.start(30)
     mongos = os.path.join(os.environ.get('MONGOBIN', ''), 'mongos')
     self.server2 = Server(mongos, {'configdb': self.server.info()['uri']})
     self.assertTrue(self.server2.start())
     self.assertTrue(self.server2.info()['statuses'].get('mongos', False))
     self.server2.stop()
     self.server2.cleanup()

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:10，代码来源:test_servers.py

示例3: test_set_parameter

 def test_set_parameter(self):
     cfg = {"setParameter": {"textSearchEnabled": True,
                             "enableTestCommands": 1}}
     server = Server(self.mongod, cfg)
     server.start()
     c = pymongo.MongoClient(server.hostname)
     c.foo.bar.insert({"data": "text stuff"})
     # No Exception.
     c.foo.bar.ensure_index([("data", pymongo.TEXT)])
     # No Exception.
     c.admin.command("sleep", secs=1)

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:11，代码来源:test_servers.py

示例4: test_start_with_repair

 def test_start_with_repair(self):
     self.server.cleanup()
     self.server = Server(self.mongod, {"journal": False})
     self.server.start(30)
     os.kill(self.server.pid, 9)
     self.assertTrue(self.server._is_locked)
     self.assertTrue(self.server.start(20))

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:7，代码来源:test_servers.py

示例5: test_mongodb_auth_uri

    def test_mongodb_auth_uri(self):
        if SERVER_VERSION < (2, 4):
            raise SkipTest("Need to be able to set 'authenticationMechanisms' "
                           "parameter to test.")

        proc_params = {
            'setParameter': {
                'authenticationMechanisms': 'MONGODB-X509'
            }
        }
        ssl_params = {
            'sslPEMKeyFile': certificate('server.pem'),
            'sslCAFile': certificate('ca.pem'),
            'sslMode': 'requireSSL',
            'sslAllowInvalidCertificates': True
        }
        self.server = Server(
            'mongod', proc_params, ssl_params,
            login=TEST_SUBJECT, auth_source='$external')
        self.server.start()

        self.assertIn('mongodb_auth_uri', self.server.info())
        auth_uri = self.server.info()['mongodb_auth_uri']
        self.assertIn(self.server.hostname, auth_uri)
        self.assertIn(TEST_SUBJECT, auth_uri)
        self.assertIn('authSource=$external', auth_uri)
        self.assertIn('authMechanism=MONGODB-X509', auth_uri)

开发者ID:agilemobiledev，项目名称:mongo-orchestration，代码行数:27，代码来源:test_servers.py

示例6: test_ssl_auth

    def test_ssl_auth(self):
        if SERVER_VERSION < (2, 4):
            raise SkipTest("Need to be able to set 'authenticationMechanisms' "
                           "parameter to test.")

        proc_params = {
            'setParameter': {
                'authenticationMechanisms': 'MONGODB-X509'
            }
        }
        ssl_params = {
            'sslPEMKeyFile': certificate('server.pem'),
            'sslCAFile': certificate('ca.pem'),
            'sslMode': 'requireSSL',
            'sslAllowInvalidCertificates': True
        }
        # Should not raise an Exception.
        self.server = Server(
            'mongod', proc_params, ssl_params,
            login=TEST_SUBJECT, auth_source='$external')
        self.server.start()
        # Should create an extra user. Doesn't raise.
        client = pymongo.MongoClient(
            self.server.hostname, ssl_certfile=DEFAULT_CLIENT_CERT,
            ssl_cert_reqs=ssl.CERT_NONE)
        client['$external'].authenticate(
            DEFAULT_SUBJECT, mechanism='MONGODB-X509')
        # Should also create the user we requested. Doesn't raise.
        client = pymongo.MongoClient(
            self.server.hostname, ssl_certfile=certificate('client.pem'),
            ssl_cert_reqs=ssl.CERT_NONE)
        client['$external'].authenticate(
            TEST_SUBJECT, mechanism='MONGODB-X509')

开发者ID:agilemobiledev，项目名称:mongo-orchestration，代码行数:33，代码来源:test_servers.py

示例7: test_init_path

 def test_init_path(self):
     self.server.cleanup()
     mongod = os.path.join(os.environ.get('MONGOBIN', ''), 'mongod')
     log_dir = os.path.join(tempfile.gettempdir(), os.path.split(tempfile.mktemp())[-1])
     log_path = tempfile.mktemp(dir=log_dir)
     db_path = os.path.join(tempfile.gettempdir(), os.path.split(tempfile.mktemp())[-1])
     self.assertFalse(os.path.exists(log_dir))
     self.assertFalse(os.path.exists(db_path))
     self.server = Server(mongod, {'logpath': log_path, 'dbpath': db_path})
     self.assertTrue(os.path.exists(log_dir))
     self.assertTrue(os.path.exists(db_path))

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:11，代码来源:test_servers.py

示例8: test_set_parameter

 def test_set_parameter(self):
     if SERVER_VERSION < (2, 4):
         raise SkipTest(
             "Need at least MongoDB >= 2.4 to test setParameter.")
     self.server.cleanup()
     cfg = {"setParameter": {"textSearchEnabled": True,
                             "enableTestCommands": 1}}
     self.server = Server(self.mongod, cfg)
     self.server.start()
     c = pymongo.MongoClient(self.server.hostname)
     c.foo.bar.insert({"data": "text stuff"})
     # No Exception.
     c.foo.bar.ensure_index([("data", pymongo.TEXT)])
     # No Exception.
     c.admin.command("sleep", secs=1)

开发者ID:bsamek，项目名称:mongo-orchestration，代码行数:15，代码来源:test_servers.py

示例9: test_operations

    def test_operations(self):
        server_id = self.servers.create('mongod', {}, autostart=False)
        self.assertTrue(len(self.servers) == 1)
        self.assertTrue(server_id in self.servers)
        server_id2 = 'server-id2'
        server2 = Server(os.path.join(os.environ.get('MONGOBIN', ''), 'mongod'), {})
        server2.start(30)
        server2_pid = server2.info()['procInfo']['pid']
        self.servers[server_id2] = server2
        self.assertTrue(self.servers[server_id2]['procInfo']['pid'] == server2_pid)
        self.assertTrue(server_id2 in self.servers)
        for h_id in self.servers:
            self.assertTrue(h_id in (server_id, server_id2))

        operator.delitem(self.servers, server_id2)
        self.assertFalse(server_id2 in self.servers)
        server2.stop()
        server2.cleanup()

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:18，代码来源:test_servers.py

示例10: test_ssl

 def test_ssl(self):
     ssl_params = {
         'sslPEMKeyFile': certificate('server.pem'),
         'sslCAFile': certificate('ca.pem'),
         'sslMode': 'requireSSL',
         'sslAllowInvalidCertificates': True
     }
     # Should not raise an Exception.
     self.server = Server('mongod', {}, ssl_params)
     self.server.start()
     # Server should require SSL.
     with self.assertRaises(pymongo.errors.ConnectionFailure):
         connected(pymongo.MongoClient(self.server.hostname))
     # Doesn't raise with certificate provided.
     connected(pymongo.MongoClient(
         self.server.hostname, ssl_certfile=certificate('client.pem'),
         ssl_cert_reqs=ssl.CERT_NONE))

开发者ID:agilemobiledev，项目名称:mongo-orchestration，代码行数:17，代码来源:test_servers.py

示例11: test_info

    def test_info(self):
        self.server.start(30)
        info = self.server.info()
        for item in ("uri", "mongodb_uri", "statuses", "serverInfo", "procInfo", "orchestration"):
            self.assertTrue(item in info)

        self.assertTrue(info['mongodb_uri'].find(info['uri']))
        self.assertTrue(info['mongodb_uri'].find('mongodb://') == 0)
        fd_log, log_path = tempfile.mkstemp()
        os.close(fd_log)
        db_path = tempfile.mkdtemp()
        params = {'logpath': log_path, 'dbpath': db_path}
        server2 = Server(self.mongod, params)
        server2.start()
        info2 = server2.info()
        for param, value in params.items():
            self.assertTrue(info2['procInfo']['params'].get(param, value) == value)
        server2.stop()
        info = server2.info()
        self.assertEqual(len(info['serverInfo']), 0)
        self.assertEqual(len(info['statuses']), 0)
        self.assertEqual(info['orchestration'], 'servers')
        server2.cleanup()

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:23，代码来源:test_servers.py

示例12: test_scram_with_ssl

 def test_scram_with_ssl(self):
     ssl_params = {
         'sslPEMKeyFile': certificate('server.pem'),
         'sslCAFile': certificate('ca.pem'),
         'sslMode': 'requireSSL',
         'sslAllowInvalidCertificates': True
     }
     # Should not raise an Exception.
     self.server = Server(
         'mongod', {}, ssl_params, login='luke', password='ekul')
     self.server.start()
     # Should create the user we requested. No raise on authenticate.
     client = pymongo.MongoClient(
         self.server.hostname, ssl_certfile=certificate('client.pem'),
         ssl_cert_reqs=ssl.CERT_NONE)
     client.admin.authenticate('luke', 'ekul')
     # This should be the only user.
     self.assertEqual(len(client.admin.command('usersInfo')['users']), 1)
     self.assertFalse(client['$external'].command('usersInfo')['users'])

开发者ID:agilemobiledev，项目名称:mongo-orchestration，代码行数:19，代码来源:test_servers.py

示例13: setUp

 def setUp(self):
     PortPool().change_range()
     self.mongod = os.path.join(os.environ.get('MONGOBIN', ''), 'mongod')
     self.server = Server(self.mongod, {}, auth_key='secret', login='admin', password='admin')
     self.server.start()

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:5，代码来源:test_servers.py

示例14: ServerAuthTestCase

class ServerAuthTestCase(unittest.TestCase):
    def setUp(self):
        PortPool().change_range()
        self.mongod = os.path.join(os.environ.get('MONGOBIN', ''), 'mongod')
        self.server = Server(self.mongod, {}, auth_key='secret', login='admin', password='admin')
        self.server.start()

    def tearDown(self):
        if hasattr(self, 'server'):
            self.server.stop()
            self.server.cleanup()

    def test_mongos(self):
        self.server.stop()
        self.server.cleanup()
        self.server = Server(self.mongod, {'configsvr': True}, auth_key='secret')
        self.server.start(30)
        mongos = os.path.join(os.environ.get('MONGOBIN', ''), 'mongos')
        self.server2 = Server(mongos, {'configdb': self.server.info()['uri']}, auth_key='secret', login='admin', password='admin')
        self.server2.start()

        for server in (self.server, self.server2):
            c = pymongo.MongoClient(server.host, server.port)
            self.assertRaises(pymongo.errors.OperationFailure, c.admin.collection_names)
            self.assertTrue(c.admin.authenticate('admin', 'admin'))
            self.assertTrue(isinstance(c.admin.collection_names(), list))
            self.assertTrue(c.admin.logout() is None)
            self.assertRaises(pymongo.errors.OperationFailure, c.admin.collection_names)

        self.server2.stop()
        self.server2.cleanup()

    def test_auth_connection(self):
        self.assertTrue(isinstance(self.server.connection.admin.collection_names(), list))
        c = pymongo.MongoClient(self.server.host, self.server.port)
        self.assertRaises(pymongo.errors.OperationFailure, c.admin.collection_names)
        self.server.restart()
        c = pymongo.MongoClient(self.server.host, self.server.port)
        self.assertRaises(pymongo.errors.OperationFailure, c.admin.collection_names)

    def test_auth_admin(self):
        c = pymongo.MongoClient(self.server.host, self.server.port)
        self.assertRaises(pymongo.errors.OperationFailure, c.admin.collection_names)
        self.assertTrue(c.admin.authenticate('admin', 'admin'))
        self.assertTrue(isinstance(c.admin.collection_names(), list))
        self.assertTrue(c.admin.logout() is None)
        self.assertRaises(pymongo.errors.OperationFailure, c.admin.collection_names)

    def test_auth_collection(self):
        c = pymongo.MongoClient(self.server.host, self.server.port)
        self.assertTrue(c.admin.authenticate('admin', 'admin'))
        db = c.test_server_auth
        db.add_user('user', 'userpass', roles=['readWrite'])
        c.admin.logout()

        self.assertTrue(db.authenticate('user', 'userpass'))
        self.assertTrue(db.foo.insert({'foo': 'bar'}, safe=True, wtimeout=1000))
        self.assertTrue(isinstance(db.foo.find_one(), dict))
        db.logout()
        self.assertRaises(pymongo.errors.OperationFailure, db.foo.find_one)

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:60，代码来源:test_servers.py

示例15: init

    def __init__(self, params):
        """init configuration acording params"""
        self.id = params.get('id', None) or str(uuid4())
        self.admin_added = False
        self.login = params.get('login', '')
        self.password = params.get('password', '')
        self.auth_key = params.get('auth_key', None)
        self.auth_source = params.get('authSource', 'admin')
        self._version = params.get('version')
        self._configsvrs = []
        self._routers = []
        self._shards = {}
        self.tags = {}

        self.sslParams = params.get('sslParams', {})
        self.kwargs = {}
        self.restart_required = self.login or self.auth_key
        self.x509_extra_user = False

        if self.sslParams:
            self.kwargs.update(DEFAULT_SSL_OPTIONS)

        # Determine what to do with config servers via mongos version.
        mongos_name = os.path.join(Servers().bin_path(self._version), 'mongos')
        mongos = Server(name=mongos_name, procParams={})
        mongos_version = mongos.version
        mongos.cleanup()
        configsvr_configs = params.get('configsvrs', [{}])
        self.uses_rs_configdb = (mongos_version >= (3, 1, 2) and
                                 len(configsvr_configs) == 1)
        self.configdb_singleton = (
            ReplicaSets() if self.uses_rs_configdb else Servers())
        if self.uses_rs_configdb:
            self.__init_configrs(configsvr_configs[0])
        elif mongos_version >= (3, 3, 2):
            raise ShardedClusterError(
                'mongos >= 3.3.2 requires the config database to be backed by '
                'a replica set.')
        elif mongos_version >= (3, 1, 2) and len(configsvr_configs) != 3:
            raise ShardedClusterError(
                "mongos >= 3.1.2 needs a config replica set or 3 old-style "
                "config servers.")
        else:
            self.__init_configsvrs(configsvr_configs)

        for r in params.get('routers', [{}]):
            self.router_add(r)
        for cfg in params.get('shards', []):
            shard_params = cfg.get('shardParams', {})
            shard_tags = shard_params.pop('tags', None)
            info = self.member_add(cfg.get('id', None), shard_params)
            if shard_tags:
                self.tags[info['id']] = shard_tags

        if self.tags:
            for sh_id in self.tags:
                logger.debug('Add tags %r to %s' % (self.tags[sh_id], sh_id))
                db = self.connection().get_database(
                    'config',
                    write_concern=write_concern.WriteConcern(fsync=True))
                db.shards.update(
                    {'_id': sh_id},
                    {'$addToSet': {'tags': {'$each': self.tags[sh_id]}}})

        shard_configs = [s.get('shardParams', {}).get('procParams', {})
                         for s in params.get('shards', [])]
        if self.login:
            # Do we need to add an extra x509 user?
            def only_x509(config):
                set_params = config.get('setParameter', {})
                auth_mechs = set_params.get('authenticationMechanisms', '')
                auth_mechs = auth_mechs.split(',')
                if len(auth_mechs) == 1 and auth_mechs[0] == 'MONGODB-X509':
                    return True
                return False

            any_only_x509 = lambda l: any(map(only_x509, l))
            rs_shard_configs = [
                m.get('procParams', {})
                for s in params.get('shards', [])
                for m in s.get('shardParams', {}).get('members', [])
            ]
            router_configs = params.get('routers', [])

            self.x509_extra_user = (any_only_x509(configsvr_configs) or
                                    any_only_x509(shard_configs) or
                                    any_only_x509(rs_shard_configs) or
                                    any_only_x509(router_configs))

            self._add_users(
                self.connection().get_database(
                    self.auth_source, write_concern=write_concern.WriteConcern(
                        fsync=True)))

            # Secondary user given from request.
            secondary_login = {
                'name': self.login,
                'roles': self._user_roles(self.connection())
            }
            if self.password:
#.........这里部分代码省略.........

开发者ID:bsamek，项目名称:mongo-orchestration，代码行数:101，代码来源:sharded_clusters.py

示例16: ServerTestCase

class ServerTestCase(unittest.TestCase):
    def setUp(self):
        PortPool().change_range()
        self.mongod = os.path.join(os.environ.get('MONGOBIN', ''), 'mongod')
        self.server = Server(self.mongod, {})

    def tearDown(self):
        if hasattr(self, 'server'):
            self.server.stop()
            self.server.cleanup()

    def test_server(self):
        self.assertTrue(isinstance(self.server, Server))

    def test_init_path(self):
        self.server.cleanup()
        mongod = os.path.join(os.environ.get('MONGOBIN', ''), 'mongod')
        log_dir = os.path.join(tempfile.gettempdir(), os.path.split(tempfile.mktemp())[-1])
        log_path = tempfile.mktemp(dir=log_dir)
        db_path = os.path.join(tempfile.gettempdir(), os.path.split(tempfile.mktemp())[-1])
        self.assertFalse(os.path.exists(log_dir))
        self.assertFalse(os.path.exists(db_path))
        self.server = Server(mongod, {'logpath': log_path, 'dbpath': db_path})
        self.assertTrue(os.path.exists(log_dir))
        self.assertTrue(os.path.exists(db_path))

    def test_mongos(self):
        self.server.cleanup()
        self.server = Server(self.mongod, {'configsvr': True})
        self.server.start(30)
        mongos = os.path.join(os.environ.get('MONGOBIN', ''), 'mongos')
        self.server2 = Server(mongos, {'configdb': self.server.info()['uri']})
        self.assertTrue(self.server2.start())
        self.assertTrue(self.server2.info()['statuses'].get('mongos', False))
        self.server2.stop()
        self.server2.cleanup()

    def test_run_command(self):
        self.server.start(30)

    def test_info(self):
        self.server.start(30)
        info = self.server.info()
        for item in ("uri", "mongodb_uri", "statuses", "serverInfo", "procInfo", "orchestration"):
            self.assertTrue(item in info)

        self.assertTrue(info['mongodb_uri'].find(info['uri']))
        self.assertTrue(info['mongodb_uri'].find('mongodb://') == 0)
        fd_log, log_path = tempfile.mkstemp()
        os.close(fd_log)
        db_path = tempfile.mkdtemp()
        params = {'logpath': log_path, 'dbpath': db_path}
        server2 = Server(self.mongod, params)
        server2.start()
        info2 = server2.info()
        for param, value in params.items():
            self.assertTrue(info2['procInfo']['params'].get(param, value) == value)
        server2.stop()
        info = server2.info()
        self.assertEqual(len(info['serverInfo']), 0)
        self.assertEqual(len(info['statuses']), 0)
        self.assertEqual(info['orchestration'], 'servers')
        server2.cleanup()

    def test_command(self):
        self.assertRaises(pymongo.errors.PyMongoError, self.server.run_command, 'serverStatus', None, False)
        self.server.start(30)
        self.assertEqual(self.server.run_command('serverStatus', arg=None, is_eval=False).get('ok', -1), 1)
        self.assertEqual(self.server.run_command('db.getName()', arg=None, is_eval=True), 'admin')

    def test_start(self):
        self.assertNotIn('pid', self.server.info()['procInfo'])
        self.assertTrue(self.server.start(30))
        self.assertTrue(self.server.info()['procInfo']['pid'] > 0)

        fake_server = Server('fake_proc_', {})
        self.assertRaises(OSError, fake_server.start, 5)
        fake_server.cleanup()

    def test_start_with_repair(self):
        self.server.cleanup()
        self.server = Server(self.mongod, {"journal": False})
        self.server.start(30)
        os.kill(self.server.pid, 9)
        self.assertTrue(self.server._is_locked)
        self.assertTrue(self.server.start(20))

    def test_stop(self):
        self.assertTrue(self.server.start(60))
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server = self.server.hostname.split(':')[0]
        s.connect((server, self.server.cfg['port']))
        self.assertTrue(self.server.stop())
        self.assertRaises(socket.error, s.connect, (server, self.server.cfg['port']))

    def test_restart(self):
        self.assertTrue(self.server.start(30))
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server = self.server.hostname.split(':')[0]
        s.connect((server, self.server.cfg['port']))
#.........这里部分代码省略.........

开发者ID:jyemin，项目名称:mongo-orchestration，代码行数:101，代码来源:test_servers.py

示例17: ServerSSLTestCase

class ServerSSLTestCase(SSLTestCase):

    def tearDown(self):
        self.server.stop()
        self.server.cleanup()

    def test_ssl_auth(self):
        if SERVER_VERSION < (2, 4):
            raise SkipTest("Need to be able to set 'authenticationMechanisms' "
                           "parameter to test.")

        proc_params = {
            'setParameter': {
                'authenticationMechanisms': 'MONGODB-X509'
            }
        }
        ssl_params = {
            'sslPEMKeyFile': certificate('server.pem'),
            'sslCAFile': certificate('ca.pem'),
            'sslMode': 'requireSSL',
            'sslAllowInvalidCertificates': True
        }
        # Should not raise an Exception.
        self.server = Server(
            'mongod', proc_params, ssl_params,
            login=TEST_SUBJECT, auth_source='$external')
        self.server.start()
        # Should create an extra user. Doesn't raise.
        client = pymongo.MongoClient(
            self.server.hostname, ssl_certfile=DEFAULT_CLIENT_CERT,
            ssl_cert_reqs=ssl.CERT_NONE)
        client['$external'].authenticate(
            DEFAULT_SUBJECT, mechanism='MONGODB-X509')
        # Should also create the user we requested. Doesn't raise.
        client = pymongo.MongoClient(
            self.server.hostname, ssl_certfile=certificate('client.pem'),
            ssl_cert_reqs=ssl.CERT_NONE)
        client['$external'].authenticate(
            TEST_SUBJECT, mechanism='MONGODB-X509')

    def test_scram_with_ssl(self):
        ssl_params = {
            'sslPEMKeyFile': certificate('server.pem'),
            'sslCAFile': certificate('ca.pem'),
            'sslMode': 'requireSSL',
            'sslAllowInvalidCertificates': True
        }
        # Should not raise an Exception.
        self.server = Server(
            'mongod', {}, ssl_params, login='luke', password='ekul')
        self.server.start()
        # Should create the user we requested. No raise on authenticate.
        client = pymongo.MongoClient(
            self.server.hostname, ssl_certfile=certificate('client.pem'),
            ssl_cert_reqs=ssl.CERT_NONE)
        client.admin.authenticate('luke', 'ekul')
        # This should be the only user.
        self.assertEqual(len(client.admin.command('usersInfo')['users']), 1)
        self.assertFalse(client['$external'].command('usersInfo')['users'])

    def test_ssl(self):
        ssl_params = {
            'sslPEMKeyFile': certificate('server.pem'),
            'sslCAFile': certificate('ca.pem'),
            'sslMode': 'requireSSL',
            'sslAllowInvalidCertificates': True
        }
        # Should not raise an Exception.
        self.server = Server('mongod', {}, ssl_params)
        self.server.start()
        # Server should require SSL.
        with self.assertRaises(pymongo.errors.ConnectionFailure):
            connected(pymongo.MongoClient(self.server.hostname))
        # Doesn't raise with certificate provided.
        connected(pymongo.MongoClient(
            self.server.hostname, ssl_certfile=certificate('client.pem'),
            ssl_cert_reqs=ssl.CERT_NONE))

    def test_mongodb_auth_uri(self):
        if SERVER_VERSION < (2, 4):
            raise SkipTest("Need to be able to set 'authenticationMechanisms' "
                           "parameter to test.")

        proc_params = {
            'setParameter': {
                'authenticationMechanisms': 'MONGODB-X509'
            }
        }
        ssl_params = {
            'sslPEMKeyFile': certificate('server.pem'),
            'sslCAFile': certificate('ca.pem'),
            'sslMode': 'requireSSL',
            'sslAllowInvalidCertificates': True
        }
        self.server = Server(
            'mongod', proc_params, ssl_params,
            login=TEST_SUBJECT, auth_source='$external')
        self.server.start()

        self.assertIn('mongodb_auth_uri', self.server.info())
#.........这里部分代码省略.........

开发者ID:agilemobiledev，项目名称:mongo-orchestration，代码行数:101，代码来源:test_servers.py

注：本文中的mongo_orchestration.servers.Server类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python sharded_clusters.ShardedCluster类代码示例发布时间：2022-05-27

Python replica_sets.ReplicaSet类代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13931|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10292|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4169|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4064|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3889|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3539|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3067|2022-01-22

8 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2726|2022-01-22

9 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2682|2022-05-25

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2346|2022-01-22

客服电话

电子邮件

Python servers.Server类代码示例

示例1: test_start

示例2: test_mongos

示例3: test_set_parameter

示例4: test_start_with_repair

示例5: test_mongodb_auth_uri

示例6: test_ssl_auth

示例7: test_init_path

示例8: test_set_parameter

示例9: test_operations

示例10: test_ssl

示例11: test_info

示例12: test_scram_with_ssl

示例13: setUp

示例14: ServerAuthTestCase

示例15: init

示例16: ServerTestCase

示例17: ServerSSLTestCase

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053

客服电话

电子邮件

Python servers.Server类代码示例

示例1: test_start

示例2: test_mongos

示例3: test_set_parameter

示例4: test_start_with_repair

示例5: test_mongodb_auth_uri

示例6: test_ssl_auth

示例7: test_init_path

示例8: test_set_parameter

示例9: test_operations

示例10: test_ssl

示例11: test_info

示例12: test_scram_with_ssl

示例13: setUp

示例14: ServerAuthTestCase

示例15: __init__

示例16: ServerTestCase

示例17: ServerSSLTestCase

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053

示例15: init