scrapy/tests/test_utils_python.py

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"<div>Price \xa3</div>", True),
        (b"\x02\xa3", False),
    ],
)
def test_binaryistext(value: bytes, expected: bool) -> None:
    assert binary_is_text(value) is expected


def test_get_func_args():
    def f1(a, b, c):
        pass

    def f2(a, b=None, c=None):
        pass

    def f3(a, b=None, *, c=None):
        pass

    class A:
        def __init__(self, a, b, c):
            pass

        def method(self, a, b, c):
            pass

    class Callable:
        def __call__(self, a, b, c):
            pass

    a = A(1, 2, 3)
    cal = Callable()
    partial_f1 = functools.partial(f1, None)
    partial_f2 = functools.partial(f1, b=None)
    partial_f3 = functools.partial(partial_f2, None)

    assert get_func_args(f1) == ["a", "b", "c"]
    assert get_func_args(f2) == ["a", "b", "c"]
    assert get_func_args(f3) == ["a", "b", "c"]
    assert get_func_args(A) == ["a", "b", "c"]
    assert get_func_args(a.method) == ["a", "b", "c"]
    assert get_func_args(partial_f1) == ["b", "c"]
    assert get_func_args(partial_f2) == ["a", "c"]
    assert get_func_args(partial_f3) == ["c"]
    assert get_func_args(cal) == ["a", "b", "c"]
    assert get_func_args(object) == []
    assert get_func_args(str.split, stripself=True) == ["sep", "maxsplit"]
    assert get_func_args(" ".join, stripself=True) == ["iterable"]

    if sys.version_info >= (3, 13) or platform.python_implementation() == "PyPy":
        # the correct and correctly extracted signature
        assert get_func_args(operator.itemgetter(2), stripself=True) == ["obj"]
    elif platform.python_implementation() == "CPython":
        # ["args", "kwargs"] is a correct result for the pre-3.13 incorrect function signature
        # [] is an incorrect result on even older CPython (https://github.com/python/cpython/issues/86951)
        assert get_func_args(operator.itemgetter(2), stripself=True) in [
            [],
            ["args", "kwargs"],
        ]


@pytest.mark.parametrize(
    ("value", "expected"),
    [
        ([1, None, 3, 4], [1, 3, 4]),
        ((1, None, 3, 4), (1, 3, 4)),
        (
            {"one": 1, "none": None, "three": 3, "four": 4},
            {"one": 1, "three": 3, "four": 4},
        ),
    ],
)
def test_without_none_values(
    value: Mapping[_KT, _VT] | Iterable[_KT], expected: dict[_KT, _VT] | Iterable[_KT]
) -> None:
    assert without_none_values(value) == expected