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
|
import inspect
import pytest
from aiostream.core import pipable_operator, sources_operator
from aiostream.test_utils import add_resource
from aiostream import stream, streamcontext, operator
@pytest.mark.asyncio
async def test_streamcontext(assert_cleanup):
with assert_cleanup() as loop:
xs = stream.range(3) | add_resource.pipe(1)
async with streamcontext(xs) as streamer:
it = iter(range(3))
async for item in streamer:
assert item == next(it)
assert loop.steps == [1]
with assert_cleanup() as loop:
xs = stream.range(5) | add_resource.pipe(1)
async with xs.stream() as streamer:
it = iter(range(5))
async for item in streamer:
assert item == next(it)
assert loop.steps == [1]
@pytest.mark.parametrize(
"operator_param", [operator, pipable_operator, sources_operator]
)
def test_operator_from_method(operator_param):
with pytest.raises(ValueError):
class A:
@operator_param
async def method(self, arg):
yield 1
with pytest.raises(ValueError):
class B:
@operator_param
async def method(cls, arg):
yield 1
with pytest.raises(ValueError):
class C:
@operator_param
@classmethod
async def method(cls, arg):
yield 1
@pytest.mark.asyncio
async def test_error_on_sync_iteration():
xs = stream.range(3)
# Stream raises a TypeError
with pytest.raises(TypeError):
for x in xs:
assert False
# Streamer raises a TypeError
async with xs.stream() as streamer:
with pytest.raises(TypeError):
for x in streamer:
assert False
@pytest.mark.asyncio
async def test_error_on_entering_a_stream():
xs = stream.range(3)
# Stream raises a TypeError
with pytest.raises(TypeError) as ctx:
async with xs: # type: ignore
assert False
assert "Use the `stream` method" in str(ctx.value)
def test_compatibility():
@operator
async def test1():
yield 1
with pytest.raises(AttributeError):
test1.pipe # type: ignore
match = "The `pipable` argument is deprecated."
with pytest.warns(DeprecationWarning, match=match):
@operator()
async def test2():
yield 1
with pytest.raises(AttributeError):
test2.pipe # type: ignore
with pytest.warns(DeprecationWarning, match=match):
@operator(pipable=False)
async def test3():
yield 1
with pytest.raises(AttributeError):
test3.pipe # type: ignore
with pytest.warns(DeprecationWarning, match=match):
@operator(pipable=True)
async def test4(source):
yield 1
test4.pipe # type: ignore
with pytest.warns(DeprecationWarning, match=match):
async def test5(source):
yield 1
test5_operator = operator(test5, pipable=True)
test5_operator.pipe # type: ignore
def test_pipable_operator_with_variadic_args():
with pytest.raises(ValueError):
@pipable_operator
async def test1(*args):
yield 1
def test_sources_operator_with_postional_args():
with pytest.raises(ValueError):
@sources_operator
async def test1(source):
yield 1
def test_introspection_for_operator():
# Extract original information
original = stream.range.original # type: ignore
original_doc = original.__doc__
assert original_doc is not None
assert original_doc.splitlines()[0] == "Generate a given range of numbers."
assert (
str(inspect.signature(original))
== "(*args: 'int', interval: 'float' = 0.0) -> 'AsyncIterator[int]'"
)
# Check the stream operator
assert str(stream.range) == repr(stream.range) == "aiostream.stream.create.range"
assert stream.range.__module__ == "aiostream.stream.create"
assert stream.range.__doc__ == original_doc
# Check the raw method
assert stream.range.raw.__qualname__ == "range.raw"
assert stream.range.raw.__module__ == "aiostream.stream.create"
assert stream.range.raw.__doc__ == original_doc
assert (
str(inspect.signature(stream.range.raw))
== "(*args: 'int', interval: 'float' = 0.0) -> 'AsyncIterator[int]'"
)
# Check the __call__ method
assert stream.range.__call__.__qualname__ == "range.__call__"
assert stream.range.__call__.__module__ == "aiostream.stream.create"
assert stream.range.__call__.__doc__ == original_doc
assert (
str(inspect.signature(stream.range.__call__))
== "(*args: 'int', interval: 'float' = 0.0) -> 'Stream[int]'"
)
def test_introspection_for_pipable_operator():
# Extract original information
original = stream.take.original # type: ignore
original_doc = original.__doc__
assert original_doc is not None
assert (
original_doc.splitlines()[0]
== "Forward the first ``n`` elements from an asynchronous sequence."
)
assert (
str(inspect.signature(original))
== "(source: 'AsyncIterable[T]', n: 'int') -> 'AsyncIterator[T]'"
)
# Check the stream operator
assert str(stream.take) == repr(stream.take) == "aiostream.stream.select.take"
assert stream.take.__module__ == "aiostream.stream.select"
assert stream.take.__doc__ == original_doc
# Check the raw method
assert stream.take.raw.__qualname__ == "take.raw"
assert stream.take.raw.__module__ == "aiostream.stream.select"
assert stream.take.raw.__doc__ == original_doc
assert (
str(inspect.signature(stream.take.raw))
== "(source: 'AsyncIterable[T]', n: 'int') -> 'AsyncIterator[T]'"
)
# Check the __call__ method
assert stream.take.__call__.__qualname__ == "take.__call__"
assert stream.take.__call__.__module__ == "aiostream.stream.select"
assert stream.take.__call__.__doc__ == original_doc
assert (
str(inspect.signature(stream.take.__call__))
== "(source: 'AsyncIterable[T]', n: 'int') -> 'Stream[T]'"
)
# Check the pipe method
assert stream.take.pipe.__qualname__ == "take.pipe"
assert stream.take.pipe.__module__ == "aiostream.stream.select"
assert (
stream.take.pipe.__doc__
== 'Piped version of the "take" stream operator.\n\n ' + original_doc
)
assert (
str(inspect.signature(stream.take.pipe))
== "(n: 'int') -> 'Callable[[AsyncIterable[X]], Stream[T]]'"
)
def test_introspection_for_sources_operator():
# Extract original information
original = stream.zip.original # type: ignore
original_doc = original.__doc__
assert original_doc is not None
assert (
original_doc.splitlines()[0]
== "Combine and forward the elements of several asynchronous sequences."
)
assert (
str(inspect.signature(original))
== "(*sources: 'AsyncIterable[T]', strict: 'bool' = False) -> 'AsyncIterator[tuple[T, ...]]'"
)
# Check the stream operator
assert str(stream.zip) == repr(stream.zip) == "aiostream.stream.combine.zip"
assert stream.zip.__module__ == "aiostream.stream.combine"
assert stream.zip.__doc__ == original_doc
# Check the raw method
assert stream.zip.raw.__qualname__ == "zip.raw"
assert stream.zip.raw.__module__ == "aiostream.stream.combine"
assert stream.zip.raw.__doc__ == original_doc
assert (
str(inspect.signature(stream.zip.raw))
== "(*sources: 'AsyncIterable[T]', strict: 'bool' = False) -> 'AsyncIterator[tuple[T, ...]]'"
)
# Check the __call__ method
assert stream.zip.__call__.__qualname__ == "zip.__call__"
assert stream.zip.__call__.__module__ == "aiostream.stream.combine"
assert stream.zip.__call__.__doc__ == original_doc
assert (
str(inspect.signature(stream.zip.__call__))
== "(*sources: 'AsyncIterable[T]', strict: 'bool' = False) -> 'Stream[tuple[T, ...]]'"
)
# Check the pipe method
assert stream.zip.pipe.__qualname__ == "zip.pipe"
assert stream.zip.pipe.__module__ == "aiostream.stream.combine"
assert (
stream.zip.pipe.__doc__
== 'Piped version of the "zip" stream operator.\n\n ' + original_doc
)
assert (
str(inspect.signature(stream.zip.pipe))
== "(*sources: 'AsyncIterable[T]', strict: 'bool' = False) -> 'Callable[[AsyncIterable[Any]], Stream[tuple[T, ...]]]'"
)
|