import asyncio
import logging
import platform
import re
import signal
import subprocess
import sys
import warnings
from abc import ABC, abstractmethod
from collections.abc import Generator
from pathlib import Path
from typing import Any

import pytest
from packaging.version import parse as parse_version
from pexpect.popen_spawn import PopenSpawn
from twisted.internet.defer import Deferred, inlineCallbacks
from w3lib import __version__ as w3lib_version
from zope.interface.exceptions import MultipleInvalid

import scrapy
from scrapy import Spider
from scrapy.crawler import (
    AsyncCrawlerProcess,
    AsyncCrawlerRunner,
    Crawler,
    CrawlerProcess,
    CrawlerRunner,
)
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.extensions.throttle import AutoThrottle
from scrapy.settings import Settings, default_settings
from scrapy.utils.asyncio import call_later
from scrapy.utils.defer import (
    deferred_f_from_coro_f,
    deferred_from_coro,
    maybe_deferred_to_future,
)
from scrapy.utils.log import (
    _uninstall_scrapy_root_handler,
    configure_logging,
    get_scrapy_root_handler,
)
from scrapy.utils.spider import DefaultSpider
from scrapy.utils.test import get_crawler, get_reactor_settings
from tests.mockserver.http import MockServer
from tests.utils import get_script_run_env

BASE_SETTINGS: dict[str, Any] = {}


def get_raw_crawler(spidercls=None, settings_dict=None):
    """get_crawler alternative that only calls the __init__ method of the
    crawler."""
    settings = Settings()
    settings.setdict(get_reactor_settings())
    settings.setdict(settings_dict or {})
    return Crawler(spidercls or DefaultSpider, settings)


class TestBaseCrawler:
    def assertOptionIsDefault(self, settings, key):
        assert isinstance(settings, Settings)
        assert settings[key] == getattr(default_settings, key)


