forked from robinhood/faust
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_simple.py
More file actions
30 lines (22 loc) · 730 Bytes
/
test_simple.py
File metadata and controls
30 lines (22 loc) · 730 Bytes
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
from typing import Mapping
from mode import setup_logging
import pytest
from .app import simple
setup_logging(loglevel='INFO')
def _build_data(i: int) -> Mapping:
return {'A': {'the': {'quick': {'brown': {'fox': i}}}}}
@pytest.mark.asyncio
async def test_simple_ask() -> None:
for i in range(100):
value = _build_data(i)
assert await simple.ask(value=value) == value
@pytest.mark.asyncio
async def test_simple_map() -> None:
values = [_build_data(i) for i in range(100)]
check = set(range(100))
replies = set()
async for reply in simple.map(values):
v = reply['A']['the']['quick']['brown']['fox']
assert v in check
replies.add(v)
assert replies == check