forked from aio-libs/aiohttp
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_locks.py
55 lines (41 loc) · 1.31 KB
/
test_locks.py
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
# Tests of custom aiohttp locks implementations
import asyncio
from typing import Any, Union
import pytest
from aiohttp.locks import EventResultOrError
class TestEventResultOrError:
async def test_set_exception(self, loop: Any) -> None:
ev = EventResultOrError(loop=loop)
async def c() -> Union[int, Exception]:
try:
await ev.wait()
except Exception as e:
return e
return 1
t = loop.create_task(c())
await asyncio.sleep(0)
e = Exception()
ev.set(exc=e)
assert (await t) == e
async def test_set(self, loop: Any) -> None:
ev = EventResultOrError(loop=loop)
async def c() -> int:
await ev.wait()
return 1
t = loop.create_task(c())
await asyncio.sleep(0)
ev.set()
assert (await t) == 1
async def test_cancel_waiters(self, loop: Any) -> None:
ev = EventResultOrError(loop=loop)
async def c() -> None:
await ev.wait()
t1 = loop.create_task(c())
t2 = loop.create_task(c())
await asyncio.sleep(0)
ev.cancel()
ev.set()
with pytest.raises(asyncio.CancelledError):
await t1
with pytest.raises(asyncio.CancelledError):
await t2