Search code examples
pythonpytestfixturespython-3.9

Disconnect fixture's connection at the end of test module in PyTest


I have a test module running with PyTest.

A fixture establishes a connection to a Redis client and flushes all at the end of each test:

@pytest.fixture
def redis_conn():
    conn = redis.Redis(decode_responses=True, **config.redis_test_credentials)
    yield conn
    conn.flushall()

In addition to that, I need to call this after all tests from the module have finished:

conn.connection.disconnect()

Some things I've thought about and why they didn't work:

  • I can't use a pytest.fixture(scope='module') here because then conn.flushall() would run only after this module's tests are all done

  • I also don't see how to use pytest_unconfigure() because I don't see a way to access the conn object from there.

How to make sure that conn.connection.disconnect() is executed after all tests from the module are done, while keeping conn.flushall() after each test?

-- EDIT

There is one additional constraint that I omitted, which is that the redis_conn fixture is used into a function-level mock:

@pytest.fixture
def mock_redis_conn(mocker, redis_conn):
    """ Mock Redis connection """
    mocker.patch("mymodule.api.redis_conn", new=redis_conn)

This mocked mymodule.api.redis_conn should effectively call flushall() after each test is run, which prevents me from scoping this mock to the module level.


Solution

  • You could implement fixtures which depends on others fixtures.

    from unittest.mock import MagicMock
    
    import pytest
    
    class RedisConn:
        """just stub for example"""
        def flush(self):
            raise NotImplementedError()
    
    connection = RedisConn()
    
    @pytest.fixture(scope="module")
    def conn():
        conn = MagicMock()  # here you open connection to testing Redis instance
        print("open connection")
        yield conn
        print("close connection")
    
    @pytest.fixture(scope="function")
    def flush(conn, mocker):
        mocker.patch(f"{__name__}.connection", new=conn)
        print("do nothing")
        yield
        print(f"flush {connection}")
        connection.flush()
    
    def test_main_1(flush):
        assert isinstance(connection, MagicMock)
        print("test 1 body")
    
    def test_main_2(flush):
        assert isinstance(connection, MagicMock)
        print("test 2 body")
    
    def test_main_3():
        assert not isinstance(connection, MagicMock)
        assert isinstance(connection, RedisConn)
    
    
    
    if __name__ == "__main__":
        pytest.main([__file__, "-s"])
    

    prints

    open connection
    do nothing
    test 1 body
    .
    flush <MagicMock id='139710977083536'>
    do nothing
    test 2 body
    .
    flush <MagicMock id='139710977083536'>
    .
    close connection