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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
|
import asyncio
import sys
import time
from functools import wraps
import pytest
from conftest import get_config, network_delay
from bonsai import LDAPEntry
from bonsai.asyncio import AIOConnectionPool
from bonsai.pool import ClosedPool
import bonsai.errors
if sys.platform == "win32" and sys.version_info.minor >= 8:
# Enforce SelectorEventLoop as it's no longer default on Windows since Python 3.8.
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
def asyncio_test(func):
@wraps(func)
def wrapper(*args, **kwargs):
future = func(*args, **kwargs)
asyncio.run(future)
return wrapper
@asyncio_test
async def test_connection(client):
"""Test opening a connection."""
conn = await client.connect(True)
assert conn is not None
assert conn.closed == False
@asyncio_test
async def test_search(client):
"""Test search."""
async with client.connect(True) as conn:
res = await conn.search()
assert res is not None
@asyncio_test
async def test_add_and_delete(client, basedn):
"""Test adding and deleting an LDAP entry."""
async with client.connect(True) as conn:
entry = LDAPEntry("cn=async_test,%s" % basedn)
entry["objectclass"] = [
"top",
"inetOrgPerson",
"person",
"organizationalPerson",
]
entry["sn"] = "async_test"
try:
await conn.add(entry)
except bonsai.errors.AlreadyExists:
await conn.delete(entry.dn)
await conn.add(entry)
except:
pytest.fail("Unexpected error.")
res = await conn.search()
assert entry in res
await entry.delete()
res = await conn.search()
assert entry not in res
@asyncio_test
async def test_recursive_delete(client, basedn):
"""Test removing a subtree recursively."""
org1 = bonsai.LDAPEntry("ou=testusers,%s" % basedn)
org1.update({"objectclass": ["organizationalUnit", "top"], "ou": "testusers"})
org2 = bonsai.LDAPEntry("ou=tops,ou=testusers,%s" % basedn)
org2.update({"objectclass": ["organizationalUnit", "top"], "ou": "tops"})
entry = bonsai.LDAPEntry("cn=tester,ou=tops,ou=testusers,%s" % basedn)
entry.update(
{"objectclass": ["top", "inetorgperson"], "cn": "tester", "sn": "example"}
)
try:
with (await client.connect(True)) as conn:
await conn.add(org1)
await conn.add(org2)
await conn.add(entry)
with pytest.raises(bonsai.errors.NotAllowedOnNonleaf):
await conn.delete(org1.dn)
await conn.delete(org1.dn, recursive=True)
res = await conn.search(org1.dn, 2)
assert res == []
except bonsai.LDAPError as err:
pytest.fail("Recursive delete is failed: %s" % err)
@asyncio_test
async def test_modify_and_rename(client, basedn):
"""Test modifying and renaming LDAP entry."""
async with client.connect(True) as conn:
entry = LDAPEntry("cn=async_test,%s" % basedn)
entry["objectclass"] = [
"top",
"inetOrgPerson",
"person",
"organizationalPerson",
]
entry["sn"] = "async_test"
oldname = "cn=async_test,%s" % basedn
newname = "cn=async_test2,%s" % basedn
res = await conn.search(newname, 0)
if res:
await res[0].delete()
try:
await conn.add(entry)
except bonsai.errors.AlreadyExists:
await conn.delete(entry.dn)
await conn.add(entry)
except:
pytest.fail("Unexpected error.")
entry["sn"] = "async_test2"
await entry.modify()
await entry.rename(newname)
res = await conn.search(entry.dn, 0, attrlist=["sn"])
assert entry["sn"] == res[0]["sn"]
res = await conn.search(oldname, 0)
assert res == []
await conn.delete(entry.dn)
@asyncio_test
async def test_obj_err(client, basedn):
"""Test object class violation error."""
entry = LDAPEntry("cn=async_test,%s" % basedn)
entry["cn"] = ["async_test"]
with pytest.raises(bonsai.errors.ObjectClassViolation):
async with client.connect(True) as conn:
await conn.add(entry)
@asyncio_test
async def test_whoami(client):
"""Test whoami."""
async with client.connect(True) as conn:
cfg = get_config()
obj = await conn.whoami()
expected_res = [
"dn:%s" % cfg["SIMPLEAUTH"]["user"],
cfg["SIMPLEAUTH"]["adusername"],
]
assert obj in expected_res
@pytest.mark.timeout(18)
@asyncio_test
async def test_connection_timeout(client):
"""Test connection timeout."""
with network_delay(6.0):
with pytest.raises(asyncio.TimeoutError):
await client.connect(True, timeout=8.0)
@pytest.mark.timeout(18)
@asyncio_test
async def test_search_timeout(client):
"""Test search timeout."""
async with client.connect(True) as conn:
with network_delay(5.1):
with pytest.raises(asyncio.TimeoutError):
await conn.search(timeout=4.0)
@asyncio_test
async def test_paged_search(client, basedn):
"""Test paged results control."""
client.auto_page_acquire = False
search_dn = "ou=nerdherd,%s" % basedn
async with client.connect(True) as conn:
res = await conn.paged_search(search_dn, 1, page_size=2)
for ent in res:
assert isinstance(ent, bonsai.LDAPEntry)
page = 1 # First page is already acquired.
while True:
if len(res) > 2:
pytest.fail("The size of the page is greater than expected.")
msgid = res.acquire_next_page()
if msgid is None:
break
res = await conn.get_result(msgid)
page += 1
assert page == 3
@asyncio_test
async def test_paged_search_with_auto_acq(client, basedn):
"""Test paged search with auto page acquiring."""
search_dn = "ou=nerdherd,%s" % basedn
async with client.connect(True) as conn:
cnt = 0
result = await conn.paged_search(search_dn, 1, page_size=3)
async for item in result:
assert isinstance(item, LDAPEntry)
cnt += 1
assert cnt == 6
@asyncio_test
async def test_async_with(client):
"""Test async with context manager (with backward compatibility)."""
mgr = client.connect(True)
aexit = type(mgr).__aexit__
aenter = type(mgr).__aenter__(mgr)
conn = await aenter
try:
assert conn.closed == False
_ = await conn.whoami()
except:
if not (await aexit(mgr, *sys.exc_info())):
raise
else:
await aexit(mgr, None, None, None)
assert conn.closed
async def keep(pool, delay):
conn = await pool.get()
await asyncio.sleep(delay)
await pool.put(conn)
@asyncio_test
async def test_pool_get_put(client):
"""Test getting and putting back connection from pool."""
delay = 2
pool = AIOConnectionPool(client, minconn=1, maxconn=1)
with pytest.raises(ClosedPool):
_ = await pool.get()
await pool.open()
assert pool.closed == False
assert pool.idle_connection == 1
task1 = asyncio.ensure_future(keep(pool, delay))
task2 = asyncio.ensure_future(keep(pool, delay))
start = time.time()
await task1
await task2
assert time.time() - start >= delay * 2
# With enough connection in the pool for both tasks.
pool.max_connection = 2
task1 = asyncio.ensure_future(keep(pool, delay))
task2 = asyncio.ensure_future(keep(pool, delay))
start = time.time()
await task1
await task2
assert time.time() - start >= delay
@asyncio_test
async def test_pool_close(client):
"""Test closing the pool."""
pool = AIOConnectionPool(client, minconn=1, maxconn=1)
await pool.open()
assert pool.closed == False
assert pool.idle_connection == 1
await pool.close()
assert pool.closed == True
assert pool.idle_connection == 0
@asyncio_test
async def test_pool_pass_param(client):
"""Test passing parameter to connect."""
pool = AIOConnectionPool(client, minconn=1, maxconn=1, timeout=0)
with pytest.raises(asyncio.TimeoutError):
await pool.open()
_ = await pool.get()
@asyncio_test
async def test_pool_spawn(client):
"""Test context manager."""
pool = AIOConnectionPool(client, minconn=1, maxconn=1)
assert pool.idle_connection == 0
async with pool.spawn() as conn:
assert pool.shared_connection == 1
_ = await conn.whoami()
assert pool.idle_connection == 1
assert pool.shared_connection == 0
|
