aio-libs / aio-libs/aiopg

Support Test Isolation

Aperta
#726 2 commenti 0 reazioni 0 assegnatari Vedi su GitHub
Lingua principale
Python
Stelle
1.4k
Fork
170
Metriche di merge delle PR
Nessuna PR unita negli ultimi 30g

Descrizione

Hello everybody,
I've seen a useful feature that might be useful for _aiopg_ as well. To write unit tests for the own project more easily, it would be great to offer some kind of Test Isolation (the library [databases](https://www.encode.io/databases) supports such feature).

They describe their feature briefly [here](https://www.encode.io/databases/tests_and_migrations/). What happens is, that in case of an boolean flag, it's ensured that every db statement is getting rolledback at the end (when the db connection get's closed). Even the usage of a "pool" is possible, as you simply receive the same connection all the time. The advantage is, that you can write unit tests, that somehow mutate data in the database that get rolled back in the end. This is a bit easier than clearing a test database all the time and insert the required data (which is indeed possible as well), but you can be sure that the database lasts in the same state then before.
So if you use this database for other purposes as well this is not a problem at all.

I tried to find a way to do such thing with aiopg as well but unfortunately I was not able to do so, as there are too much things I have to override. The problem is that the way a pool is instantiated, can not be changed by simply overriding the Pool class. The code provided was just a first simple sketch, I think there are more elegant, robust and complete ways of implementing such feature.

```
class TestPool(Pool):
"""
We override some methods here trying to return the same connection every time and encapsulate everything within
one single DB Transaction that gets rolledback on the end
"""

def __init__(self, dsn, minsize, maxsize, timeout, *, enable_json, enable_hstore, enable_uuid, echo, on_connect,
pool_recycle, **kwargs):

super().__init__(dsn, minsize, maxsize, timeout, enable_json=enable_json, enable_hstore=enable_hstore,
enable_uuid=enable_uuid, echo=echo, on_connect=on_connect, pool_recycle=pool_recycle, **kwargs)

self._conn = None

async def _acquire(self):

await self._begin()
return self._conn

async def wait_closed(self):

await self._rollback()
return await super().wait_closed()

async def _begin(self):

if self._conn is None:
self._conn = await super()._acquire()
"""
Now we try to start a transaction
"""
async with self._conn.cursor() as cur:
await cur.execute("BEGIN;")

async def _rollback(self):

if self._conn is not None:
"""
Now we are going to rollback the transaction
"""
async with self._conn.cursor() as cur:
await cur.execute("ROLLBACK;")

self._conn = None

def __del__(self):

await self._rollback()
super().__del__()
```

Maybe this feature is helpful for others as well.

Best regards

Daniel Hofstetter

Guida per i contributori

Apri la guida per i contributori

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.