gwp 0.2.1

A standalone, pure Rust gRPC wire protocol for GQL (ISO/IEC 39075)
Documentation 
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

"""Integration tests for transactions."""

import pytest

from gwp_py import GqlConnection


@pytest.mark.asyncio
async def test_begin_and_commit(test_server):
    async with await GqlConnection.connect(test_server) as conn:
        async with await conn.create_session() as session:
            tx = await session.begin_transaction()
            cursor = await tx.execute("INSERT INTO t VALUES (1)")
            await cursor.collect_rows()
            await tx.commit()


@pytest.mark.asyncio
async def test_begin_and_rollback(test_server):
    async with await GqlConnection.connect(test_server) as conn:
        async with await conn.create_session() as session:
            tx = await session.begin_transaction()
            cursor = await tx.execute("INSERT INTO t VALUES (1)")
            await cursor.collect_rows()
            await tx.rollback()


@pytest.mark.asyncio
async def test_transaction_context_manager_commit(test_server):
    """Transaction auto-commits on clean exit."""
    async with await GqlConnection.connect(test_server) as conn:
        async with await conn.create_session() as session:
            async with await session.begin_transaction() as tx:
                cursor = await tx.execute("MATCH (n) RETURN n")
                rows = await cursor.collect_rows()
                assert len(rows) == 2


@pytest.mark.asyncio
async def test_transaction_context_manager_rollback(test_server):
    """Transaction auto-rollbacks on exception."""
    async with await GqlConnection.connect(test_server) as conn:
        async with await conn.create_session() as session:
            with pytest.raises(ValueError):
                async with await session.begin_transaction() as tx:
                    await tx.execute("INSERT INTO t VALUES (1)")
                    raise ValueError("test error")


@pytest.mark.asyncio
async def test_transaction_execute_match(test_server):
    """Execute a MATCH within a transaction."""
    async with await GqlConnection.connect(test_server) as conn:
        async with await conn.create_session() as session:
            async with await session.begin_transaction() as tx:
                cursor = await tx.execute("MATCH (n:Person) RETURN n.name")
                cols = await cursor.column_names()
                assert "name" in cols
                rows = await cursor.collect_rows()
                assert len(rows) == 2