-
Notifications
You must be signed in to change notification settings - Fork 14
Expand file tree
/
Copy pathtest_cursor.py
More file actions
115 lines (91 loc) · 3.35 KB
/
test_cursor.py
File metadata and controls
115 lines (91 loc) · 3.35 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
if TYPE_CHECKING:
from psqlpy import ConnectionPool, Cursor
pytestmark = pytest.mark.anyio
async def test_cursor_fetchmany(
number_database_records: int,
test_cursor: Cursor,
) -> None:
"""Test cursor fetch with custom number of fetch."""
result = await test_cursor.fetchmany(size=number_database_records // 2)
assert len(result.result()) == number_database_records // 2
async def test_cursor_fetchone(
test_cursor: Cursor,
) -> None:
result = await test_cursor.fetchone()
assert len(result.result()) == 1
async def test_cursor_fetchall(
number_database_records: int,
test_cursor: Cursor,
) -> None:
result = await test_cursor.fetchall()
assert len(result.result()) == number_database_records
async def test_cursor_start(
psql_pool: ConnectionPool,
table_name: str,
number_database_records: int,
) -> None:
connection = await psql_pool.connection()
cursor = connection.cursor(
querystring=f"SELECT * FROM {table_name}",
)
await cursor.start()
results = await cursor.fetchall()
assert len(results.result()) == number_database_records
cursor.close()
async def test_cursor_as_async_context_manager(
psql_pool: ConnectionPool,
table_name: str,
number_database_records: int,
) -> None:
connection = await psql_pool.connection()
async with connection.cursor(
querystring=f"SELECT * FROM {table_name}",
) as cursor:
results = await cursor.fetchall()
assert len(results.result()) == number_database_records
async def test_cursor_as_async_iterator(
psql_pool: ConnectionPool,
table_name: str,
number_database_records: int,
) -> None:
connection = await psql_pool.connection()
all_results = []
async with connection.cursor(
querystring=f"SELECT * FROM {table_name}",
) as cursor:
async for results in cursor:
all_results.extend(results.result())
assert len(all_results) == number_database_records
async def test_cursor_send_underlying_connection_to_pool(
psql_pool: ConnectionPool,
table_name: str,
) -> None:
"""Test send underlying connection to the pool."""
async with psql_pool.acquire() as connection:
async with connection.transaction() as transaction:
async with transaction.cursor(
querystring=f"SELECT * FROM {table_name}",
) as cursor:
await cursor.fetchmany(10)
assert not psql_pool.status().available
assert not psql_pool.status().available
assert not psql_pool.status().available
assert psql_pool.status().available == 1
async def test_cursor_send_underlying_connection_to_pool_manually(
psql_pool: ConnectionPool,
table_name: str,
) -> None:
"""Test send underlying connection to the pool."""
async with psql_pool.acquire() as connection:
async with connection.transaction() as transaction:
cursor = transaction.cursor(querystring=f"SELECT * FROM {table_name}")
await cursor.start()
await cursor.fetchmany(10)
assert not psql_pool.status().available
cursor.close()
assert not psql_pool.status().available
assert not psql_pool.status().available
assert psql_pool.status().available == 1