from __future__ import annotations

from collections.abc import AsyncIterator, Iterable
from inspect import isasyncgen
from typing import TYPE_CHECKING, Any, cast
from unittest import mock

import pytest
from twisted.internet import defer

from scrapy.core.spidermw import SpiderMiddlewareManager
from scrapy.exceptions import ScrapyDeprecationWarning, _InvalidOutput
from scrapy.http import Request, Response
from scrapy.spiders import Spider
from scrapy.utils.asyncgen import collect_asyncgen
from scrapy.utils.asyncio import call_later
from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.spider import DefaultSpider
from scrapy.utils.test import get_crawler
from tests.utils.decorators import coroutine_test

if TYPE_CHECKING:
    from twisted.python.failure import Failure

    from scrapy.crawler import Crawler


class TestSpiderMiddleware:
    def setup_method(self) -> None:
        self.request = Request("http://example.com/index.html")
        self.response = Response(self.request.url, request=self.request)
        self.crawler = get_crawler(Spider, {"SPIDER_MIDDLEWARES_BASE": {}})
        self.crawler.spider = self.crawler._create_spider("foo")
        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)

    async def _scrape_response(self) -> Any:
        """Execute spider mw manager's scrape_response_async method and return the result.
        Raise exception in case of failure.
        """

        async def scrape_func(
            response: Response | Failure, request: Request
        ) -> Iterable[Any]:
            return mock.MagicMock()

        return await self.mwman.scrape_response_async(
            scrape_func, self.response, self.request
        )


class TestProcessSpiderInputInvalidOutput(TestSpiderMiddleware):
    """Invalid return value for process_spider_input method"""

    @coroutine_test
    async def test_invalid_process_spider_input(self):
        class InvalidProcessSpiderInputMiddleware:
            def process_spider_input(self, response):
                return 1

        self.mwman._add_middleware(InvalidProcessSpiderInputMiddleware())
        with pytest.raises(_InvalidOutput):
            await self._scrape_response()


class TestProcessSpiderExceptionInvalidOutput(TestSpiderMiddleware):
    """Invalid return value for process_spider_exception method"""

    @coroutine_test
    async def test_invalid_process_spider_exception(self):
        class InvalidProcessSpiderOutputExceptionMiddleware:
            def process_spider_exception(self, response, exception):
                return 1

        class RaiseExceptionProcessSpiderOutputMiddleware:
            async def process_spider_output(self, response, result):
                raise RuntimeError
                yield  # pylint: disable=unreachable

        self.mwman._add_middleware(InvalidProcessSpiderOutputExceptionMiddleware())
        self.mwman._add_middleware(RaiseExceptionProcessSpiderOutputMiddleware())
        it = await self._scrape_response()
        with pytest.raises(_InvalidOutput):
            await collect_asyncgen(it)


class TestProcessSpiderExceptionReRaise(TestSpiderMiddleware):
    """Re raise the exception by returning None"""

    @coroutine_test
    async def test_process_spider_exception_return_none(self):
        class ProcessSpiderExceptionReturnNoneMiddleware:
            def process_spider_exception(self, response, exception):
                return None

        class RaiseExceptionProcessSpiderOutputMiddleware:
            async def process_spider_output(self, response, result):
                1 / 0
                yield

        self.mwman._add_middleware(ProcessSpiderExceptionReturnNoneMiddleware())
        self.mwman._add_middleware(RaiseExceptionProcessSpiderOutputMiddleware())
        it = await self._scrape_response()
        with pytest.raises(ZeroDivisionError):
            await collect_asyncgen(it)


