1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
# Copyright 2012-2015 MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Utilities for testing Motor with Tornado."""
import concurrent.futures
import functools
from test.test_environment import CA_PEM, CLIENT_PEM, env
from test.version import Version
from unittest import SkipTest
from bson import SON
from mockupdb import MockupDB
from tornado import testing
import motor
async def get_command_line(client):
command_line = await client.admin.command("getCmdLineOpts")
assert command_line["ok"] == 1, "getCmdLineOpts() failed"
return command_line
async def server_is_mongos(client):
ismaster_response = await client.admin.command("ismaster")
return ismaster_response.get("msg") == "isdbgrid"
async def skip_if_mongos(client):
is_mongos = await server_is_mongos(client)
if is_mongos:
raise SkipTest("connected to mongos")
async def remove_all_users(db):
await db.command({"dropAllUsersFromDatabase": 1})
class MotorTest(testing.AsyncTestCase):
longMessage = True # Used by unittest.TestCase
ssl = False # If True, connect with SSL, skip if mongod isn't SSL
def setUp(self):
super().setUp()
if self.ssl and not env.mongod_started_with_ssl:
raise SkipTest("mongod doesn't support SSL, or is down")
self.cx = self.motor_client()
self.db = self.cx.motor_test
self.collection = self.db.test_collection
async def make_test_data(self):
await self.collection.delete_many({})
await self.collection.insert_many([{"_id": i} for i in range(200)])
# Workaround for https://github.com/pytest-dev/pytest/issues/12263.
def runTest(self):
pass
make_test_data.__test__ = False
async def set_fail_point(self, client, command_args):
cmd = SON([("configureFailPoint", "failCommand")])
cmd.update(command_args)
await client.admin.command(cmd)
def get_client_kwargs(self, **kwargs):
if env.mongod_started_with_ssl:
kwargs.setdefault("tlsCAFile", CA_PEM)
kwargs.setdefault("tlsCertificateKeyFile", CLIENT_PEM)
kwargs.setdefault("tls", env.mongod_started_with_ssl)
kwargs.setdefault("io_loop", self.io_loop)
return kwargs
def motor_client(self, uri=None, *args, **kwargs):
"""Get a MotorClient.
Ignores self.ssl, you must pass 'ssl' argument. You'll probably need to
close the client to avoid file-descriptor problems after AsyncTestCase
calls self.io_loop.close(all_fds=True).
"""
return motor.MotorClient(uri or env.uri, *args, **self.get_client_kwargs(**kwargs))
def motor_rsc(self, uri=None, *args, **kwargs):
"""Get an open MotorClient for replica set.
Ignores self.ssl, you must pass 'ssl' argument.
"""
return motor.MotorClient(uri or env.rs_uri, *args, **self.get_client_kwargs(**kwargs))
def tearDown(self):
env.sync_cx.motor_test.test_collection.delete_many({})
if hasattr(self, "cx"):
self.cx.close()
super().tearDown()
class MotorReplicaSetTestBase(MotorTest):
def setUp(self):
super().setUp()
if not env.is_replica_set:
raise SkipTest("Not connected to a replica set")
self.rsc = self.motor_rsc()
class MotorMockServerTest(MotorTest):
executor = concurrent.futures.ThreadPoolExecutor(1)
def server(self, *args, **kwargs):
server = MockupDB(*args, **kwargs)
server.run()
self.addCleanup(server.stop)
return server
def client_server(self, *args, **kwargs):
server = self.server(*args, **kwargs)
client = motor.motor_tornado.MotorClient(server.uri, io_loop=self.io_loop)
self.addCleanup(client.close)
return client, server
async def run_thread(self, fn, *args, **kwargs):
return await self.io_loop.run_in_executor(None, functools.partial(fn, *args, **kwargs))
class AsyncVersion(Version):
"""Version class that can be instantiated with an async client from
within a coroutine."""
@classmethod
async def from_client(cls, client):
info = await client.server_info()
if "versionArray" in info:
return cls.from_version_array(info["versionArray"])
return cls.from_string(info["version"])
|
