test_session.TestCreationApi.test_create_edge_no_source() - Code Metrics - goblin-ogm/goblin - Measure and Improve Code Quality continuously with Scrutinizer

TestCreationApi.test_create_edge_no_source() A
last analyzed 2022-05-13 18:52 UTC

↳ Parent: test_session

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	4
dl	0
loc	7
rs	10
c	0
b	0
f	0

"""Functional sessions tests"""

import pytest
from gremlin_python.process.traversal import Binding

from goblin import element
from goblin.session import bindprop


def test_bindprop(person_class):
    db_val, (binding, val) = bindprop(
        person_class, 'name', 'dave', binding='n1')
    assert db_val == 'name'
    assert binding == 'n1'
    assert val == 'dave'


class TestCreationApi:
    @pytest.mark.asyncio
    async def test_create_vertex(self, app, person_class):
        session = await app.session()
        jon = person_class()
        jon.name = 'jonathan'
        jon.age = 38
        leif = person_class()
        leif.name = 'leifur'
        leif.age = 28
        session.add(jon, leif)
        assert not hasattr(jon, 'id')
        assert not hasattr(leif, 'id')
        await session.flush()
        assert hasattr(jon, 'id')
        assert session.current[app._get_hashable_id(jon.id)] is jon
        assert jon.name == 'jonathan'
        assert jon.age == 38
        assert hasattr(leif, 'id')
        assert session.current[app._get_hashable_id(leif.id)] is leif
        assert leif.name == 'leifur'
        assert leif.age == 28
        await app.close()

    @pytest.mark.asyncio
    async def test_create_inherited(self, app, inherited_class):
        session = await app.session()
        jon = inherited_class()
        jon.name = 'jonathan'
        jon.age = 38
        session.add(jon)
        assert not hasattr(jon, 'id')
        await session.flush()
        assert hasattr(jon, 'id')
        assert session.current[app._get_hashable_id(jon.id)] is jon
        assert jon.name == 'jonathan'
        assert jon.age == 38
        await app.close()

    @pytest.mark.skip(reason="Test not working on all backends")
    @pytest.mark.asyncio
    async def test_create_edge(self, app, person_class, place_class,
                               lives_in_class):
        session = await app.session()
        jon = person_class()
        jon.name = 'jonathan'
        jon.age = 38
        montreal = place_class()
        montreal.name = 'Montreal'
        lives_in = lives_in_class(jon, montreal)
        session.add(jon, montreal, lives_in)
        await session.flush()
        assert hasattr(lives_in, 'id')
        assert session.current[app._get_hashable_id(lives_in.id)] is lives_in
        assert lives_in.source is jon
        assert lives_in.target is montreal
        # Not supported by current graphson version
        assert lives_in.source.__label__ == 'person'
        assert lives_in.target.__label__ == 'place'
        await app.close()

    @pytest.mark.asyncio
    async def test_create_edge_no_source(self, app, lives_in, person):
        session = await app.session()
        lives_in.source = person
        with pytest.raises(Exception):
            await session.save(lives_in)
        await app.close()

    @pytest.mark.asyncio
    async def test_create_edge_no_target(self, app, lives_in, place):
        session = await app.session()
        lives_in.target = place
        with pytest.raises(Exception):
            await session.save(lives_in)
        await app.close()

    @pytest.mark.asyncio
    async def test_create_edge_no_source_target(self, app, lives_in):
        session = await app.session()
        with pytest.raises(Exception):
            await session.save(lives_in)
        await app.close()

    @pytest.mark.asyncio
    async def test_get_vertex(self, app, person_class):
        session = await app.session()
        jon = person_class()
        jon.name = 'jonathan'
        jon.age = 38
        await session.save(jon)
        jid = jon.id
        result = await session.get_vertex(jon)
        assert result.id == jid
        assert result is jon
        await app.close()

    @pytest.mark.asyncio
    async def test_get_edge(self, app, person_class, place_class,
                            lives_in_class):
        session = await app.session()
        jon = person_class()
        jon.name = 'jonathan'
        jon.age = 38
        montreal = place_class()
        montreal.name = 'Montreal'
        lives_in = lives_in_class(jon, montreal)
        session.add(jon, montreal, lives_in)
        await session.flush()
        lid = lives_in.id
        result = await session.get_edge(lives_in)
        assert result.id == lid
        assert result is lives_in
        await app.close()

    @pytest.mark.asyncio
    async def test_get_vertex_doesnt_exist(self, app, person):
        session = await app.session()
        person._id = 100000
        result = await session.get_vertex(person)
        assert not result
        await app.close()

    @pytest.mark.asyncio
    async def test_get_edge_doesnt_exist(self, app, knows, person_class):
        session = await app.session()
        jon = person_class()
        leif = person_class()
        works_with = knows
        works_with.source = jon
        works_with.target = leif
        works_with._id = 1000000
        result = await session.get_edge(works_with)
        assert not result
        await app.close()

    @pytest.mark.asyncio
    async def test_remove_vertex(self, app, person):
        session = await app.session()
        person.name = 'dave'
        person.age = 35
        await session.save(person)
        result = await session.g.V(Binding('vid', person.id)).next()
        assert result is person
        rid = result.id
        await session.remove_vertex(person)
        result = await session.g.V(Binding('rid', rid)).next()
        assert not result
        assert rid not in session.current
        await app.close()

    @pytest.mark.asyncio
    async def test_remove_vertex_foreign_session(self, app, person):
        session1 = await app.session()
        session2 = await app.session()
        person.name = 'dave'
        person.age = 35
        await session1.save(person)
        result = await session1.g.V(Binding('vid', person.id)).next()
        assert result is person
        rid = result.id
        await session2.remove_vertex(person)
        result = await session2.g.V(Binding('rid', rid)).next()
        assert not result
        result = await session1.g.V(Binding('vid', person.id)).next()
        assert result is None
        # This is why we warn
        assert rid in session1.current
        assert person is session1.current[rid]
        await app.close()

    @pytest.mark.asyncio
    async def test_remove_edge(self, app, person_class, place_class,
                               lives_in_class):
        session = await app.session()
        jon = person_class()
        jon.name = 'jonathan'
        jon.age = 38
        montreal = place_class()
        montreal.name = 'Montreal'
        lives_in = lives_in_class(jon, montreal)
        session.add(jon, montreal, lives_in)
        await session.flush()
        result = await session.g.E(Binding('eid', lives_in.id)).next()
        assert result is lives_in
        rid = result.id
        await session.remove_edge(lives_in)
        result = await session.g.E(Binding('rid', rid)).next()
        assert not result
        await app.close()

    @pytest.mark.asyncio
    async def test_remove_edge_foreign_session(self, app, person_class,
                                               place_class, lives_in_class):
        session1 = await app.session()
        session2 = await app.session()
        jon = person_class()
        jon.name = 'jonathan'
        jon.age = 38
        montreal = place_class()
        montreal.name = 'Montreal'
        lives_in = lives_in_class(jon, montreal)
        session1.add(jon, montreal, lives_in)
        await session1.flush()
        result = await session1.g.E(Binding('eid', lives_in.id)).next()
        assert result is lives_in
        rid = result.id
        await session2.remove_edge(lives_in)
        result = await session1.g.E(Binding('rid', rid)).next()
        assert not result
        # This is why whe warm
        assert rid in session1.current
        assert lives_in is session1.current[rid]
        await app.close()

    @pytest.mark.asyncio
    async def test_update_vertex(self, app, person):
        session = await app.session()
        person.name = 'dave'
        person.age = 35
        result = await session.save(person)
        assert result.name == 'dave'
        assert result.age == 35
        person.name = 'david'
        person.age = None
        result = await session.save(person)
        assert result is person
        assert result.name == 'david'
        assert not result.age
        await app.close()

    #@pytest.mark.skipif(pytest.config.getoption('provider') == 'dse', reason='DSE')
    @pytest.mark.asyncio
    async def test_update_edge(self, app, person_class, knows):
        session = await app.session()
        dave = person_class()
        leif = person_class()
        knows.source = dave
        knows.target = leif
        knows.notes = 'online'
        session.add(dave, leif)
        await session.flush()
        result = await session.save(knows)
        assert knows.notes == 'online'
        knows.notes = None
        result = await session.save(knows)
        assert result is knows
        assert not result.notes
        await app.close()


