-
Notifications
You must be signed in to change notification settings - Fork 85
/
mongo.py
136 lines (106 loc) · 3.74 KB
/
mongo.py
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
import logging
import pytest
from pytest_server_fixtures import CONFIG
from pytest_fixture_config import yield_requires_config
from .base2 import TestServerV2
log = logging.getLogger(__name__)
def _mongo_server():
""" This does the actual work - there are several versions of this used
with different scopes.
"""
test_server = MongoTestServer()
try:
test_server.start()
yield test_server
finally:
test_server.teardown()
@pytest.yield_fixture(scope='function')
@yield_requires_config(CONFIG, ['mongo_bin'])
def mongo_server():
""" Function-scoped MongoDB server started in a local thread.
This also provides a temp workspace.
We tear down, and cleanup mongos at the end of the test.
For completeness, we tidy up any outstanding mongo temp directories
at the start and end of each test session
Attributes
----------
api (`pymongo.MongoClient`) : PyMongo Client API connected to this server
.. also inherits all attributes from the `workspace` fixture
"""
for server in _mongo_server():
yield server
@pytest.yield_fixture(scope='session')
@yield_requires_config(CONFIG, ['mongo_bin'])
def mongo_server_sess():
""" Same as mongo_server fixture, scoped as session instead.
"""
for server in _mongo_server():
yield server
@pytest.yield_fixture(scope='class')
@yield_requires_config(CONFIG, ['mongo_bin'])
def mongo_server_cls(request):
""" Same as mongo_server fixture, scoped for test classes.
"""
for server in _mongo_server():
request.cls.mongo_server = server
yield server
@pytest.yield_fixture(scope='module')
@yield_requires_config(CONFIG, ['mongo_bin'])
def mongo_server_module():
""" Same as mongo_server fixture, scoped for test modules.
"""
for server in _mongo_server():
yield server
class MongoTestServer(TestServerV2):
def __init__(self, delete=True, **kwargs):
super(MongoTestServer, self).__init__(delete=delete, **kwargs)
self._port = self._get_port(27017)
self.api = None
@property
def cmd(self):
return 'mongod'
@property
def cmd_local(self):
return CONFIG.mongo_bin
def get_args(self, **kwargs):
cmd = [
'--bind_ip=%s' % self._listen_hostname,
'--port=%s' % self.port,
'--nounixsocket',
'--syncdelay=0',
'--nojournal',
'--quiet',
]
if 'workspace' in kwargs:
cmd.append('--dbpath=%s' % str(kwargs['workspace']))
return cmd
@property
def image(self):
return CONFIG.mongo_image
@property
def port(self):
return self._port
def check_server_up(self):
"""Test connection to the server."""
import pymongo
from pymongo.errors import AutoReconnect, ConnectionFailure
# Hostname must exist before continuing
# Some server class (e.g. Docker) will only allocate an IP after the
# container has started.
if not self.hostname:
return False
log.info("Connecting to Mongo at %s:%s" % (self.hostname, self.port))
try:
with pymongo.MongoClient(self.hostname, self.port, serverselectiontimeoutms=200) as initial_api:
initial_api.list_database_names()
# Configure the client with default timeouts in case the server goes slow
self.api = pymongo.MongoClient(self.hostname, self.port)
return True
except (AutoReconnect, ConnectionFailure) as e:
pass
return False
def teardown(self):
if self.api:
self.api.close()
self.api = None
super(MongoTestServer, self).teardown()