class TestBaseAsyncSpiderMiddleware(TestSpiderMiddleware):
    """Helpers for testing sync, async and mixed middlewares.

    Should work for process_spider_output and, when it's supported, process_start.
    """

    ITEM_TYPE: type | tuple[type, ...]
    RESULT_COUNT = 3  # to simplify checks, let everything return 3 objects

    @staticmethod
    def _construct_mw_setting(
        *mw_classes: type[Any], start_index: int | None = None
    ) -> dict[type[Any], int]:
        if start_index is None:
            start_index = 10
        return {i: c for c, i in enumerate(mw_classes, start=start_index)}

    def _callback(self) -> Any:
        yield {"foo": 1}
        yield {"foo": 2}
        yield {"foo": 3}

    async def _scrape_func(
        self, response: Response | Failure, request: Request
    ) -> Iterable[Any] | AsyncIterator[Any]:
        return cast("Iterable[Any] | AsyncIterator[Any]", self._callback())

    async def _get_middleware_result(
        self, *mw_classes: type[Any], start_index: int | None = None
    ) -> Any:
        setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
        self.crawler = get_crawler(
            Spider, {"SPIDER_MIDDLEWARES_BASE": {}, "SPIDER_MIDDLEWARES": setting}
        )
        self.crawler.spider = self.crawler._create_spider("foo")
        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
        return await self.mwman.scrape_response_async(
            self._scrape_func, self.response, self.request
        )

    async def _test_asyncgen_base(
        self,
        *mw_classes: type[Any],
        start_index: int | None = None,
    ) -> None:
        result = await self._get_middleware_result(*mw_classes, start_index=start_index)
        assert isinstance(result, AsyncIterator)
        result_list = await collect_asyncgen(result)
        assert len(result_list) == self.RESULT_COUNT
        assert isinstance(result_list[0], self.ITEM_TYPE)


class ProcessSpiderOutputSyncMiddleware:
    def process_spider_output(self, response, result):
        yield from result


class ProcessSpiderOutputAsyncGenMiddleware:
    async def process_spider_output(self, response, result):
        async for r in result:
            yield r


class ProcessSpiderOutputUniversalMiddleware:
    def process_spider_output(self, response, result):
        yield from result

    async def process_spider_output_async(self, response, result):
        async for r in result:
            yield r


class ProcessSpiderExceptionSimpleIterableMiddleware:
    def process_spider_exception(self, response, exception):
        yield {"foo": 1}
        yield {"foo": 2}
        yield {"foo": 3}


class ProcessSpiderExceptionAsyncIteratorMiddleware:
    async def process_spider_exception(self, response, exception):
        yield {"foo": 1}
        d: defer.Deferred[None] = defer.Deferred()
        call_later(0, d.callback, None)
        await maybe_deferred_to_future(d)
        yield {"foo": 2}
        yield {"foo": 3}


class TestProcessSpiderOutputSimple(TestBaseAsyncSpiderMiddleware):
    """process_spider_output tests for simple callbacks"""

    ITEM_TYPE = dict
    MW_SYNC = ProcessSpiderOutputSyncMiddleware
    MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
    MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware

    @coroutine_test
    async def test_sync(self):
        """Unsupported sync mw"""
        with pytest.raises(
            TypeError, match=r"doesn't support asynchronous spider output"
        ):
            await self._get_middleware_result(self.MW_SYNC)

    @coroutine_test
    async def test_asyncgen(self):
        """Asyncgen mw"""
        await self._test_asyncgen_base(self.MW_ASYNCGEN)

    @coroutine_test
    async def test_universal(self):
        """Universal mw"""
        await self._test_asyncgen_base(self.MW_UNIVERSAL)

    @coroutine_test
    async def test_universal_asyncgen(self):
        """Universal mw -> asyncgen mw"""
        await self._test_asyncgen_base(self.MW_ASYNCGEN, self.MW_UNIVERSAL)

    @coroutine_test
    async def test_asyncgen_universal(self):
        """Asyncgen mw -> universal mw"""
        await self._test_asyncgen_base(self.MW_UNIVERSAL, self.MW_ASYNCGEN)


class TestProcessSpiderOutputAsyncGen(TestProcessSpiderOutputSimple):
    """process_spider_output tests for async generator callbacks"""

    async def _callback(self) -> Any:
        for item in super()._callback():
            yield item


class ProcessStartSimpleMiddleware:
    async def process_start(self, start):
        async for item_or_request in start:
            yield item_or_request


