mirror of https://github.com/scrapy/scrapy.git
61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
"""Test that certain resources are not leaked during earlier tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from scrapy.utils.log import LogCounterHandler
|
|
from scrapy.utils.reactor import is_asyncio_reactor_installed, is_reactor_installed
|
|
from tests.utils.decorators import coroutine_test
|
|
|
|
|
|
def test_counter_handler() -> None:
|
|
"""Test that ``LogCounterHandler`` is always properly removed.
|
|
|
|
It's added in ``LogCount.spider_opened()`` and removed in
|
|
``LogCount.spider_closed()``.
|
|
"""
|
|
c = sum(1 for h in logging.root.handlers if isinstance(h, LogCounterHandler))
|
|
assert c == 0
|
|
|
|
|
|
def test_stderr_log_handler() -> None:
|
|
"""Test that the Scrapy root handler is always properly removed.
|
|
|
|
It's added in ``configure_logging()``, called by ``{Async,}CrawlerProcess``
|
|
(without ``install_root_handler=False``). It can be removed with
|
|
``_uninstall_scrapy_root_handler()`` if installing it was really needed.
|
|
"""
|
|
c = sum(1 for h in logging.root.handlers if type(h) is logging.StreamHandler) # pylint: disable=unidiomatic-typecheck
|
|
assert c == 0
|
|
|
|
|
|
@pytest.mark.only_asyncio
|
|
@coroutine_test
|
|
async def test_pending_asyncio_tasks() -> None:
|
|
"""Test that there are no pending asyncio tasks."""
|
|
# note that pytest-asyncio uses separate loops per function so this isn't as useful there
|
|
tasks = []
|
|
for t in asyncio.all_tasks():
|
|
coro = t.get_coro()
|
|
if (
|
|
coro is not None
|
|
and getattr(coro, "__name__", None) != "test_pending_asyncio_tasks"
|
|
):
|
|
tasks.append(t)
|
|
assert not tasks
|
|
|
|
|
|
def test_installed_reactor(reactor_pytest: str) -> None:
|
|
"""Test that the correct reactor is installed."""
|
|
match reactor_pytest:
|
|
case "asyncio":
|
|
assert is_asyncio_reactor_installed()
|
|
case "default":
|
|
assert not is_asyncio_reactor_installed()
|
|
case "none":
|
|
assert not is_reactor_installed()
|