aio-libs / aio-libs/aiopg

Support Test Isolation

未關閉
#726 2 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
主要語言
Python
星號
1.4k
分支
170
PR 合併指標
30 天內沒有已合併 PR

描述

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

貢獻指南

開啟貢獻指南

評估

這個 Issue 還沒有評估資料。

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。