conftest.knows_class() - Code Metrics - Inspection of "Added better error handling and fixed g.E() traver..." - goblin-ogm/goblin - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c5801a...8fec65 )

by Jeffrey

created 2022-03-30 09:46 UTC

conftest.knows_class() A

↳ Parent: conftest

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	0
dl	0
loc	3
rs	10
c	0
b	0
f	0

import asyncio

import pytest
from gremlin_python.process.traversal import Cardinality

from goblin import Goblin, driver, element, properties
from goblin.driver import (
    Connection, DriverRemoteConnection, GraphSONMessageSerializer)
from goblin.provider import TinkerGraph


def pytest_generate_tests(metafunc):
    if 'cluster' in metafunc.fixturenames:
        metafunc.parametrize("cluster", ['c1', 'c2'], indirect=True)


def pytest_addoption(parser):
    parser.addoption(
        '--provider', default='tinkergraph', choices=(
            'tinkergraph',
            'dse',
        ))
    parser.addoption('--gremlin-host', default='gremlin-server')
    parser.addoption('--gremlin-port', default='8182')


def db_name_factory(x, y):
    return "{}__{}".format(y, x)


class HistoricalName(element.VertexProperty):
    notes = properties.Property(properties.String)
    year = properties.Property(properties.Integer)  # this is dumb but handy


class Location(element.VertexProperty):
    year = properties.Property(properties.Integer)


class Person(element.Vertex):
    __label__ = 'person'
    name = properties.Property(properties.String)
    age = properties.Property(
        properties.Integer, db_name='custom__person__age')
    birthplace = element.VertexProperty(properties.String)
    location = Location(properties.String, card=Cardinality.list_)
    nicknames = element.VertexProperty(
        properties.String,
        card=Cardinality.list_,
        db_name_factory=db_name_factory)


class Place(element.Vertex):
    name = properties.Property(properties.String)
    zipcode = properties.Property(properties.Integer, db_name_factory=db_name_factory)
    historical_name = HistoricalName(properties.String, card=Cardinality.list_)
    important_numbers = element.VertexProperty(properties.Integer, card=Cardinality.set_)
    incorporated = element.VertexProperty(properties.Boolean, default=False)


class Inherited(Person):
    pass


class Knows(element.Edge):
    __label__ = 'knows'
    notes = properties.Property(properties.String, default='N/A')


class LivesIn(element.Edge):
    notes = properties.Property(properties.String)


@pytest.fixture
def provider(request):
    provider = request.config.getoption('provider')
    if provider == 'tinkergraph':
        return TinkerGraph
    elif provider == 'dse':
        try:
            import goblin_dse
        except ImportError:
            raise RuntimeError(
                "Couldn't run tests with DSEGraph provider: the goblin_dse "
                "package must be installed")
        else:
            return goblin_dse.DSEGraph


@pytest.fixture
def aliases(request):
    if request.config.getoption('provider') == 'tinkergraph':
        return {'g': 'g'}
    elif request.config.getoption('provider') == 'dse':
        return {'g': 'testgraph.g'}


@pytest.fixture
def gremlin_server():
    return driver.GremlinServer


@pytest.fixture
def unused_server_url(unused_tcp_port):
    return 'http://localhost:{}/gremlin'.format(unused_tcp_port)


@pytest.fixture
def gremlin_host(request):
    return request.config.getoption('gremlin_host')


@pytest.fixture
def gremlin_port(request):
    return request.config.getoption('gremlin_port')


@pytest.fixture
def gremlin_url(gremlin_host, gremlin_port):
    return "http://{}:{}/gremlin".format(gremlin_host, gremlin_port)


@pytest.fixture
def cluster(request, gremlin_host, gremlin_port, event_loop, provider,
            aliases):
    if request.param == 'c1':
        cluster = driver.Cluster(
            event_loop,
            hosts=[gremlin_host],
            port=gremlin_port,
            aliases=aliases,
            provider=provider)
    elif request.param == 'c2':
        cluster = driver.Cluster(
            event_loop,
            hosts=[gremlin_host],
            port=gremlin_port,
            aliases=aliases,
            provider=provider)
    return cluster



