test_graph

Complexity

Total Complexity

5

Size/Duplication

Total Lines	56
Duplicated Lines	0 %

Importance

Changes

0

3 Functions

Rating	Name	Duplication	Size	Complexity
A	test_side_effects()	0	23	2
A	test_submit_traversal()	0	12	1
A	test_generate_traversal()	0	8	2

import pytest
from aiogremlin import process
from gremlin_python.process.traversal import Binding, T

from goblin import driver


@pytest.mark.asyncio
async def test_generate_traversal(remote_graph, remote_connection):
    async with remote_connection:
        g = remote_graph.traversal().withRemote(remote_connection)
        traversal = g.V().hasLabel(('v1', 'person'))
        assert isinstance(traversal,
                          process.graph_traversal.AsyncGraphTraversal)
        assert traversal.bytecode.step_instructions[1][1][1] == 'person'


@pytest.mark.asyncio
async def test_submit_traversal(remote_graph, remote_connection):
    g = remote_graph.traversal().withRemote(remote_connection)
    resp = g.addV('person').property('name', 'leifur').valueMap(True)
    leif = await resp.next()
    assert leif['name'][0] == 'leifur'
    assert leif[T.label] == 'person'
    resp = g.V(Binding('vid', leif[T.id])).drop()
    none = await resp.next()
    assert none is None

    await remote_connection.close()


#@pytest.mark.skipif(pytest.config.getoption('provider') == 'dse', reason="need custom alias")
@pytest.mark.asyncio
async def test_side_effects(remote_graph, remote_connection):
    async with remote_connection:
        remote_connection._message_serializer = \
            driver.GraphSONMessageSerializer
        g = remote_graph.traversal().withRemote(remote_connection)
        # create some nodes
        resp = g.addV('person').property('name', 'leifur')
        leif = await resp.next()
        resp = g.addV('person').property('name', 'dave')
        dave = await resp.next()

        resp = g.addV('person').property('name', 'jon')
        jonthan = await resp.next()

        traversal = g.V().aggregate('a').aggregate('b')
        await traversal.iterate()
        keys = await traversal.side_effects.keys()
        assert keys == set(['a', 'b'])
        side_effects = await traversal.side_effects.get('a')
        assert side_effects
        side_effects = await traversal.side_effects.get('b')
        assert side_effects