Issues (15)

tests/conftest.py (3 issues)

1
import asyncio
2
3
import pytest
4
from gremlin_python.process.traversal import Cardinality
5
6
from goblin import Goblin, driver, element, properties
7
from goblin.driver import (
8
    Connection, DriverRemoteConnection, GraphSONMessageSerializer)
9
from goblin.provider import TinkerGraph
10
11
12
def pytest_generate_tests(metafunc):
13
    if 'cluster' in metafunc.fixturenames:
14
        metafunc.parametrize("cluster", ['c1', 'c2'], indirect=True)
15
16
17
def pytest_addoption(parser):
18
    parser.addoption(
19
        '--provider', default='tinkergraph', choices=(
20
            'tinkergraph',
21
            'dse',
22
        ))
23
    parser.addoption('--gremlin-host', default='gremlin-server')
24
    parser.addoption('--gremlin-port', default='8182')
25
26
27
def db_name_factory(x, y):
28
    return "{}__{}".format(y, x)
29
30
31
class HistoricalName(element.VertexProperty):
32
    notes = properties.Property(properties.String)
33
    year = properties.Property(properties.Integer)  # this is dumb but handy
34
35
36
class Location(element.VertexProperty):
37
    year = properties.Property(properties.Integer)
38
39
40
class Person(element.Vertex):
41
    __label__ = 'person'
42
    name = properties.Property(properties.String)
43
    age = properties.Property(
44
        properties.Integer, db_name='custom__person__age')
45
    birthplace = element.VertexProperty(properties.String)
46
    location = Location(properties.String, card=Cardinality.list_)
47
    nicknames = element.VertexProperty(
48
        properties.String,
49
        card=Cardinality.list_,
50
        db_name_factory=db_name_factory)
51
52
53
class Place(element.Vertex):
54
    name = properties.Property(properties.String)
55
    zipcode = properties.Property(properties.Integer, db_name_factory=db_name_factory)
56
    historical_name = HistoricalName(properties.String, card=Cardinality.list_)
57
    important_numbers = element.VertexProperty(properties.Integer, card=Cardinality.set_)
58
    incorporated = element.VertexProperty(properties.Boolean, default=False)
59
60
61
class Inherited(Person):
62
    pass
63
64
65
class Knows(element.Edge):
66
    __label__ = 'knows'
67
    notes = properties.Property(properties.String, default='N/A')
68
69
70
class LivesIn(element.Edge):
71
    notes = properties.Property(properties.String)
72
73
74
@pytest.fixture
75
def provider(request):
76
    provider = request.config.getoption('provider')
77
    if provider == 'tinkergraph':
78
        return TinkerGraph
79
    elif provider == 'dse':
80
        try:
81
            import goblin_dse
82
        except ImportError:
83
            raise RuntimeError(
84
                "Couldn't run tests with DSEGraph provider: the goblin_dse "
85
                "package must be installed")
86
        else:
87
            return goblin_dse.DSEGraph
88
89
90
@pytest.fixture
91
def aliases(request):
92
    if request.config.getoption('provider') == 'tinkergraph':
93
        return {'g': 'g'}
94
    elif request.config.getoption('provider') == 'dse':
95
        return {'g': 'testgraph.g'}
96
97
98
@pytest.fixture
99
def gremlin_server():
100
    return driver.GremlinServer
101
102
103
@pytest.fixture
104
def unused_server_url(unused_tcp_port):
105
    return 'http://localhost:{}/gremlin'.format(unused_tcp_port)
106
107
108
@pytest.fixture
109
def gremlin_host(request):
110
    return request.config.getoption('gremlin_host')
111
112
113
@pytest.fixture
114
def gremlin_port(request):
115
    return request.config.getoption('gremlin_port')
116
117
118
@pytest.fixture
119
def gremlin_url(gremlin_host, gremlin_port):
120
    return "http://{}:{}/gremlin".format(gremlin_host, gremlin_port)
121
122
123
@pytest.fixture
124
def cluster(request, gremlin_host, gremlin_port, event_loop, provider,
125
            aliases):
126
    if request.param == 'c1':
127
        cluster = driver.Cluster(
128
            event_loop,
129
            hosts=[gremlin_host],
130
            port=gremlin_port,
131
            aliases=aliases,
132
            provider=provider)
133
    elif request.param == 'c2':
134
        cluster = driver.Cluster(
135
            event_loop,
136
            hosts=[gremlin_host],
137
            port=gremlin_port,
138
            aliases=aliases,
139
            provider=provider)
140
    return cluster
0 ignored issues
show
The variable cluster does not seem to be defined for all execution paths.
Loading history...
141
142
143
@pytest.fixture
144
def connection(gremlin_url, event_loop, provider):
145
    try:
146
        conn = event_loop.run_until_complete(
147
            driver.Connection.open(
148
                gremlin_url,
149
                event_loop,
150
                message_serializer=GraphSONMessageSerializer,
151
                provider=provider))
152
    except OSError:
153
        pytest.skip('Gremlin Server is not running')
154
    return conn
155
156
157
@pytest.fixture
158
def remote_connection(event_loop, gremlin_url):
159
    try:
160
        remote_conn = event_loop.run_until_complete(
161
            DriverRemoteConnection.open(gremlin_url, 'g'))
162
    except OSError:
163
        pytest.skip('Gremlin Server is not running')
164
    else:
165
        return remote_conn
166
167
168
@pytest.fixture
169
def connection_pool(gremlin_url, event_loop, provider):
170
    return driver.ConnectionPool(
171
        gremlin_url,
172
        event_loop,
173
        None,
174
        '',
175
        '',
176
        4,
177
        1,
178
        16,
179
        64,
180
        None,
181
        driver.GraphSONMessageSerializer,
182
        provider=provider)
183
184
185
@pytest.fixture
186
def remote_graph():
187
    return driver.Graph()
188
189
190
@pytest.fixture
191
def app(gremlin_host, gremlin_port, event_loop, provider, aliases):
192
    app = event_loop.run_until_complete(
193
        Goblin.open(
194
            event_loop,
195
            provider=provider,
196
            aliases=aliases,
197
            hosts=[gremlin_host],
198
            port=gremlin_port))
199
200
    app.register(Person, Place, Knows, LivesIn)
201
    return app
202
203
204
# Instance fixtures
205
@pytest.fixture
206
def string():
207
    return properties.String()
208
209
210
@pytest.fixture
211
def integer():
212
    return properties.Integer()
213
214
215
@pytest.fixture
216
def flt():
217
    return properties.Float()
218
219
220
@pytest.fixture
221
def boolean():
222
    return properties.Boolean()
223
224
225
@pytest.fixture
226
def historical_name():
227
    return HistoricalName()
228
229
230
@pytest.fixture
231
def person():
232
    return Person()
233
234
235
@pytest.fixture
236
def place():
237
    return Place()
238
239
240
@pytest.fixture
241
def knows():
242
    return Knows()
243
244
245
@pytest.fixture
246
def lives_in():
247
    return LivesIn()
248
249
250
@pytest.fixture
251
def place_name():
252
    return PlaceName()
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable PlaceName does not seem to be defined.
Loading history...
253
254
255
# Class fixtures
256
@pytest.fixture
257
def cluster_class(event_loop):
258
    return driver.Cluster
259
260
261
@pytest.fixture
262
def string_class():
263
    return properties.String
264
265
266
@pytest.fixture
267
def integer_class():
268
    return properties.Integer
269
270
271
@pytest.fixture
272
def historical_name_class():
273
    return HistoricalName
274
275
276
@pytest.fixture
277
def person_class():
278
    return Person
279
280
281
@pytest.fixture
282
def inherited_class():
283
    return Inherited
284
285
286
@pytest.fixture
287
def place_class():
288
    return Place
289
290
291
@pytest.fixture
292
def knows_class():
293
    return Knows
294
295
296
@pytest.fixture
297
def lives_in_class():
298
    return LivesIn
299
300
301
@pytest.fixture
302
def place_name_class():
303
    return PlaceName
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable PlaceName does not seem to be defined.
Loading history...
304
305
306
@pytest.fixture
307
def string_class():
308
    return properties.String
309
310
311
@pytest.fixture
312
def integer_class():
313
    return properties.Integer
314
315
316
@pytest.fixture
317
def flt_class():
318
    return properties.Float
319
320
321
@pytest.fixture
322
def boolean_class():
323
    return properties.Boolean
324
325
326
@pytest.fixture(autouse=True)
327
def add_doctest_default(doctest_namespace, tmpdir, event_loop, app):
328
    doctest_namespace['Person'] = Person
329
    doctest_namespace['loop'] = event_loop
330
    doctest_namespace['app'] = app
331
    config = tmpdir.join('config.yml')
332
    config.write(
333
        "scheme: 'ws'\n"
334
        "hosts: ['localhost']\n"
335
        "port': 8182\n"
336
        "ssl_certfile: ''\n"
337
        "ssl_keyfile: ''\n"
338
        "ssl_password: ''\n"
339
        "username: ''\n"
340
        "password: ''\n"
341
        "response_timeout: None\n"
342
        "max_conns: 4\n"
343
        "min_conns: 1\n"
344
        "max_times_acquired: 16\n"
345
        "max_inflight: 64\n"
346
        "message_serializer: 'goblin.driver.GraphSONMessageSerializer'\n"
347
    )
348
    with tmpdir.as_cwd():
349
        yield
350