class TestProcessStartSimple(TestBaseAsyncSpiderMiddleware):
    """process_start tests for simple start"""

    ITEM_TYPE = (Request, dict)
    MW_SIMPLE = ProcessStartSimpleMiddleware

    async def _get_processed_start(
        self, *mw_classes: type[Any]
    ) -> AsyncIterator[Any] | None:
        class TestSpider(Spider):
            name = "test"

            async def start(self):
                for i in range(2):
                    yield Request(f"https://example.com/{i}", dont_filter=True)
                yield {"name": "test item"}

        setting = self._construct_mw_setting(*mw_classes)
        self.crawler = get_crawler(
            TestSpider, {"SPIDER_MIDDLEWARES_BASE": {}, "SPIDER_MIDDLEWARES": setting}
        )
        self.crawler.spider = self.crawler._create_spider()
        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
        return await self.mwman.process_start()

    @coroutine_test
    async def test_simple(self):
        """Simple mw"""
        start = await self._get_processed_start(self.MW_SIMPLE)
        assert isasyncgen(start)
        start_list = await collect_asyncgen(start)
        assert len(start_list) == self.RESULT_COUNT
        assert isinstance(start_list[0], self.ITEM_TYPE)


class UniversalMiddlewareNoSync:
    async def process_spider_output_async(self, response, result):
        yield


class UniversalMiddlewareBothSync:
    def process_spider_output(self, response, result):
        yield

    def process_spider_output_async(self, response, result):
        yield


class UniversalMiddlewareBothAsync:
    async def process_spider_output(self, response, result):
        yield

    async def process_spider_output_async(self, response, result):
        yield


class TestUniversalMiddlewareManager:
    @pytest.fixture
    def crawler(self) -> Crawler:
        return get_crawler(Spider)

    @pytest.fixture
    def mwman(self, crawler: Crawler) -> SpiderMiddlewareManager:
        return SpiderMiddlewareManager.from_crawler(crawler)

    def test_simple_mw(self, mwman: SpiderMiddlewareManager) -> None:
        mw = ProcessSpiderOutputSyncMiddleware()
        with pytest.raises(
            TypeError, match=r"doesn't support asynchronous spider output"
        ):
            mwman._add_middleware(mw)

    def test_async_mw(self, mwman: SpiderMiddlewareManager) -> None:
        mw = ProcessSpiderOutputAsyncGenMiddleware()
        mwman._add_middleware(mw)
        assert (
            mwman.methods["process_spider_output"][0] == mw.process_spider_output  # pylint: disable=comparison-with-callable
        )

    def test_universal_mw(self, mwman: SpiderMiddlewareManager) -> None:
        mw = ProcessSpiderOutputUniversalMiddleware()
        mwman._add_middleware(mw)
        assert (
            mwman.methods["process_spider_output"][0] == mw.process_spider_output_async  # pylint: disable=comparison-with-callable
        )

    def test_universal_mw_no_sync(
        self, mwman: SpiderMiddlewareManager, caplog: pytest.LogCaptureFixture
    ) -> None:
        mwman._add_middleware(UniversalMiddlewareNoSync())
        assert (
            "UniversalMiddlewareNoSync has process_spider_output_async()"
            " without process_spider_output" in caplog.text
        )
        assert mwman.methods["process_spider_output"][0] is None

    def test_universal_mw_both_sync(
        self, mwman: SpiderMiddlewareManager, caplog: pytest.LogCaptureFixture
    ) -> None:
        mw = UniversalMiddlewareBothSync()
        mwman._add_middleware(mw)
        assert (
            "UniversalMiddlewareBothSync.process_spider_output_async "
            "is not an async generator function" in caplog.text
        )
        assert (
            mwman.methods["process_spider_output"][0] == mw.process_spider_output  # pylint: disable=comparison-with-callable
        )

    def test_universal_mw_both_async(
        self, mwman: SpiderMiddlewareManager, caplog: pytest.LogCaptureFixture
    ) -> None:
        mwman._add_middleware(UniversalMiddlewareBothAsync())
        assert (
            "UniversalMiddlewareBothAsync.process_spider_output "
            "is an async generator function while process_spider_output_async() exists"
            in caplog.text
        )
        assert mwman.methods["process_spider_output"][0] is None


