scrapy/conftest.py at master · wavebyrd/scrapy · GitHub

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
from __future__ import annotations

import importlib
from pathlib import Path
from typing import TYPE_CHECKING

import pytest
from twisted.web.http import H2_ENABLED

from scrapy.utils.reactor import set_asyncio_event_loop_policy
from scrapy.utils.reactorless import install_reactor_import_hook
from tests.keys import generate_keys
from tests.mockserver.http import MockServer

if TYPE_CHECKING:
    from collections.abc import Generator


def _py_files(folder):
    return (str(p) for p in Path(folder).rglob("*.py"))


collect_ignore = [
    # may need extra deps
    "docs/_ext",
    # not a test, but looks like a test
    "scrapy/utils/testproc.py",
    "scrapy/utils/testsite.py",
    # contains scripts to be run by tests/test_crawler.py::AsyncCrawlerProcessSubprocess
    *_py_files("tests/AsyncCrawlerProcess"),
    # contains scripts to be run by tests/test_crawler.py::AsyncCrawlerRunnerSubprocess
    *_py_files("tests/AsyncCrawlerRunner"),
    # contains scripts to be run by tests/test_crawler.py::CrawlerProcessSubprocess
    *_py_files("tests/CrawlerProcess"),
    # contains scripts to be run by tests/test_crawler.py::CrawlerRunnerSubprocess
    *_py_files("tests/CrawlerRunner"),
]

base_dir = Path(__file__).parent
ignore_file_path = base_dir / "tests" / "ignores.txt"
with ignore_file_path.open(encoding="utf-8") as reader:
    for line in reader:
        file_path = line.strip()
        if file_path and file_path[0] != "#":
            collect_ignore.append(file_path)

if not H2_ENABLED:
    collect_ignore.extend(
        (
            "scrapy/core/downloader/handlers/http2.py",
            *_py_files("scrapy/core/http2"),
        )
    )

try:
    import httpx  # noqa: F401
except ImportError:
    collect_ignore.append("scrapy/core/downloader/handlers/_httpx.py")


def pytest_addoption(parser, pluginmanager):
    if pluginmanager.hasplugin("twisted"):
        return
    # add the full choice set so that pytest doesn't complain about invalid choices in some cases
    parser.addoption(
        "--reactor",
        default="none",
        choices=["asyncio", "default", "none"],
    )


@pytest.fixture(scope="session")
def mockserver() -> Generator[MockServer]:
    with MockServer() as mockserver:
        yield mockserver


@pytest.fixture(scope="session")
def reactor_pytest(request) -> str:
    return request.config.getoption("--reactor")


def pytest_configure(config):
    if config.getoption("--reactor") == "asyncio":
        # Needed on Windows to switch from proactor to selector for Twisted reactor compatibility.
        # If we decide to run tests with both, we will need to add a new option and check it here.
        set_asyncio_event_loop_policy()
    elif config.getoption("--reactor") == "none":
        install_reactor_import_hook()


def pytest_runtest_setup(item):
    # Skip tests based on reactor markers
    reactor = item.config.getoption("--reactor")

    if item.get_closest_marker("requires_reactor") and reactor == "none":
        pytest.skip('This test is only run when the --reactor value is not "none"')

    if item.get_closest_marker("only_asyncio") and reactor not in {"asyncio", "none"}:
        pytest.skip(
            'This test is only run when the --reactor value is "asyncio" (default) or "none"'
        )

    if item.get_closest_marker("only_not_asyncio") and reactor in {"asyncio", "none"}:
        pytest.skip(
            'This test is only run when the --reactor value is not "asyncio" (default) or "none"'
        )

    # Skip tests requiring optional dependencies
    optional_deps = [
        "uvloop",
        "botocore",
        "boto3",
        "mitmproxy",
    ]

    for module in optional_deps:
        if item.get_closest_marker(f"requires_{module}"):
            try:
                importlib.import_module(module)
            except ImportError:
                pytest.skip(f"{module} is not installed")


# Generate localhost certificate files, needed by some tests
generate_keys()