Function test_named_savepoints_successful_exit

tests/test_transaction.py:363–414

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

(conn, commands)

def test_named_savepoints_successful_exit(conn, commands):
    """
    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 commands
    with conn.transaction() as tx:
        assert commands.popall() == ["BEGIN"]
        assert not tx.savepoint_name
    assert commands.popall() == ["COMMIT"]

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

    assert commands.popall() == ['RELEASE "_pg3_1"']
    conn.rollback()
    assert commands.popall() == ["ROLLBACK"]

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

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

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


def test_named_savepoints_exception_exit(conn, commands):