class TestBuiltinMiddlewareSimple(TestBaseAsyncSpiderMiddleware):
    ITEM_TYPE = dict
    MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
    MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware

    async def _get_middleware_result(
        self, *mw_classes: type[Any], start_index: int | None = None
    ) -> Any:
        setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
        self.crawler = get_crawler(Spider, {"SPIDER_MIDDLEWARES": setting})
        self.crawler.spider = self.crawler._create_spider("foo")
        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
        return await self.mwman.scrape_response_async(
            self._scrape_func, self.response, self.request
        )

    @coroutine_test
    async def test_just_builtin(self):
        await self._test_asyncgen_base()

    @coroutine_test
    async def test_builtin_async(self):
        """Upgrade"""
        await self._test_asyncgen_base(self.MW_ASYNCGEN, start_index=1000)

    @coroutine_test
    async def test_builtin_universal(self):
        await self._test_asyncgen_base(self.MW_UNIVERSAL, start_index=1000)

    @coroutine_test
    async def test_async_builtin(self):
        """Upgrade"""
        await self._test_asyncgen_base(self.MW_ASYNCGEN)

    @coroutine_test
    async def test_universal_builtin(self):
        await self._test_asyncgen_base(self.MW_UNIVERSAL)


class TestBuiltinMiddlewareAsyncGen(TestBuiltinMiddlewareSimple):
    async def _callback(self) -> Any:
        for item in super()._callback():
            yield item


class TestProcessSpiderException(TestBaseAsyncSpiderMiddleware):
    ITEM_TYPE = dict
    MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
    MW_EXC_SIMPLE = ProcessSpiderExceptionSimpleIterableMiddleware
    MW_EXC_ASYNCGEN = ProcessSpiderExceptionAsyncIteratorMiddleware

    def _callback(self) -> Any:
        1 / 0

    @coroutine_test
    async def test_exc_simple(self):
        """Simple exc mw"""
        await self._test_asyncgen_base(self.MW_EXC_SIMPLE)

    @coroutine_test
    async def test_exc_async(self):
        """Async exc mw"""
        await self._test_asyncgen_base(self.MW_EXC_ASYNCGEN)

    @coroutine_test
    async def test_exc_async_async(self):
        """Async exc mw -> async output mw"""
        await self._test_asyncgen_base(self.MW_ASYNCGEN, self.MW_EXC_ASYNCGEN)

    @coroutine_test
    async def test_exc_simple_async(self):
        """Simple exc mw -> async output mw; upgrade"""
        await self._test_asyncgen_base(self.MW_ASYNCGEN, self.MW_EXC_SIMPLE)


class TestDeprecatedSpiderArg(TestSpiderMiddleware):
    @coroutine_test
    async def test_deprecated_mw_spider_arg(self):
        class DeprecatedSpiderArgMiddleware1:
            def process_spider_input(self, response, spider):
                return None

            async def process_spider_output(self, response, result, spider):
                1 / 0
                yield

        class DeprecatedSpiderArgMiddleware2:
            def process_spider_exception(self, response, exception, spider):
                return []

        with pytest.warns(
            ScrapyDeprecationWarning,
            match=r"process_spider_exception\(\) requires a spider argument",
        ):
            self.mwman._add_middleware(DeprecatedSpiderArgMiddleware2())
        with (
            pytest.warns(
                ScrapyDeprecationWarning,
                match=r"process_spider_input\(\) requires a spider argument",
            ),
            pytest.warns(
                ScrapyDeprecationWarning,
                match=r"process_spider_output\(\) requires a spider argument",
            ),
        ):
            self.mwman._add_middleware(DeprecatedSpiderArgMiddleware1())
        it = await self._scrape_response()
        await collect_asyncgen(it)

    @coroutine_test
    async def test_deprecated_mwman_spider_arg(self):
        with pytest.warns(
            ScrapyDeprecationWarning,
            match=r"Passing a spider argument to SpiderMiddlewareManager.process_start\(\)"
            r" is deprecated and the passed value is ignored",
        ):
            await self.mwman.process_start(DefaultSpider())

    @coroutine_test
    async def test_deprecated_mwman_spider_arg_no_crawler(self):
        with pytest.warns(
            ScrapyDeprecationWarning,
            match=r"MiddlewareManager.__init__\(\) was called without the crawler argument",
        ):
            mwman = SpiderMiddlewareManager()
        with pytest.warns(
            ScrapyDeprecationWarning,
            match=r"Passing a spider argument to SpiderMiddlewareManager.process_start\(\)"
            r" is deprecated, SpiderMiddlewareManager should be instantiated with a Crawler",
        ):
            await mwman.process_start(DefaultSpider())