@pytest.fixture
def connection(gremlin_url, event_loop, provider):
    try:
        conn = event_loop.run_until_complete(
            driver.Connection.open(
                gremlin_url,
                event_loop,
                message_serializer=GraphSONMessageSerializer,
                provider=provider))
    except OSError:
        pytest.skip('Gremlin Server is not running')
    return conn


@pytest.fixture
def remote_connection(event_loop, gremlin_url):
    try:
        remote_conn = event_loop.run_until_complete(
            DriverRemoteConnection.open(gremlin_url, 'g'))
    except OSError:
        pytest.skip('Gremlin Server is not running')
    else:
        return remote_conn


@pytest.fixture
def connection_pool(gremlin_url, event_loop, provider):
    return driver.ConnectionPool(
        gremlin_url,
        event_loop,
        None,
        '',
        '',
        4,
        1,
        16,
        64,
        None,
        driver.GraphSONMessageSerializer,
        provider=provider)


@pytest.fixture
def remote_graph():
    return driver.Graph()


@pytest.fixture
def app(gremlin_host, gremlin_port, event_loop, provider, aliases):
    app = event_loop.run_until_complete(
        Goblin.open(
            event_loop,
            provider=provider,
            aliases=aliases,
            hosts=[gremlin_host],
            port=gremlin_port))

    app.register(Person, Place, Knows, LivesIn)
    return app


# Instance fixtures
@pytest.fixture
def string():
    return properties.String()


@pytest.fixture
def integer():
    return properties.Integer()


@pytest.fixture
def flt():
    return properties.Float()


@pytest.fixture
def boolean():
    return properties.Boolean()


@pytest.fixture
def historical_name():
    return HistoricalName()


@pytest.fixture
def person():
    return Person()


@pytest.fixture
def place():
    return Place()


@pytest.fixture
def knows():
    return Knows()


@pytest.fixture
def lives_in():
    return LivesIn()


@pytest.fixture
def place_name():
    return PlaceName()



# Class fixtures
@pytest.fixture
def cluster_class(event_loop):
    return driver.Cluster


@pytest.fixture
def string_class():
    return properties.String


@pytest.fixture
def integer_class():
    return properties.Integer


@pytest.fixture
def historical_name_class():
    return HistoricalName


@pytest.fixture
def person_class():
    return Person


@pytest.fixture
def inherited_class():
    return Inherited


@pytest.fixture
def place_class():
    return Place


@pytest.fixture
def knows_class():
    return Knows


@pytest.fixture
def lives_in_class():
    return LivesIn


@pytest.fixture
def place_name_class():
    return PlaceName



@pytest.fixture
def string_class():
    return properties.String


@pytest.fixture
def integer_class():
    return properties.Integer


@pytest.fixture
def flt_class():
    return properties.Float


@pytest.fixture
def boolean_class():
    return properties.Boolean


@pytest.fixture(autouse=True)
def add_doctest_default(doctest_namespace, tmpdir, event_loop, app):
    doctest_namespace['Person'] = Person
    doctest_namespace['loop'] = event_loop
    doctest_namespace['app'] = app
    config = tmpdir.join('config.yml')
    config.write(
        "scheme: 'ws'\n"
        "hosts: ['localhost']\n"
        "port': 8182\n"
        "ssl_certfile: ''\n"
        "ssl_keyfile: ''\n"
        "ssl_password: ''\n"
        "username: ''\n"
        "password: ''\n"
        "response_timeout: None\n"
        "max_conns: 4\n"
        "min_conns: 1\n"
        "max_times_acquired: 16\n"
        "max_inflight: 64\n"
        "message_serializer: 'goblin.driver.GraphSONMessageSerializer'\n"
    )
    with tmpdir.as_cwd():
        yield


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 introduced 2020-01-30 17:37 UTC by Report Bug Copy Issue Report 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 2020-01-30 17:37 UTC by Report Bug Copy Issue Report 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 2020-01-30 17:37 UTC by Report Bug Copy Issue Report 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

goblin-ogm / goblin

Push — master ( c5801a...8fec65 )

conftest.knows_class() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like