websauna.tests.model.test_retry module

Test SQL transaction conflict resolution.

class websauna.tests.model.test_retry.ConflictResolverThread(dbsession_factory)

Bases: threading.Thread

Launch two of these and they should cause database conflict and then retryable resolves it.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class websauna.tests.model.test_retry.ConflictThread(session_factory)

Bases: threading.Thread

Launch two of these and they should cause database conflict.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

websauna.tests.model.test_retry.dbsession_factory(test_request)

websauna.tests.model.test_retry.get_test_model()

websauna.tests.model.test_retry.test_conflict_resolved(test_instance, dbsession_factory, dbsession)

Use conflict resolver to resolve conflict between two transactions and see code retry is correctly run.

websauna.tests.model.test_retry.test_conflict_some_other_exception(dbsession)

See that unknown exceptions are correctly reraised by managed_transaction.

websauna.tests.model.test_retry.test_give_up(test_instance, dbsession_factory, dbsession)

See that the conflict resolver gives up after using given number of attempts to replay transactions.

websauna.tests.model.test_retry.test_instance(dbsession)

websauna.tests.model.test_retry.test_sql_transaction_conflict(test_instance, dbsession_factory)

Run database to a transaction conflict and see what exception it spits out, and make sure we know this is the exception we expect.