test_pool

Complexity

Total Complexity

8

Size/Duplication

Total Lines	90
Duplicated Lines	0 %

Importance

Changes

0

7 Functions

Rating	Name	Duplication	Size	Complexity
A	test_acquire_multiple()	0	7	1
A	test_pool_init()	0	5	1
A	test_acquire_multiple_and_share()	0	13	1
A	test_max_acquired()	0	11	2
A	test_acquire_release()	0	11	1
A	test_release_notify()	0	17	1
A	test_share()	0	8	1

import asyncio

import pytest


@pytest.mark.asyncio
async def test_pool_init(connection_pool):
    await connection_pool.init_pool()
    assert len(connection_pool._available) == 1
    await connection_pool.close()


@pytest.mark.asyncio
async def test_acquire_release(connection_pool):
    conn = await connection_pool.acquire()
    assert not len(connection_pool._available)
    assert len(connection_pool._acquired) == 1
    assert conn.times_acquired == 1
    connection_pool.release(conn)
    assert len(connection_pool._available) == 1
    assert not len(connection_pool._acquired)
    assert not conn.times_acquired
    await connection_pool.close()


@pytest.mark.asyncio
async def test_acquire_multiple(connection_pool):
    conn1 = await connection_pool.acquire()
    conn2 = await connection_pool.acquire()
    assert conn1 is not conn2
    assert len(connection_pool._acquired) == 2
    await connection_pool.close()


@pytest.mark.asyncio
async def test_share(connection_pool):
    connection_pool._max_conns = 1
    conn1 = await connection_pool.acquire()
    conn2 = await connection_pool.acquire()
    assert conn1 is conn2
    assert conn1.times_acquired == 2
    await connection_pool.close()


@pytest.mark.asyncio
async def test_acquire_multiple_and_share(connection_pool):
    connection_pool._max_conns = 2
    connection_pool._max_times_acquired = 2
    conn1 = await connection_pool.acquire()
    conn2 = await connection_pool.acquire()
    assert conn1 is not conn2
    conn3 = await connection_pool.acquire()
    conn4 = await connection_pool.acquire()
    assert conn3 is not conn4
    assert conn3 is conn1
    assert conn4 is conn2
    await connection_pool.close()


@pytest.mark.asyncio
async def test_max_acquired(connection_pool):
    connection_pool._max_conns = 2
    connection_pool._max_times_acquired = 2
    conn1 = await connection_pool.acquire()
    conn2 = await connection_pool.acquire()
    conn3 = await connection_pool.acquire()
    conn4 = await connection_pool.acquire()
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(connection_pool.acquire(), timeout=0.1)
    await connection_pool.close()


@pytest.mark.asyncio
async def test_release_notify(connection_pool):
    connection_pool._max_conns = 2
    connection_pool._max_times_acquired = 2
    conn1 = await connection_pool.acquire()
    conn2 = await connection_pool.acquire()
    conn3 = await connection_pool.acquire()
    conn4 = await connection_pool.acquire()

    async def release(conn):
        conn.release()

    results = await asyncio.gather(
        *[connection_pool.acquire(), release(conn4)])
    conn4 = results[0]
    assert conn4 is conn2
    await connection_pool.close()