class TestTraversalApi:
    @pytest.mark.asyncio
    async def test_all(self, app, person_class):
        session = await app.session()
        dave = person_class()
        leif = person_class()
        jon = person_class()
        session.add(dave, leif, jon)
        await session.flush()
        resp = session.traversal(person_class)
        results = []
        async for msg in resp:
            results.append(msg)
        assert results
        await app.close()

    @pytest.mark.asyncio
    async def test_next_one(self, app, person_class):
        session = await app.session()
        dave = person_class()
        leif = person_class()
        jon = person_class()
        session.add(dave, leif, jon)
        await session.flush()
        resp = await session.traversal(person_class).next()
        assert isinstance(resp, person_class)
        await app.close()

    @pytest.mark.asyncio
    async def test_traversal_bindprop(self, app, person_class):
        session = await app.session()
        itziri = person_class()
        itziri.name = 'itziri'
        result1 = await session.save(itziri)
        bound_name = bindprop(person_class, 'name', 'itziri', binding='v1')
        p1 = await session.traversal(person_class).has(*bound_name).next()
        await app.close()

    @pytest.mark.asyncio
    async def test_next_none(self, app):
        session = await app.session()
        none = await session.g.V().hasLabel('a very unlikey label').next()
        assert not none
        await app.close()

    @pytest.mark.asyncio
    async def test_vertex_deserialization(self, app, person_class):
        session = await app.session()
        resp = await session.g.addV('person').property(person_class.name,
                                                       'leif').property(
                                                           'place_of_birth',
                                                           'detroit').next()
        assert isinstance(resp, person_class)
        assert resp.name == 'leif'
        assert resp.place_of_birth == 'detroit'
        await app.close()

    @pytest.mark.skip(reason="Test not working on all backends")
    @pytest.mark.asyncio
    async def test_edge_desialization(self, app, knows_class):
        session = await app.session()
        p1 = await session.g.addV('person').next()
        p2 = await session.g.addV('person').next()
        e1 = await session.g.V(Binding('p1_id', p1.id)).addE('knows').to(
            session.g.V(Binding('p2_id', p2.id))).property(
                knows_class.notes, 'somehow').property('how_long', 1).next()
        assert isinstance(e1, knows_class)
        assert e1.notes == 'somehow'
        assert e1.how_long == 1
        await app.close()

    @pytest.mark.asyncio
    async def test_unregistered_vertex_deserialization(self, app):
        session = await app.session()
        dave = await session.g.addV('unregistered').property('name',
                                                             'dave').next()
        assert isinstance(dave, element.GenericVertex)
        assert dave.name == 'dave'
        assert dave.__label__ == 'unregistered'
        await app.close()

    @pytest.mark.asyncio
    async def test_unregistered_edge_desialization(self, app):
        session = await app.session()
        p1 = await session.g.addV('person').next()
        p2 = await session.g.addV('person').next()
        e1 = await session.g.V(Binding(
            'p1_id', p1.id)).addE('unregistered').to(
                session.g.V(Binding('p2_id', p2.id))).property('how_long',
                                                               1).next()
        assert isinstance(e1, element.GenericEdge)
        assert e1.how_long == 1
        assert e1.__label__ == 'unregistered'
        await app.close()

    @pytest.mark.asyncio
    async def test_property_deserialization(self, app):
        # In a future version this should deserialize to a goblin vert prop???
        session = await app.session()
        p1 = await session.g.addV('person').property('name', 'leif').next()
        traversal = session.g.V(Binding('p1_id', p1.id)).properties('name')
        name = await traversal.next()
        assert name.value == 'leif'
        assert name.label == 'name'
        await app.close()

    @pytest.mark.asyncio
    async def test_non_element_deserialization(self, app):
        session = await app.session()
        p1 = await session.g.addV('person').property('name', 'leif').next()
        one = await session.g.V(Binding('p1_id', p1.id)).count().next()
        assert one == 1
        await app.close()

    @pytest.mark.asyncio
    async def test_deserialize_nested_map(self, app, person_class):
        session = await app.session()
        await session.g.addV('person').property(person_class.name,
                                                'leif').property(
                                                    'place_of_birth',
                                                    'detroit').next()

        await session.g.addV('person').property(person_class.name,
                                                'David').property(
                                                    person_class.nicknames,
                                                    'davebshow').next()
        traversal = session.g.V().hasLabel('person').as_('x').valueMap().as_(
            'y').select('x', 'y').fold()
        resp = await traversal.next()
        for item in resp:
            # FIXME BROKEN
            # assert isinstance(item['x'], person_class)
            assert isinstance(item['y'], dict)
        await app.close()