class TestCrawler(TestBaseCrawler):
    def test_populate_spidercls_settings(self):
        spider_settings = {"TEST1": "spider", "TEST2": "spider"}
        project_settings = {
            **BASE_SETTINGS,
            "TEST1": "project",
            "TEST3": "project",
            **get_reactor_settings(),
        }

        class CustomSettingsSpider(DefaultSpider):
            custom_settings = spider_settings

        settings = Settings()
        settings.setdict(project_settings, priority="project")
        crawler = Crawler(CustomSettingsSpider, settings)
        crawler._apply_settings()

        assert crawler.settings.get("TEST1") == "spider"
        assert crawler.settings.get("TEST2") == "spider"
        assert crawler.settings.get("TEST3") == "project"

        assert not settings.frozen
        assert crawler.settings.frozen

    def test_crawler_accepts_dict(self):
        crawler = get_crawler(DefaultSpider, {"foo": "bar"})
        assert crawler.settings["foo"] == "bar"
        self.assertOptionIsDefault(crawler.settings, "RETRY_ENABLED")

    def test_crawler_accepts_None(self):
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", ScrapyDeprecationWarning)
            crawler = Crawler(DefaultSpider)
        self.assertOptionIsDefault(crawler.settings, "RETRY_ENABLED")

    def test_crawler_rejects_spider_objects(self):
        with pytest.raises(ValueError, match="spidercls argument must be a class"):
            Crawler(DefaultSpider())

    @inlineCallbacks
    def test_crawler_crawl_twice_seq_unsupported(self):
        crawler = get_raw_crawler(NoRequestsSpider, BASE_SETTINGS)
        yield crawler.crawl()
        with pytest.raises(RuntimeError, match="more than once on the same instance"):
            yield crawler.crawl()

    @deferred_f_from_coro_f
    async def test_crawler_crawl_async_twice_seq_unsupported(self):
        crawler = get_raw_crawler(NoRequestsSpider, BASE_SETTINGS)
        await crawler.crawl_async()
        with pytest.raises(RuntimeError, match="more than once on the same instance"):
            await crawler.crawl_async()

    @inlineCallbacks
    def test_crawler_crawl_twice_parallel_unsupported(self):
        crawler = get_raw_crawler(NoRequestsSpider, BASE_SETTINGS)
        d1 = crawler.crawl()
        d2 = crawler.crawl()
        yield d1
        with pytest.raises(RuntimeError, match="Crawling already taking place"):
            yield d2

    @pytest.mark.only_asyncio
    @deferred_f_from_coro_f
    async def test_crawler_crawl_async_twice_parallel_unsupported(self):
        crawler = get_raw_crawler(NoRequestsSpider, BASE_SETTINGS)
        t1 = asyncio.create_task(crawler.crawl_async())
        t2 = asyncio.create_task(crawler.crawl_async())
        await t1
        with pytest.raises(RuntimeError, match="Crawling already taking place"):
            await t2

    def test_get_addon(self):
        class ParentAddon:
            pass

        class TrackingAddon(ParentAddon):
            instances = []

            def __init__(self):
                TrackingAddon.instances.append(self)

            def update_settings(self, settings):
                pass

        settings = {
            **BASE_SETTINGS,
            "ADDONS": {
                TrackingAddon: 0,
            },
        }
        crawler = get_crawler(settings_dict=settings)
        assert len(TrackingAddon.instances) == 1
        expected = TrackingAddon.instances[-1]

        addon = crawler.get_addon(TrackingAddon)
        assert addon == expected

        addon = crawler.get_addon(DefaultSpider)
        assert addon is None

        addon = crawler.get_addon(ParentAddon)
        assert addon == expected

        class ChildAddon(TrackingAddon):
            pass

        addon = crawler.get_addon(ChildAddon)
        assert addon is None

    @inlineCallbacks
    def test_get_downloader_middleware(self):
        class ParentDownloaderMiddleware:
            pass

        class TrackingDownloaderMiddleware(ParentDownloaderMiddleware):
            instances = []

            def __init__(self):
                TrackingDownloaderMiddleware.instances.append(self)

        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                return cls(crawler=crawler)

            def __init__(self, crawler, **kwargs: Any):
                super().__init__(**kwargs)
                self.crawler = crawler

            async def start(self):
                MySpider.result = crawler.get_downloader_middleware(MySpider.cls)
                return
                yield

        settings = {
            **BASE_SETTINGS,
            "DOWNLOADER_MIDDLEWARES": {
                TrackingDownloaderMiddleware: 0,
            },
        }

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = TrackingDownloaderMiddleware
        yield crawler.crawl()
        assert len(TrackingDownloaderMiddleware.instances) == 1
        assert MySpider.result == TrackingDownloaderMiddleware.instances[-1]

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = DefaultSpider
        yield crawler.crawl()
        assert MySpider.result is None

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ParentDownloaderMiddleware
        yield crawler.crawl()
        assert MySpider.result == TrackingDownloaderMiddleware.instances[-1]

        class ChildDownloaderMiddleware(TrackingDownloaderMiddleware):
            pass

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ChildDownloaderMiddleware
        yield crawler.crawl()
        assert MySpider.result is None

    def test_get_downloader_middleware_not_crawling(self):
        crawler = get_raw_crawler(settings_dict=BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            crawler.get_downloader_middleware(DefaultSpider)

    @inlineCallbacks
    def test_get_downloader_middleware_no_engine(self):
        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                try:
                    crawler.get_downloader_middleware(DefaultSpider)
                except Exception as e:
                    MySpider.result = e
                    raise

        crawler = get_raw_crawler(MySpider, BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            yield crawler.crawl()

    @inlineCallbacks
    def test_get_extension(self):
        class ParentExtension:
            pass

        class TrackingExtension(ParentExtension):
            instances = []

            def __init__(self):
                TrackingExtension.instances.append(self)

        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                return cls(crawler=crawler)

            def __init__(self, crawler, **kwargs: Any):
                super().__init__(**kwargs)
                self.crawler = crawler

            async def start(self):
                MySpider.result = crawler.get_extension(MySpider.cls)
                return
                yield

        settings = {
            **BASE_SETTINGS,
            "EXTENSIONS": {
                TrackingExtension: 0,
            },
        }

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = TrackingExtension
        yield crawler.crawl()
        assert len(TrackingExtension.instances) == 1
        assert MySpider.result == TrackingExtension.instances[-1]

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = DefaultSpider
        yield crawler.crawl()
        assert MySpider.result is None

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ParentExtension
        yield crawler.crawl()
        assert MySpider.result == TrackingExtension.instances[-1]

        class ChildExtension(TrackingExtension):
            pass

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ChildExtension
        yield crawler.crawl()
        assert MySpider.result is None

    def test_get_extension_not_crawling(self):
        crawler = get_raw_crawler(settings_dict=BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            crawler.get_extension(DefaultSpider)

    @inlineCallbacks
    def test_get_extension_no_engine(self):
        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                try:
                    crawler.get_extension(DefaultSpider)
                except Exception as e:
                    MySpider.result = e
                    raise

        crawler = get_raw_crawler(MySpider, BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            yield crawler.crawl()

    @inlineCallbacks
    def test_get_item_pipeline(self):
        class ParentItemPipeline:
            pass

        class TrackingItemPipeline(ParentItemPipeline):
            instances = []

            def __init__(self):
                TrackingItemPipeline.instances.append(self)

        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                return cls(crawler=crawler)

            def __init__(self, crawler, **kwargs: Any):
                super().__init__(**kwargs)
                self.crawler = crawler

            async def start(self):
                MySpider.result = crawler.get_item_pipeline(MySpider.cls)
                return
                yield

        settings = {
            **BASE_SETTINGS,
            "ITEM_PIPELINES": {
                TrackingItemPipeline: 0,
            },
        }

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = TrackingItemPipeline
        yield crawler.crawl()
        assert len(TrackingItemPipeline.instances) == 1
        assert MySpider.result == TrackingItemPipeline.instances[-1]

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = DefaultSpider
        yield crawler.crawl()
        assert MySpider.result is None

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ParentItemPipeline
        yield crawler.crawl()
        assert MySpider.result == TrackingItemPipeline.instances[-1]

        class ChildItemPipeline(TrackingItemPipeline):
            pass

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ChildItemPipeline
        yield crawler.crawl()
        assert MySpider.result is None

    def test_get_item_pipeline_not_crawling(self):
        crawler = get_raw_crawler(settings_dict=BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            crawler.get_item_pipeline(DefaultSpider)

    @inlineCallbacks
    def test_get_item_pipeline_no_engine(self):
        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                try:
                    crawler.get_item_pipeline(DefaultSpider)
                except Exception as e:
                    MySpider.result = e
                    raise

        crawler = get_raw_crawler(MySpider, BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            yield crawler.crawl()

    @inlineCallbacks
    def test_get_spider_middleware(self):
        class ParentSpiderMiddleware:
            pass

        class TrackingSpiderMiddleware(ParentSpiderMiddleware):
            instances = []

            def __init__(self):
                TrackingSpiderMiddleware.instances.append(self)

        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                return cls(crawler=crawler)

            def __init__(self, crawler, **kwargs: Any):
                super().__init__(**kwargs)
                self.crawler = crawler

            async def start(self):
                MySpider.result = crawler.get_spider_middleware(MySpider.cls)
                return
                yield

        settings = {
            **BASE_SETTINGS,
            "SPIDER_MIDDLEWARES": {
                TrackingSpiderMiddleware: 0,
            },
        }

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = TrackingSpiderMiddleware
        yield crawler.crawl()
        assert len(TrackingSpiderMiddleware.instances) == 1
        assert MySpider.result == TrackingSpiderMiddleware.instances[-1]

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = DefaultSpider
        yield crawler.crawl()
        assert MySpider.result is None

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ParentSpiderMiddleware
        yield crawler.crawl()
        assert MySpider.result == TrackingSpiderMiddleware.instances[-1]

        class ChildSpiderMiddleware(TrackingSpiderMiddleware):
            pass

        crawler = get_raw_crawler(MySpider, settings)
        MySpider.cls = ChildSpiderMiddleware
        yield crawler.crawl()
        assert MySpider.result is None

    def test_get_spider_middleware_not_crawling(self):
        crawler = get_raw_crawler(settings_dict=BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            crawler.get_spider_middleware(DefaultSpider)

    @inlineCallbacks
    def test_get_spider_middleware_no_engine(self):
        class MySpider(Spider):
            name = "myspider"

            @classmethod
            def from_crawler(cls, crawler):
                try:
                    crawler.get_spider_middleware(DefaultSpider)
                except Exception as e:
                    MySpider.result = e
                    raise

        crawler = get_raw_crawler(MySpider, BASE_SETTINGS)
        with pytest.raises(RuntimeError):
            yield crawler.crawl()


class TestSpiderSettings:
    def test_spider_custom_settings(self):
        class MySpider(scrapy.Spider):
            name = "spider"
            custom_settings = {"AUTOTHROTTLE_ENABLED": True}

        crawler = get_crawler(MySpider)
        enabled_exts = [e.__class__ for e in crawler.extensions.middlewares]
        assert AutoThrottle in enabled_exts


class TestCrawlerLogging:
    def test_no_root_handler_installed(self):
        handler = get_scrapy_root_handler()
        if handler is not None:
            logging.root.removeHandler(handler)

        class MySpider(scrapy.Spider):
            name = "spider"

        get_crawler(MySpider)
        assert get_scrapy_root_handler() is None

    @deferred_f_from_coro_f
    async def test_spider_custom_settings_log_level(self, tmp_path):
        log_file = Path(tmp_path, "log.txt")
        log_file.write_text("previous message\n", encoding="utf-8")

        info_count = None

        class MySpider(scrapy.Spider):
            name = "spider"
            custom_settings = {
                "LOG_LEVEL": "INFO",
                "LOG_FILE": str(log_file),
            }

            async def start(self):
                info_count_start = crawler.stats.get_value("log_count/INFO")
                logging.debug("debug message")  # noqa: LOG015
                logging.info("info message")  # noqa: LOG015
                logging.warning("warning message")  # noqa: LOG015
                logging.error("error message")  # noqa: LOG015
                nonlocal info_count
                info_count = (
                    crawler.stats.get_value("log_count/INFO") - info_count_start
                )
                return
                yield

        try:
            configure_logging()
            assert get_scrapy_root_handler().level == logging.DEBUG
            crawler = get_crawler(MySpider)
            assert get_scrapy_root_handler().level == logging.INFO
            await crawler.crawl_async()
        finally:
            _uninstall_scrapy_root_handler()

        logged = log_file.read_text(encoding="utf-8")

        assert "previous message" in logged
        assert "debug message" not in logged
        assert "info message" in logged
        assert "warning message" in logged
        assert "error message" in logged
        assert crawler.stats.get_value("log_count/ERROR") == 1
        assert crawler.stats.get_value("log_count/WARNING") == 1
        assert info_count == 1
        assert crawler.stats.get_value("log_count/DEBUG", 0) == 0

    def test_spider_custom_settings_log_append(self, tmp_path):
        log_file = Path(tmp_path, "log.txt")
        log_file.write_text("previous message\n", encoding="utf-8")

        class MySpider(scrapy.Spider):
            name = "spider"
            custom_settings = {
                "LOG_FILE": str(log_file),
                "LOG_FILE_APPEND": False,
            }

        try:
            configure_logging()
            get_crawler(MySpider)
            logging.debug("debug message")  # noqa: LOG015
        finally:
            _uninstall_scrapy_root_handler()

        logged = log_file.read_text(encoding="utf-8")

        assert "previous message" not in logged
        assert "debug message" in logged


class SpiderLoaderWithWrongInterface:
    def unneeded_method(self):
        pass


class TestCrawlerRunner(TestBaseCrawler):
    def test_spider_manager_verify_interface(self):
        settings = Settings(
            {
                "SPIDER_LOADER_CLASS": SpiderLoaderWithWrongInterface,
            }
        )
        with pytest.raises(MultipleInvalid):
            CrawlerRunner(settings)

    def test_crawler_runner_accepts_dict(self):
        runner = CrawlerRunner({"foo": "bar"})
        assert runner.settings["foo"] == "bar"
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")

    def test_crawler_runner_accepts_None(self):
        runner = CrawlerRunner()
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")


class TestAsyncCrawlerRunner(TestBaseCrawler):
    def test_spider_manager_verify_interface(self):
        settings = Settings(
            {
                "SPIDER_LOADER_CLASS": SpiderLoaderWithWrongInterface,
            }
        )
        with pytest.raises(MultipleInvalid):
            AsyncCrawlerRunner(settings)

    def test_crawler_runner_accepts_dict(self):
        runner = AsyncCrawlerRunner({"foo": "bar"})
        assert runner.settings["foo"] == "bar"
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")

    def test_crawler_runner_accepts_None(self):
        runner = AsyncCrawlerRunner()
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")


class TestCrawlerProcess(TestBaseCrawler):
    def test_crawler_process_accepts_dict(self):
        runner = CrawlerProcess({"foo": "bar"}, install_root_handler=False)
        assert runner.settings["foo"] == "bar"
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")

    def test_crawler_process_accepts_None(self):
        runner = CrawlerProcess(install_root_handler=False)
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")


@pytest.mark.only_asyncio
class TestAsyncCrawlerProcess(TestBaseCrawler):
    def test_crawler_process_accepts_dict(self):
        runner = AsyncCrawlerProcess({"foo": "bar"}, install_root_handler=False)
        assert runner.settings["foo"] == "bar"
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")

    def test_crawler_process_accepts_None(self):
        runner = AsyncCrawlerProcess(install_root_handler=False)
        self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED")


class ExceptionSpider(scrapy.Spider):
    name = "exception"

    @classmethod
    def from_crawler(cls, crawler, *args, **kwargs):
        raise ValueError("Exception in from_crawler method")


class NoRequestsSpider(scrapy.Spider):
    name = "no_request"

    async def start(self):
        return
        yield


class TestCrawlerRunnerHasSpider:
    @staticmethod
    def _runner():
        return CrawlerRunner(get_reactor_settings())

    @staticmethod
    def _crawl(runner, spider):
        return runner.crawl(spider)

    @inlineCallbacks
    def test_crawler_runner_bootstrap_successful(self):
        runner = self._runner()
        yield self._crawl(runner, NoRequestsSpider)
        assert not runner.bootstrap_failed

    @inlineCallbacks
    def test_crawler_runner_bootstrap_successful_for_several(self):
        runner = self._runner()
        yield self._crawl(runner, NoRequestsSpider)
        yield self._crawl(runner, NoRequestsSpider)
        assert not runner.bootstrap_failed

    @inlineCallbacks
    def test_crawler_runner_bootstrap_failed(self):
        runner = self._runner()

        try:
            yield self._crawl(runner, ExceptionSpider)
        except ValueError:
            pass
        else:
            pytest.fail("Exception should be raised from spider")

        assert runner.bootstrap_failed

    @inlineCallbacks
    def test_crawler_runner_bootstrap_failed_for_several(self):
        runner = self._runner()

        try:
            yield self._crawl(runner, ExceptionSpider)
        except ValueError:
            pass
        else:
            pytest.fail("Exception should be raised from spider")

        yield self._crawl(runner, NoRequestsSpider)

        assert runner.bootstrap_failed

    @inlineCallbacks
    def test_crawler_runner_asyncio_enabled_true(
        self, reactor_pytest: str
    ) -> Generator[Deferred[Any], Any, None]:
        if reactor_pytest != "asyncio":
            runner = CrawlerRunner(
                settings={
                    "TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor",
                }
            )
            with pytest.raises(
                Exception,
                match=r"The installed reactor \(.*?\) does not match the requested one \(.*?\)",
            ):
                yield self._crawl(runner, NoRequestsSpider)
        else:
            CrawlerRunner(
                settings={
                    "TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor",
                }
            )


@pytest.mark.only_asyncio
class TestAsyncCrawlerRunnerHasSpider(TestCrawlerRunnerHasSpider):
    @staticmethod
    def _runner():
        return AsyncCrawlerRunner(get_reactor_settings())

    @staticmethod
    def _crawl(runner, spider):
        return deferred_from_coro(runner.crawl(spider))

    def test_crawler_runner_asyncio_enabled_true(self):
        pytest.skip("This test is only for CrawlerRunner")


class ScriptRunnerMixin(ABC):
    @property
    @abstractmethod
    def script_dir(self) -> Path:
        raise NotImplementedError

    @staticmethod
    def get_script_dir(name: str) -> Path:
        return Path(__file__).parent.resolve() / name

    def get_script_args(self, script_name: str, *script_args: str) -> list[str]:
        script_path = self.script_dir / script_name
        return [sys.executable, str(script_path), *script_args]

    def run_script(self, script_name: str, *script_args: str) -> str:
        args = self.get_script_args(script_name, *script_args)
        p = subprocess.Popen(
            args,
            env=get_script_run_env(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        _, stderr = p.communicate()
        return stderr.decode("utf-8")


class TestCrawlerProcessSubprocessBase(ScriptRunnerMixin):
    """Common tests between CrawlerProcess and AsyncCrawlerProcess,
    with the same file names and expectations.
    """

    def test_simple(self):
        log = self.run_script("simple.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )

    def test_multi(self):
        log = self.run_script("multi.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "ReactorAlreadyInstalledError" not in log

    def test_reactor_default(self):
        log = self.run_script("reactor_default.py")
        assert "Spider closed (finished)" not in log
        assert (
            "does not match the requested one "
            "(twisted.internet.asyncioreactor.AsyncioSelectorReactor)"
        ) in log

    def test_asyncio_enabled_no_reactor(self):
        log = self.run_script("asyncio_enabled_no_reactor.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "RuntimeError" not in log

    def test_asyncio_enabled_reactor(self):
        log = self.run_script("asyncio_enabled_reactor.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "RuntimeError" not in log

    @pytest.mark.skipif(
        parse_version(w3lib_version) >= parse_version("2.0.0"),
        reason="w3lib 2.0.0 and later do not allow invalid domains.",
    )
    def test_ipv6_default_name_resolver(self):
        log = self.run_script("default_name_resolver.py")
        assert "Spider closed (finished)" in log
        assert (
            "'downloader/exception_type_count/twisted.internet.error.DNSLookupError': 1,"
            in log
        )
        assert (
            "twisted.internet.error.DNSLookupError: DNS lookup failed: no results for hostname lookup: ::1."
            in log
        )

    def test_caching_hostname_resolver_ipv6(self):
        log = self.run_script("caching_hostname_resolver_ipv6.py")
        assert "Spider closed (finished)" in log
        assert "twisted.internet.error.DNSLookupError" not in log

    def test_caching_hostname_resolver_finite_execution(
        self, mockserver: MockServer
    ) -> None:
        log = self.run_script("caching_hostname_resolver.py", mockserver.url("/"))
        assert "Spider closed (finished)" in log
        assert "ERROR: Error downloading" not in log
        assert "TimeoutError" not in log
        assert "twisted.internet.error.DNSLookupError" not in log

    def test_twisted_reactor_asyncio(self):
        log = self.run_script("twisted_reactor_asyncio.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )

    def test_twisted_reactor_asyncio_custom_settings(self):
        log = self.run_script("twisted_reactor_custom_settings.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )

    def test_twisted_reactor_asyncio_custom_settings_same(self):
        log = self.run_script("twisted_reactor_custom_settings_same.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )

    @pytest.mark.requires_uvloop
    def test_custom_loop_asyncio(self):
        log = self.run_script("asyncio_custom_loop.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "Using asyncio event loop: uvloop.Loop" in log

    @pytest.mark.requires_uvloop
    def test_custom_loop_asyncio_deferred_signal(self):
        log = self.run_script("asyncio_deferred_signal.py", "uvloop.Loop")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "Using asyncio event loop: uvloop.Loop" in log
        assert "async pipeline opened!" in log

    @pytest.mark.requires_uvloop
    def test_asyncio_enabled_reactor_same_loop(self):
        log = self.run_script("asyncio_enabled_reactor_same_loop.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "Using asyncio event loop: uvloop.Loop" in log

    @pytest.mark.requires_uvloop
    def test_asyncio_enabled_reactor_different_loop(self):
        log = self.run_script("asyncio_enabled_reactor_different_loop.py")
        assert "Spider closed (finished)" not in log
        assert (
            "does not match the one specified in the ASYNCIO_EVENT_LOOP "
            "setting (uvloop.Loop)"
        ) in log

    def test_default_loop_asyncio_deferred_signal(self):
        log = self.run_script("asyncio_deferred_signal.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "Using asyncio event loop: uvloop.Loop" not in log
        assert "async pipeline opened!" in log

    def test_args_change_settings(self):
        log = self.run_script("args_settings.py")
        assert "Spider closed (finished)" in log
        assert "The value of FOO is 42" in log

    def test_shutdown_graceful(self):
        sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK
        args = self.get_script_args("sleeping.py", "3")
        p = PopenSpawn(args, timeout=5)
        p.expect_exact("Spider opened")
        p.expect_exact("Crawled (200)")
        p.kill(sig)
        p.expect_exact("shutting down gracefully")
        p.expect_exact("Spider closed (shutdown)")
        p.wait()

    @inlineCallbacks
    def test_shutdown_forced(self):
        sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK
        args = self.get_script_args("sleeping.py", "10")
        p = PopenSpawn(args, timeout=5)
        p.expect_exact("Spider opened")
        p.expect_exact("Crawled (200)")
        p.kill(sig)
        p.expect_exact("shutting down gracefully")
        # sending the second signal too fast often causes problems
        d = Deferred()
        call_later(0.01, d.callback, None)
        yield d
        p.kill(sig)
        p.expect_exact("forcing unclean shutdown")
        p.wait()


class TestCrawlerProcessSubprocess(TestCrawlerProcessSubprocessBase):
    @property
    def script_dir(self) -> Path:
        return self.get_script_dir("CrawlerProcess")

    def test_reactor_default_twisted_reactor_select(self):
        log = self.run_script("reactor_default_twisted_reactor_select.py")
        if platform.system() in ["Windows", "Darwin"]:
            # The goal of this test function is to test that, when a reactor is
            # installed (the default one here) and a different reactor is
            # configured (select here), an error raises.
            #
            # In Windows the default reactor is the select reactor, so that
            # error does not raise.
            #
            # If that ever becomes the case on more platforms (i.e. if Linux
            # also starts using the select reactor by default in a future
            # version of Twisted), then we will need to rethink this test.
            assert "Spider closed (finished)" in log
        else:
            assert "Spider closed (finished)" not in log
            assert (
                "does not match the requested one "
                "(twisted.internet.selectreactor.SelectReactor)"
            ) in log

    def test_reactor_select(self):
        log = self.run_script("reactor_select.py")
        assert "Spider closed (finished)" not in log
        assert (
            "does not match the requested one "
            "(twisted.internet.asyncioreactor.AsyncioSelectorReactor)"
        ) in log

    def test_reactor_select_twisted_reactor_select(self):
        log = self.run_script("reactor_select_twisted_reactor_select.py")
        assert "Spider closed (finished)" in log
        assert "ReactorAlreadyInstalledError" not in log

    def test_reactor_select_subclass_twisted_reactor_select(self):
        log = self.run_script("reactor_select_subclass_twisted_reactor_select.py")
        assert "Spider closed (finished)" not in log
        assert (
            "does not match the requested one "
            "(twisted.internet.selectreactor.SelectReactor)"
        ) in log

    def test_twisted_reactor_select(self):
        log = self.run_script("twisted_reactor_select.py")
        assert "Spider closed (finished)" in log
        assert "Using reactor: twisted.internet.selectreactor.SelectReactor" in log

    @pytest.mark.skipif(
        platform.system() == "Windows", reason="PollReactor is not supported on Windows"
    )
    def test_twisted_reactor_poll(self):
        log = self.run_script("twisted_reactor_poll.py")
        assert "Spider closed (finished)" in log
        assert "Using reactor: twisted.internet.pollreactor.PollReactor" in log

    def test_twisted_reactor_asyncio_custom_settings_conflict(self):
        log = self.run_script("twisted_reactor_custom_settings_conflict.py")
        assert "Using reactor: twisted.internet.selectreactor.SelectReactor" in log
        assert (
            "(twisted.internet.selectreactor.SelectReactor) does not match the requested one"
            in log
        )


class TestAsyncCrawlerProcessSubprocess(TestCrawlerProcessSubprocessBase):
    @property
    def script_dir(self) -> Path:
        return self.get_script_dir("AsyncCrawlerProcess")

    def test_twisted_reactor_custom_settings_select(self):
        log = self.run_script("twisted_reactor_custom_settings_select.py")
        assert "Spider closed (finished)" not in log
        assert (
            "(twisted.internet.asyncioreactor.AsyncioSelectorReactor) "
            "does not match the requested one "
            "(twisted.internet.selectreactor.SelectReactor)"
        ) in log

    @pytest.mark.requires_uvloop
    def test_asyncio_enabled_reactor_same_loop(self):
        log = self.run_script("asyncio_custom_loop_custom_settings_same.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "Using asyncio event loop: uvloop.Loop" in log

    @pytest.mark.requires_uvloop
    def test_asyncio_enabled_reactor_different_loop(self):
        log = self.run_script("asyncio_custom_loop_custom_settings_different.py")
        assert "Spider closed (finished)" not in log
        assert (
            "does not match the one specified in the ASYNCIO_EVENT_LOOP "
            "setting (uvloop.Loop)"
        ) in log


class TestCrawlerRunnerSubprocessBase(ScriptRunnerMixin):
    """Common tests between CrawlerRunner and AsyncCrawlerRunner,
    with the same file names and expectations.
    """

    def test_simple(self):
        log = self.run_script("simple.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )

    def test_multi_parallel(self):
        log = self.run_script("multi_parallel.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert re.search(
            r"Spider opened.+Spider opened.+Closing spider.+Closing spider",
            log,
            re.DOTALL,
        )

    def test_multi_seq(self):
        log = self.run_script("multi_seq.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert re.search(
            r"Spider opened.+Closing spider.+Spider opened.+Closing spider",
            log,
            re.DOTALL,
        )

    @pytest.mark.requires_uvloop
    def test_custom_loop_same(self):
        log = self.run_script("custom_loop_same.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "Using asyncio event loop: uvloop.Loop" in log

    @pytest.mark.requires_uvloop
    def test_custom_loop_different(self):
        log = self.run_script("custom_loop_different.py")
        assert "Spider closed (finished)" not in log
        assert (
            "does not match the one specified in the ASYNCIO_EVENT_LOOP "
            "setting (uvloop.Loop)"
        ) in log


class TestCrawlerRunnerSubprocess(TestCrawlerRunnerSubprocessBase):
    @property
    def script_dir(self) -> Path:
        return self.get_script_dir("CrawlerRunner")

    def test_explicit_default_reactor(self):
        log = self.run_script("explicit_default_reactor.py")
        assert "Spider closed (finished)" in log
        assert (
            "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            not in log
        )

    def test_response_ip_address(self):
        log = self.run_script("ip_address.py")
        assert "INFO: Spider closed (finished)" in log
        assert "INFO: Host: not.a.real.domain" in log
        assert "INFO: Type: <class 'ipaddress.IPv4Address'>" in log
        assert "INFO: IP address: 127.0.0.1" in log

    def test_change_default_reactor(self):
        log = self.run_script("change_reactor.py")
        assert (
            "DEBUG: Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor"
            in log
        )
        assert "DEBUG: Using asyncio event loop" in log


class TestAsyncCrawlerRunnerSubprocess(TestCrawlerRunnerSubprocessBase):
    @property
    def script_dir(self) -> Path:
        return self.get_script_dir("AsyncCrawlerRunner")

    def test_simple_default_reactor(self):
        log = self.run_script("simple_default_reactor.py")
        assert "Spider closed (finished)" not in log
        assert "RuntimeError: AsyncCrawlerRunner requires AsyncioSelectorReactor" in log


@pytest.mark.parametrize(
    ("settings", "items"),
    [
        ({}, default_settings.LOG_VERSIONS),
        ({"LOG_VERSIONS": ["itemadapter"]}, ["itemadapter"]),
        ({"LOG_VERSIONS": []}, None),
    ],
)
def test_log_scrapy_info(settings, items, caplog):
    with caplog.at_level("INFO"):
        CrawlerProcess(settings, install_root_handler=False)
    assert (
        caplog.records[0].getMessage()
        == f"Scrapy {scrapy.__version__} started (bot: scrapybot)"
    ), repr(caplog.records[0].msg)
    if not items:
        assert len(caplog.records) == 1
        return
    version_string = caplog.records[1].getMessage()
    expected_items_pattern = "',\n '".join(
        f"{item}': '[^']+('\n +'[^']+)*" for item in items
    )
    assert re.search(r"^Versions:\n{'" + expected_items_pattern + "'}$", version_string)


@deferred_f_from_coro_f
async def test_deprecated_crawler_stop() -> None:
    crawler = get_crawler(DefaultSpider)
    d = crawler.crawl()
    await maybe_deferred_to_future(d)
    with pytest.warns(
        ScrapyDeprecationWarning, match=r"Crawler.stop\(\) is deprecated"
    ):
        await maybe_deferred_to_future(crawler.stop())