Support Test Isolation

[dhofstetter]

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 supports such feature).

They describe their feature briefly here. 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

[nikita-krokosh]

@dhofstetter Have you found a solution for this?

[dhofstetter]

@nikita-krokosh Sorry for responding so late, but as far as I know I haven't found a solution to this problem.