RegisteredTrx maintains transaction fault injection state to test transaction failure code path.
You may enable specific transaction failure calling enable_trx_failure on that transaction name. After that, the transaction's Commit method will throw an exception.
If you don't need a fault injection anymore (e.g. you want to test a successfull retry), you may call disable_trx_failure afterwards.
Example usage:
async def test_trx_fail(service_client, pgsql, userver_pg_trx):
response = await service_client.delete('/v1/key-value?key=foo')
assert response.status == 200
userver_pg_trx.enable_failure('sample_transaction_insert_key_value')
response = await service_client.post('/v1/key-value?key=foo&value=bar')
assert response.status == 500
response = await service_client.get('/v1/key-value?key=foo')
assert response.status == 404
Definition at line 1 of file sql.py.