1			"""Functional sessions tests"""
2
3			import pytest
4			from gremlin_python.process.traversal import Binding
5
6			from goblin import element
7			from goblin.session import bindprop
8
9
10			def test_bindprop(person_class):
11			db_val, (binding, val) = bindprop(
12			person_class, 'name', 'dave', binding='n1')
13			assert db_val == 'name'
14			assert binding == 'n1'
15			assert val == 'dave'
16
17
18			class TestCreationApi:
19			@pytest.mark.asyncio
20			async def test_create_vertex(self, app, person_class):
21			session = await app.session()
22			jon = person_class()
23			jon.name = 'jonathan'
24			jon.age = 38
25			leif = person_class()
26			leif.name = 'leifur'
27			leif.age = 28
28			session.add(jon, leif)
29			assert not hasattr(jon, 'id')
30			assert not hasattr(leif, 'id')
31			await session.flush()
32			assert hasattr(jon, 'id')
33			assert session.current[app._get_hashable_id(jon.id)] is jon
34			assert jon.name == 'jonathan'
35			assert jon.age == 38
36			assert hasattr(leif, 'id')
37			assert session.current[app._get_hashable_id(leif.id)] is leif
38			assert leif.name == 'leifur'
39			assert leif.age == 28
40			await app.close()
41
42			@pytest.mark.asyncio
43			async def test_create_inherited(self, app, inherited_class):
44			session = await app.session()
45			jon = inherited_class()
46			jon.name = 'jonathan'
47			jon.age = 38
48			session.add(jon)
49			assert not hasattr(jon, 'id')
50			await session.flush()
51			assert hasattr(jon, 'id')
52			assert session.current[app._get_hashable_id(jon.id)] is jon
53			assert jon.name == 'jonathan'
54			assert jon.age == 38
55			await app.close()
56
57			@pytest.mark.skip(reason="Test not working on all backends")
58			@pytest.mark.asyncio
59			async def test_create_edge(self, app, person_class, place_class,
60			lives_in_class):
61			session = await app.session()
62			jon = person_class()
63			jon.name = 'jonathan'
64			jon.age = 38
65			montreal = place_class()
66			montreal.name = 'Montreal'
67			lives_in = lives_in_class(jon, montreal)
68			session.add(jon, montreal, lives_in)
69			await session.flush()
70			assert hasattr(lives_in, 'id')
71			assert session.current[app._get_hashable_id(lives_in.id)] is lives_in
72			assert lives_in.source is jon
73			assert lives_in.target is montreal
74			# Not supported by current graphson version
75			assert lives_in.source.__label__ == 'person'
76			assert lives_in.target.__label__ == 'place'
77			await app.close()
78
79			@pytest.mark.asyncio
80			async def test_create_edge_no_source(self, app, lives_in, person):
81			session = await app.session()
82			lives_in.source = person
83			with pytest.raises(Exception):
84			await session.save(lives_in)
85			await app.close()
86
87			@pytest.mark.asyncio
88			async def test_create_edge_no_target(self, app, lives_in, place):
89			session = await app.session()
90			lives_in.target = place
91			with pytest.raises(Exception):
92			await session.save(lives_in)
93			await app.close()
94
95			@pytest.mark.asyncio
96			async def test_create_edge_no_source_target(self, app, lives_in):
97			session = await app.session()
98			with pytest.raises(Exception):
99			await session.save(lives_in)
100			await app.close()
101
102			@pytest.mark.asyncio
103			async def test_get_vertex(self, app, person_class):
104			session = await app.session()
105			jon = person_class()
106			jon.name = 'jonathan'
107			jon.age = 38
108			await session.save(jon)
109			jid = jon.id
110			result = await session.get_vertex(jon)
111			assert result.id == jid
112			assert result is jon
113			await app.close()
114
115			@pytest.mark.asyncio
116			async def test_get_edge(self, app, person_class, place_class,
117			lives_in_class):
118			session = await app.session()
119			jon = person_class()
120			jon.name = 'jonathan'
121			jon.age = 38
122			montreal = place_class()
123			montreal.name = 'Montreal'
124			lives_in = lives_in_class(jon, montreal)
125			session.add(jon, montreal, lives_in)
126			await session.flush()
127			lid = lives_in.id
128			result = await session.get_edge(lives_in)
129			assert result.id == lid
130			assert result is lives_in
131			await app.close()
132
133			@pytest.mark.asyncio
134			async def test_get_vertex_doesnt_exist(self, app, person):
135			session = await app.session()
136			person._id = 100000
137			result = await session.get_vertex(person)
138			assert not result
139			await app.close()
140
141			@pytest.mark.asyncio
142			async def test_get_edge_doesnt_exist(self, app, knows, person_class):
143			session = await app.session()
144			jon = person_class()
145			leif = person_class()
146			works_with = knows
147			works_with.source = jon
148			works_with.target = leif
149			works_with._id = 1000000
150			result = await session.get_edge(works_with)
151			assert not result
152			await app.close()
153
154			@pytest.mark.asyncio
155			async def test_remove_vertex(self, app, person):
156			session = await app.session()
157			person.name = 'dave'
158			person.age = 35
159			await session.save(person)
160			result = await session.g.V(Binding('vid', person.id)).next()
161			assert result is person
162			rid = result.id
163			await session.remove_vertex(person)
164			result = await session.g.V(Binding('rid', rid)).next()
165			assert not result
166			assert rid not in session.current
167			await app.close()
168
169			@pytest.mark.asyncio
170			async def test_remove_vertex_foreign_session(self, app, person):
171			session1 = await app.session()
172			session2 = await app.session()
173			person.name = 'dave'
174			person.age = 35
175			await session1.save(person)
176			result = await session1.g.V(Binding('vid', person.id)).next()
177			assert result is person
178			rid = result.id
179			await session2.remove_vertex(person)
180			result = await session2.g.V(Binding('rid', rid)).next()
181			assert not result
182			result = await session1.g.V(Binding('vid', person.id)).next()
183			assert result is None
184			# This is why we warn
185			assert rid in session1.current
186			assert person is session1.current[rid]
187			await app.close()
188
189			@pytest.mark.asyncio
190			async def test_remove_edge(self, app, person_class, place_class,
191			lives_in_class):
192			session = await app.session()
193			jon = person_class()
194			jon.name = 'jonathan'
195			jon.age = 38
196			montreal = place_class()
197			montreal.name = 'Montreal'
198			lives_in = lives_in_class(jon, montreal)
199			session.add(jon, montreal, lives_in)
200			await session.flush()
201			result = await session.g.E(Binding('eid', lives_in.id)).next()
202			assert result is lives_in
203			rid = result.id
204			await session.remove_edge(lives_in)
205			result = await session.g.E(Binding('rid', rid)).next()
206			assert not result
207			await app.close()
208
209			@pytest.mark.asyncio
210			async def test_remove_edge_foreign_session(self, app, person_class,
211			place_class, lives_in_class):
212			session1 = await app.session()
213			session2 = await app.session()
214			jon = person_class()
215			jon.name = 'jonathan'
216			jon.age = 38
217			montreal = place_class()
218			montreal.name = 'Montreal'
219			lives_in = lives_in_class(jon, montreal)
220			session1.add(jon, montreal, lives_in)
221			await session1.flush()
222			result = await session1.g.E(Binding('eid', lives_in.id)).next()
223			assert result is lives_in
224			rid = result.id
225			await session2.remove_edge(lives_in)
226			result = await session1.g.E(Binding('rid', rid)).next()
227			assert not result
228			# This is why whe warm
229			assert rid in session1.current
230			assert lives_in is session1.current[rid]
231			await app.close()
232
233			@pytest.mark.asyncio
234			async def test_update_vertex(self, app, person):
235			session = await app.session()
236			person.name = 'dave'
237			person.age = 35
238			result = await session.save(person)
239			assert result.name == 'dave'
240			assert result.age == 35
241			person.name = 'david'
242			person.age = None
243			result = await session.save(person)
244			assert result is person
245			assert result.name == 'david'
246			assert not result.age
247			await app.close()
248
249			#@pytest.mark.skipif(pytest.config.getoption('provider') == 'dse', reason='DSE')
250			@pytest.mark.asyncio
251			async def test_update_edge(self, app, person_class, knows):
252			session = await app.session()
253			dave = person_class()
254			leif = person_class()
255			knows.source = dave
256			knows.target = leif
257			knows.notes = 'online'
258			session.add(dave, leif)
259			await session.flush()
260			result = await session.save(knows)
261			assert knows.notes == 'online'
262			knows.notes = None
263			result = await session.save(knows)
264			assert result is knows
265			assert not result.notes
266			await app.close()
267
268
269			class TestTraversalApi:
270			@pytest.mark.asyncio
271			async def test_all(self, app, person_class):
272			session = await app.session()
273			dave = person_class()
274			leif = person_class()
275			jon = person_class()
276			session.add(dave, leif, jon)
277			await session.flush()
278			resp = session.traversal(person_class)
279			results = []
280			async for msg in resp:
281			results.append(msg)
282			assert results
283			await app.close()
284
285			@pytest.mark.asyncio
286			async def test_next_one(self, app, person_class):
287			session = await app.session()
288			dave = person_class()
289			leif = person_class()
290			jon = person_class()
291			session.add(dave, leif, jon)
292			await session.flush()
293			resp = await session.traversal(person_class).next()
294			assert isinstance(resp, person_class)
295			await app.close()
296
297			@pytest.mark.asyncio
298			async def test_traversal_bindprop(self, app, person_class):
299			session = await app.session()
300			itziri = person_class()
301			itziri.name = 'itziri'
302			result1 = await session.save(itziri)
303			bound_name = bindprop(person_class, 'name', 'itziri', binding='v1')
304			p1 = await session.traversal(person_class).has(*bound_name).next()
305			await app.close()
306
307			@pytest.mark.asyncio
308			async def test_next_none(self, app):
309			session = await app.session()
310			none = await session.g.V().hasLabel('a very unlikey label').next()
311			assert not none
312			await app.close()
313
314			@pytest.mark.asyncio
315			async def test_vertex_deserialization(self, app, person_class):
316			session = await app.session()
317			resp = await session.g.addV('person').property(person_class.name,
318			'leif').property(
319			'place_of_birth',
320			'detroit').next()
321			assert isinstance(resp, person_class)
322			assert resp.name == 'leif'
323			assert resp.place_of_birth == 'detroit'
324			await app.close()
325
326			@pytest.mark.skip(reason="Test not working on all backends")
327			@pytest.mark.asyncio
328			async def test_edge_desialization(self, app, knows_class):
329			session = await app.session()
330			p1 = await session.g.addV('person').next()
331			p2 = await session.g.addV('person').next()
332			e1 = await session.g.V(Binding('p1_id', p1.id)).addE('knows').to(
333			session.g.V(Binding('p2_id', p2.id))).property(
334			knows_class.notes, 'somehow').property('how_long', 1).next()
335			assert isinstance(e1, knows_class)
336			assert e1.notes == 'somehow'
337			assert e1.how_long == 1
338			await app.close()
339
340			@pytest.mark.asyncio
341			async def test_unregistered_vertex_deserialization(self, app):
342			session = await app.session()
343			dave = await session.g.addV('unregistered').property('name',
344			'dave').next()
345			assert isinstance(dave, element.GenericVertex)
346			assert dave.name == 'dave'
347			assert dave.__label__ == 'unregistered'
348			await app.close()
349
350			@pytest.mark.asyncio
351			async def test_unregistered_edge_desialization(self, app):
352			session = await app.session()
353			p1 = await session.g.addV('person').next()
354			p2 = await session.g.addV('person').next()
355			e1 = await session.g.V(Binding(
356			'p1_id', p1.id)).addE('unregistered').to(
357			session.g.V(Binding('p2_id', p2.id))).property('how_long',
358			1).next()
359			assert isinstance(e1, element.GenericEdge)
360			assert e1.how_long == 1
361			assert e1.__label__ == 'unregistered'
362			await app.close()
363
364			@pytest.mark.asyncio
365			async def test_property_deserialization(self, app):
366			# In a future version this should deserialize to a goblin vert prop???
367			session = await app.session()
368			p1 = await session.g.addV('person').property('name', 'leif').next()
369			traversal = session.g.V(Binding('p1_id', p1.id)).properties('name')
370			name = await traversal.next()
371			assert name.value == 'leif'
372			assert name.label == 'name'
373			await app.close()
374
375			@pytest.mark.asyncio
376			async def test_non_element_deserialization(self, app):
377			session = await app.session()
378			p1 = await session.g.addV('person').property('name', 'leif').next()
379			one = await session.g.V(Binding('p1_id', p1.id)).count().next()
380			assert one == 1
381			await app.close()
382
383			@pytest.mark.asyncio
384			async def test_deserialize_nested_map(self, app, person_class):
385			session = await app.session()
386			await session.g.addV('person').property(person_class.name,
387			'leif').property(
388			'place_of_birth',
389			'detroit').next()
390
391			await session.g.addV('person').property(person_class.name,
392			'David').property(
393			person_class.nicknames,
394			'davebshow').next()
395			traversal = session.g.V().hasLabel('person').as_('x').valueMap().as_(
396			'y').select('x', 'y').fold()
397			resp = await traversal.next()
398			for item in resp:
399			# FIXME BROKEN
400			# assert isinstance(item['x'], person_class)
401			assert isinstance(item['y'], dict)
402			await app.close()
403

goblin-ogm / goblin

TestCreationApi.test_create_edge_no_source() A last analyzed 2022-05-13 18:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestCreationApi.test_create_edge_no_source() A
last analyzed 2022-05-13 18:52 UTC