dbxs.testing.MemoryPool

class documentation

class MemoryPool: (source)

Constructor: MemoryPool.new(style)

An in-memory connection pool to an in-memory SQLite database which can be controlled a single operation at a time. Each operation that would normally be asynchronously dispatched to a thread can be invoked with the MemoryPool.pump and MemoryPool.flush methods.

Class Method	`new`	Create a synchronous memory connection pool.
Method	`additionalPump`	Add an additional callable to be called by `MemoryPool.pump` and `MemoryPool.flush`. This can be used to interleave other sources of in-memory event completion to allow test coroutines to complete, such as needing to call ...
Method	`flush`	Perform all outstanding steps of work.
Method	`pump`	Perform one step of pending work.
Instance Variable	`connectable`	The `AsyncConnectable` to be passed to the system under test.
Class Variable	`_performers`	Undocumented

@classmethod

def new(cls, style: Literal['named', 'qmark', 'named_dollar'] = 'qmark') -> MemoryPool: (source) ¶

Create a synchronous memory connection pool.

def additionalPump(self, f: Callable[[], bool]): (source) ¶

Add an additional callable to be called by MemoryPool.pump and MemoryPool.flush. This can be used to interleave other sources of in-memory event completion to allow test coroutines to complete, such as needing to call StubTreq.flush in a web application.

def flush(self) -> int: (source) ¶

Perform all outstanding steps of work.

Returns
`int`	a count of the number of steps of work performed.

def pump(self) -> bool: (source) ¶

Perform one step of pending work.

Returns
`bool`	True if any work was performed and False if no work was left.

connectable: AsyncConnectable = (source) ¶

The AsyncConnectable to be passed to the system under test.

_performers: list[Callable[[], bool]] = (source) ¶

Undocumented