How to use the bonsai.pool.ConnectionPool function in bonsai

To help you get started, we’ve selected a few bonsai examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github noirello / bonsai / tests / test_pool.py View on Github external
def test_init():
    """ Test pool initialisation. """
    cli = LDAPClient("ldap://dummy.nfo")
    with pytest.raises(ValueError):
        _ = ConnectionPool(cli, minconn=-3)
    with pytest.raises(ValueError):
        _ = ConnectionPool(cli, minconn=5, maxconn=3)
    pool = ConnectionPool(cli, minconn=2, maxconn=5)
    assert pool.closed == True
    assert pool.empty == False
    assert pool.max_connection == 5
    assert pool.shared_connection == 0
    assert pool.idle_connection == 0
github noirello / bonsai / tests / test_pool.py View on Github external
def test_init():
    """ Test pool initialisation. """
    cli = LDAPClient("ldap://dummy.nfo")
    with pytest.raises(ValueError):
        _ = ConnectionPool(cli, minconn=-3)
    with pytest.raises(ValueError):
        _ = ConnectionPool(cli, minconn=5, maxconn=3)
    pool = ConnectionPool(cli, minconn=2, maxconn=5)
    assert pool.closed == True
    assert pool.empty == False
    assert pool.max_connection == 5
    assert pool.shared_connection == 0
    assert pool.idle_connection == 0
github noirello / bonsai / tests / test_pool.py View on Github external
def test_init():
    """ Test pool initialisation. """
    cli = LDAPClient("ldap://dummy.nfo")
    with pytest.raises(ValueError):
        _ = ConnectionPool(cli, minconn=-3)
    with pytest.raises(ValueError):
        _ = ConnectionPool(cli, minconn=5, maxconn=3)
    pool = ConnectionPool(cli, minconn=2, maxconn=5)
    assert pool.closed == True
    assert pool.empty == False
    assert pool.max_connection == 5
    assert pool.shared_connection == 0
    assert pool.idle_connection == 0
github noirello / bonsai / tests / test_pool.py View on Github external
def test_get(client):
    """ Test getting a connection from the pool. """
    pool = ConnectionPool(client, minconn=1, maxconn=2)
    with pytest.raises(ClosedPool):
        _ = pool.get()
    pool.open()
    assert pool.max_connection == 2
    assert pool.idle_connection == 1
    assert pool.shared_connection == 0
    conn1 = pool.get()
    assert conn1 is not None
    assert conn1.closed == False
    assert pool.idle_connection == 0
    assert pool.shared_connection == 1
    conn2 = pool.get()
    assert pool.idle_connection == 0
    assert pool.shared_connection == 2
    with pytest.raises(EmptyPool):
        _ = pool.get()
github noirello / bonsai / tests / test_pool.py View on Github external
def test_open(client):
    """ Test opening the pool. """
    pool = ConnectionPool(client, minconn=5)
    pool.open()
    assert pool.closed != True
    assert pool.idle_connection == 5
github noirello / bonsai / tests / test_pool.py View on Github external
def test_put(client):
    """ Test putting connection back into the pool. """
    pool = ConnectionPool(client, minconn=1, maxconn=1)
    with pytest.raises(ClosedPool):
        pool.put(None)
    pool.open()
    conn = pool.get()
    assert pool.idle_connection == 0
    pool.put(conn)
    assert pool.idle_connection == 1
    other_conn = client.connect()
    with pytest.raises(PoolError):
        pool.put(other_conn)
github noirello / bonsai / tests / test_pool.py View on Github external
def test_max_connection():
    """ Test max_connection property. """
    cli = LDAPClient("ldap://dummy.nfo")
    pool = ConnectionPool(cli, minconn=5, maxconn=5)
    assert pool.max_connection == 5
    with pytest.raises(ValueError):
        pool.max_connection = 4
    pool.max_connection = 10
    assert pool.max_connection == 10
github noirello / bonsai / tests / test_pool.py View on Github external
def test_spawn(client):
    """ Test context manager. """
    pool = ConnectionPool(client, minconn=1, maxconn=1)
    assert pool.idle_connection == 0
    with pool.spawn() as conn:
        _ = conn.whoami()
    assert pool.idle_connection == 1
    assert pool.shared_connection == 0
github noirello / bonsai / tests / test_pool.py View on Github external
def test_close(client):
    """ Test closing the connection. """
    pool = ConnectionPool(client, minconn=1)
    pool.open()
    conn = pool.get()
    pool.close()
    assert pool.closed == True
    assert pool.idle_connection == 0
    assert pool.shared_connection == 0
    assert conn.closed == True