from __future__ import annotations import functools import operator import platform import sys from typing import TYPE_CHECKING, TypeVar import pytest from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen from scrapy.utils.defer import aiter_errback from scrapy.utils.python import ( MutableAsyncChain, binary_is_text, get_func_args, memoizemethod_noargs, to_bytes, to_unicode, without_none_values, ) from tests.utils.decorators import coroutine_test if TYPE_CHECKING: from collections.abc import Iterable, Mapping _KT = TypeVar("_KT") _VT = TypeVar("_VT") class TestMutableAsyncChain: @staticmethod async def g1(): for i in range(3): yield i @staticmethod async def g2(): return yield @staticmethod async def g3(): for i in range(7, 10): yield i @staticmethod async def g4(): for i in range(3, 5): yield i 1 / 0 for i in range(5, 7): yield i @coroutine_test async def test_mutableasyncchain(self): m = MutableAsyncChain(self.g1(), as_async_generator(range(3, 7))) m.extend(self.g2()) m.extend(self.g3()) assert await m.__anext__() == 0 results = await collect_asyncgen(m) assert results == list(range(1, 10)) @coroutine_test async def test_mutableasyncchain_exc(self): m = MutableAsyncChain(self.g1()) m.extend(self.g4()) m.extend(self.g3()) results = await collect_asyncgen(aiter_errback(m, lambda _: None)) assert results == list(range(5)) class TestToUnicode: def test_converting_an_utf8_encoded_string_to_unicode(self): assert to_unicode(b"lel\xc3\xb1e") == "lel\xf1e" def test_converting_a_latin_1_encoded_string_to_unicode(self): assert to_unicode(b"lel\xf1e", "latin-1") == "lel\xf1e" def test_converting_a_unicode_to_unicode_should_return_the_same_object(self): assert to_unicode("\xf1e\xf1e\xf1e") == "\xf1e\xf1e\xf1e" def test_converting_a_strange_object_should_raise_type_error(self): with pytest.raises(TypeError): to_unicode(423) def test_errors_argument(self): assert to_unicode(b"a\xedb", "utf-8", errors="replace") == "a\ufffdb" class TestToBytes: def test_converting_a_unicode_object_to_an_utf_8_encoded_string(self): assert to_bytes("\xa3 49") == b"\xc2\xa3 49" def test_converting_a_unicode_object_to_a_latin_1_encoded_string(self): assert to_bytes("\xa3 49", "latin-1") == b"\xa3 49" def test_converting_a_regular_bytes_to_bytes_should_return_the_same_object(self): assert to_bytes(b"lel\xf1e") == b"lel\xf1e" def test_converting_a_strange_object_should_raise_type_error(self): with pytest.raises(TypeError): to_bytes(pytest) def test_errors_argument(self): assert to_bytes("a\ufffdb", "latin-1", errors="replace") == b"a?b" def test_memoizemethod_noargs(): class A: @memoizemethod_noargs def cached(self): return object() def noncached(self): return object() a = A() one = a.cached() two = a.cached() three = a.noncached() assert one is two assert one is not three @pytest.mark.parametrize( ("value", "expected"), [ (b"hello", True), ("hello".encode("utf-16"), True), (b"