mirror of https://github.com/scrapy/scrapy.git
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
from twisted.internet.error import CannotListenError
|
|
from twisted.internet.protocol import ServerFactory
|
|
|
|
from scrapy.utils.reactor import (
|
|
_asyncio_reactor_path,
|
|
install_reactor,
|
|
is_asyncio_reactor_installed,
|
|
listen_tcp,
|
|
set_asyncio_event_loop,
|
|
verify_installed_asyncio_event_loop,
|
|
verify_installed_reactor,
|
|
)
|
|
from tests.utils.decorators import coroutine_test
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Generator
|
|
|
|
from twisted.internet.tcp import Port
|
|
|
|
|
|
class TestAsyncio:
|
|
@pytest.mark.requires_reactor # needs a reactor
|
|
def test_is_asyncio_reactor_installed(self, reactor_pytest: str) -> None:
|
|
# the result should depend only on the pytest --reactor argument
|
|
assert is_asyncio_reactor_installed() == (reactor_pytest == "asyncio")
|
|
|
|
@pytest.mark.requires_reactor # installs a reactor
|
|
@pytest.mark.only_asyncio
|
|
@coroutine_test
|
|
async def test_set_asyncio_event_loop(self):
|
|
install_reactor(_asyncio_reactor_path)
|
|
assert set_asyncio_event_loop(None) is asyncio.get_running_loop()
|
|
|
|
|
|
class TestVerifyWithoutReactor:
|
|
@pytest.fixture(autouse=True)
|
|
def no_reactor(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delitem(sys.modules, "twisted.internet.reactor", raising=False)
|
|
|
|
def test_verify_installed_reactor(self) -> None:
|
|
with pytest.raises(
|
|
RuntimeError, match=r"verify_installed_reactor\(\) called without"
|
|
):
|
|
verify_installed_reactor(_asyncio_reactor_path)
|
|
|
|
def test_verify_installed_asyncio_event_loop(self) -> None:
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match=r"verify_installed_asyncio_event_loop\(\) called without",
|
|
):
|
|
verify_installed_asyncio_event_loop("asyncio.SelectorEventLoop")
|
|
|
|
|
|
@pytest.mark.requires_reactor # needs a reactor
|
|
class TestListenTcp:
|
|
@pytest.fixture
|
|
def ports(self) -> Generator[list[Port]]:
|
|
opened: list[Port] = []
|
|
yield opened
|
|
for port in opened:
|
|
port.stopListening()
|
|
|
|
@pytest.mark.parametrize("portrange", [[1, 2, 3], [8000, 7000]])
|
|
def test_invalid_portrange(self, portrange: list[int]) -> None:
|
|
with pytest.raises(ValueError, match="invalid portrange"):
|
|
listen_tcp(portrange, "127.0.0.1", ServerFactory())
|
|
|
|
def test_empty_portrange(self, ports: list[Port]) -> None:
|
|
port = listen_tcp([], "127.0.0.1", ServerFactory())
|
|
ports.append(port)
|
|
assert port.getHost().port > 0
|
|
|
|
def test_single_port(self, ports: list[Port]) -> None:
|
|
port = listen_tcp([0], "127.0.0.1", ServerFactory())
|
|
ports.append(port)
|
|
assert port.getHost().port > 0
|
|
|
|
def test_skips_used_ports(self, ports: list[Port]) -> None:
|
|
used = listen_tcp([], "127.0.0.1", ServerFactory())
|
|
ports.append(used)
|
|
used_number = used.getHost().port
|
|
|
|
port = listen_tcp([used_number, used_number + 50], "127.0.0.1", ServerFactory())
|
|
ports.append(port)
|
|
assert used_number < port.getHost().port <= used_number + 50
|
|
|
|
def test_no_free_port(self, ports: list[Port]) -> None:
|
|
used = listen_tcp([], "127.0.0.1", ServerFactory())
|
|
ports.append(used)
|
|
used_number = used.getHost().port
|
|
|
|
with pytest.raises(CannotListenError):
|
|
listen_tcp([used_number, used_number], "127.0.0.1", ServerFactory())
|