Function test_named_savepoints_successful_exit

tests/test_transaction_async.py:367–418

Entering a transaction context will do one of these these things: 1. Begin an outer transaction (if one isn't already in progress) 2. Begin an outer transaction and create a savepoint (if one is named) 3. Create a savepoint (if a transaction is already in progress) either usi

(aconn, acommands)

async def test_named_savepoints_successful_exit(aconn, acommands):
    """
    Entering a transaction context will do one of these these things:
    1. Begin an outer transaction (if one isn't already in progress)
    2. Begin an outer transaction and create a savepoint (if one is named)
    3. Create a savepoint (if a transaction is already in progress)
       either using the name provided, or auto-generating a savepoint name.

    ...and exiting the context successfully will "commit" the same.
    """
    # Case 1
    # Using Transaction explicitly because conn.transaction() enters the contetx
    assert not acommands
    async with aconn.transaction() as tx:
        assert acommands.popall() == ["BEGIN"]
        assert not tx.savepoint_name
    assert acommands.popall() == ["COMMIT"]

    # Case 1 (with a transaction already started)
    await aconn.cursor().execute("select 1")
    assert acommands.popall() == ["BEGIN"]
    async with aconn.transaction() as tx:
        assert acommands.popall() == ['SAVEPOINT "_pg3_1"']
        assert tx.savepoint_name == "_pg3_1"

    assert acommands.popall() == ['RELEASE "_pg3_1"']
    await aconn.rollback()
    assert acommands.popall() == ["ROLLBACK"]

    # Case 2
    async with aconn.transaction(savepoint_name="foo") as tx:
        assert acommands.popall() == ["BEGIN", 'SAVEPOINT "foo"']
        assert tx.savepoint_name == "foo"
    assert acommands.popall() == ["COMMIT"]

    # Case 3 (with savepoint name provided)
    async with aconn.transaction():
        assert acommands.popall() == ["BEGIN"]
        async with aconn.transaction(savepoint_name="bar") as tx:
            assert acommands.popall() == ['SAVEPOINT "bar"']
            assert tx.savepoint_name == "bar"
        assert acommands.popall() == ['RELEASE "bar"']
    assert acommands.popall() == ["COMMIT"]

    # Case 3 (with savepoint name auto-generated)
    async with aconn.transaction():
        assert acommands.popall() == ["BEGIN"]
        async with aconn.transaction() as tx:
            assert acommands.popall() == ['SAVEPOINT "_pg3_2"']
            assert tx.savepoint_name == "_pg3_2"
        assert acommands.popall() == ['RELEASE "_pg3_2"']
    assert acommands.popall() == ["COMMIT"]


async def test_named_savepoints_exception_exit(aconn, acommands):