Close various garbage collectible resources (mostly in tests) (#7644)

* Set TELNETCONSOLE_ENABLED=False in get_crawler().

* Make _get_console_and_portal() a context manager.

* Allow PYTHONTRACEMALLOC in tox.

* Close the event loop in test_custom_asyncio_loop_enabled_false().

* Close the handler in _uninstall_scrapy_root_handler().

* Close queues in test_squeues.py.

* Close empty queues in init_prios().

* Close PopenSpawn stdin and stdout explicitly.

* Silence the unclosed socket warning.

* Implement more methods in CustomStatsCollector.

* Link the Twisted issue.

* Close the connection on download_maxsize.

* Fix typing.

* Add a comment about CustomStatsCollector.

* Restore the coverage.

* Properly close fixture queues.

* More robust file closing in feed storages.

* Close the file in TestMarshalItemExporter.test_nonstring_types_item().

* Cleanup temporary file handling in test_exporters.py.

* Close the file in test_stats_file_failed().

* Use an explicit TextIOWrapper in XmlItemExporter.

pyproject.toml

@@ -274,6 +274,8 @@ markers = [
 ]
 filterwarnings = [
     "ignore::DeprecationWarning:twisted.web.static",
+    # Twisted doesn't close failed sockets after CannotListenError: https://github.com/twisted/twisted/issues/6108
+    "ignore:Exception ignored in. <socket\\.socket.*laddr=..0\\.0\\.0\\.0., 0.:pytest.PytestUnraisableExceptionWarning",
 ]
 
 [tool.ruff.lint]

scrapy/core/downloader/handlers/http11.py

@@ -545,7 +545,8 @@ class _ScrapyAgent:
                 expected_size, maxsize, request, expected=True
             )
             logger.warning(warning_msg)
-            txresponse._transport.loseConnection()
+            # Abort connection immediately.
+            txresponse._transport._producer.abortConnection()
             raise DownloadCancelledError(warning_msg)
 
         if warnsize and expected_size > warnsize:

scrapy/exporters.py

@@ -171,7 +171,15 @@ class XmlItemExporter(BaseItemExporter):
         super().__init__(**kwargs)
         if not self.encoding:
             self.encoding = "utf-8"
-        self.xg = XMLGenerator(file, encoding=self.encoding)
+        # copied from xml.sax.saxutils._gettextwriter()
+        self.stream = TextIOWrapper(
+            file,
+            encoding=self.encoding,
+            errors="xmlcharrefreplace",
+            newline="\n",
+            write_through=True,
+        )
+        self.xg = XMLGenerator(self.stream, encoding=self.encoding)
 
     def _beautify_newline(self, new_item: bool = False) -> None:
         if self.indent is not None and (self.indent > 0 or new_item):
@@ -199,6 +207,7 @@ class XmlItemExporter(BaseItemExporter):
     def finish_exporting(self) -> None:
         self.xg.endElement(self.root_element)
         self.xg.endDocument()
+        self.stream.detach()  # Avoid closing the wrapped file.
 
     def _export_xml_field(self, name: str, serialized_value: Any, depth: int) -> None:
         self._beautify_indent(depth=depth)

scrapy/extensions/feedexport.py

@@ -250,20 +250,22 @@ class S3FeedStorage(BlockingFeedStorage):
 
     def _store_in_thread(self, file: IO[bytes]) -> None:
         file.seek(0)
-        if self.acl:
-            self.s3_client.upload_fileobj(
-                Bucket=self.bucketname,
-                Key=self.keyname,
-                Fileobj=file,
-                ExtraArgs={"ACL": self.acl},
-            )
-        else:
-            self.s3_client.upload_fileobj(
-                Bucket=self.bucketname,
-                Key=self.keyname,
-                Fileobj=file,
-            )
-        file.close()
+        try:
+            if self.acl:
+                self.s3_client.upload_fileobj(
+                    Bucket=self.bucketname,
+                    Key=self.keyname,
+                    Fileobj=file,
+                    ExtraArgs={"ACL": self.acl},
+                )
+            else:
+                self.s3_client.upload_fileobj(
+                    Bucket=self.bucketname,
+                    Key=self.keyname,
+                    Fileobj=file,
+                )
+        finally:
+            file.close()
 
 
 class GCSFeedStorage(BlockingFeedStorage):

scrapy/pqueues.py

@@ -141,10 +141,14 @@ class ScrapyPriorityQueue:
             q = self.qfactory(priority)
             if q:
                 self.queues[priority] = q
+            else:
+                q.close()
             if self._start_queue_cls:
                 q = self._sqfactory(priority)
                 if q:
                     self._start_queues[priority] = q
+                else:
+                    q.close()
 
         self.curprio = min(startprios)
 

scrapy/utils/ftp.py

@@ -1,4 +1,5 @@
 import posixpath
+from contextlib import closing
 from ftplib import FTP, error_perm
 from posixpath import dirname
 from typing import IO
@@ -32,7 +33,7 @@ def ftp_store_file(
     """Opens a FTP connection with passed credentials,sets current directory
     to the directory extracted from given path, then uploads the file to server
     """
-    with FTP() as ftp:
+    with FTP() as ftp, closing(file):
         ftp.connect(host, port)
         ftp.login(username, password)
         if use_active_mode:
@@ -42,4 +43,3 @@ def ftp_store_file(
         ftp_makedirs_cwd(ftp, dirname)
         command = "STOR" if overwrite else "APPE"
         ftp.storbinary(f"{command} {filename}", file)
-        file.close()

scrapy/utils/log.py

@@ -149,11 +149,12 @@ def install_scrapy_root_handler(settings: Settings) -> None:
 def _uninstall_scrapy_root_handler() -> None:
     global _scrapy_root_handler  # noqa: PLW0603
 
-    if (
-        _scrapy_root_handler is not None
-        and _scrapy_root_handler in logging.root.handlers
-    ):
+    if _scrapy_root_handler is None:
+        return
+
+    if _scrapy_root_handler in logging.root.handlers:
         logging.root.removeHandler(_scrapy_root_handler)
+    _scrapy_root_handler.close()
     _scrapy_root_handler = None
 
 

scrapy/utils/test.py

@@ -69,6 +69,7 @@ def get_crawler(
     # When needed, useful settings can be added here, e.g. ones that prevent
     # deprecation warnings.
     settings: dict[str, Any] = {
+        "TELNETCONSOLE_ENABLED": False,
         **get_reactor_settings(),
         **(settings_dict or {}),
     }

tests/test_command_runspider.py

@@ -212,6 +212,7 @@ class MySpider(scrapy.Spider):
             f"Using asyncio event loop: {loop.__module__}.{loop.__class__.__name__}"
             in log
         )
+        loop.close()
 
     def test_no_reactor(self, tmp_path: Path) -> None:
         log = self.get_log(

tests/test_command_shell.py

@@ -181,6 +181,10 @@ class TestInteractiveShell:
         p.expect_exact("HtmlResponse")
         p.sendeof()
         p.wait()  # type: ignore[no-untyped-call]
+        if p.proc.stdin:
+            p.proc.stdin.close()
+        if p.proc.stdout:
+            p.proc.stdout.close()
         logfile.seek(0)
         assert "Traceback" not in logfile.read().decode()
 

tests/test_crawler_subprocess.py

@@ -215,6 +215,10 @@ class TestCrawlerProcessSubprocessBase(ScriptRunnerMixin):
         p.expect_exact("shutting down gracefully")
         p.expect_exact("Spider closed (shutdown)")
         p.wait()  # type: ignore[no-untyped-call]
+        if p.proc.stdin:
+            p.proc.stdin.close()
+        if p.proc.stdout:
+            p.proc.stdout.close()
 
     def test_shutdown_graceful(self) -> None:
         self._test_shutdown_graceful()
@@ -232,6 +236,10 @@ class TestCrawlerProcessSubprocessBase(ScriptRunnerMixin):
         p.kill(sig)
         p.expect_exact("forcing unclean shutdown")
         p.wait()  # type: ignore[no-untyped-call]
+        if p.proc.stdin:
+            p.proc.stdin.close()
+        if p.proc.stdout:
+            p.proc.stdout.close()
 
     @coroutine_test
     async def test_shutdown_forced(self) -> None:

tests/test_exporters.py

@@ -3,7 +3,6 @@ import json
 import marshal
 import pickle
 import re
-import tempfile
 from abc import ABC, abstractmethod
 from datetime import datetime
 from io import BytesIO
@@ -242,7 +241,6 @@ class TestPickleItemExporterDataclass(TestPickleItemExporter):
 
 class TestMarshalItemExporter(TestBaseItemExporter):
     def _get_exporter(self, **kwargs):
-        self.output = tempfile.TemporaryFile()
         return MarshalItemExporter(self.output, **kwargs)
 
     def _check_output(self):
@@ -252,7 +250,7 @@ class TestMarshalItemExporter(TestBaseItemExporter):
     def test_nonstring_types_item(self):
         item = self._get_nonstring_types_item()
         item.pop("time")  # datetime is not marshallable
-        fp = tempfile.TemporaryFile()
+        fp = BytesIO()
         ie = MarshalItemExporter(fp)
         ie.start_exporting()
         ie.export_item(item)
@@ -260,6 +258,7 @@ class TestMarshalItemExporter(TestBaseItemExporter):
         del ie  # See the first “del self.ie” in this file for context.
         fp.seek(0)
         assert marshal.load(fp) == item
+        fp.close()
 
 
 class TestMarshalItemExporterDataclass(TestMarshalItemExporter):
@@ -269,7 +268,11 @@ class TestMarshalItemExporterDataclass(TestMarshalItemExporter):
 
 class TestCsvItemExporter(TestBaseItemExporter):
     def _get_exporter(self, **kwargs):
-        self.output = tempfile.TemporaryFile()
+        # We need a fresh instance for each exporter, because
+        # CsvItemExporter.stream.__del__() closes the underlying file
+        # (CsvItemExporter.finish_exporting() calls detach() but not all tests
+        # call it).
+        self.output = BytesIO()
         return CsvItemExporter(self.output, **kwargs)
 
     def assertCsvEqual(self, first, second, msg=None):
@@ -389,6 +392,11 @@ class TestCsvItemExporterDataclass(TestCsvItemExporter):
 
 class TestXmlItemExporter(TestBaseItemExporter):
     def _get_exporter(self, **kwargs):
+        # We need a fresh instance for each exporter, because
+        # XmlItemExporter.stream.__del__() closes the underlying file
+        # (XmlItemExporter.finish_exporting() calls detach() but not all tests
+        # call it).
+        self.output = BytesIO()
         return XmlItemExporter(self.output, **kwargs)
 
     def assertXmlEquivalent(self, first, second, msg=None):

tests/test_extension_telnet.py

@@ -1,61 +1,86 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Any
+
 import pytest
 from twisted.conch.telnet import ITelnetProtocol
 from twisted.cred import credentials
 
 from scrapy.extensions.telnet import TelnetConsole
+from scrapy.utils.defer import maybe_deferred_to_future
 from scrapy.utils.test import get_crawler
-from tests.utils.decorators import inline_callbacks_test
+from tests.utils.decorators import coroutine_test
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    from scrapy.crawler import Crawler
 
 pytestmark = pytest.mark.requires_reactor  # TelnetConsole requires a reactor
 
 
-class TestTelnetExtension:
-    def _get_console_and_portal(self, settings=None):
-        crawler = get_crawler(settings_dict=settings)
-        console = TelnetConsole(crawler)
+def _get_crawler(settings_dict: dict[str, Any] | None = None) -> Crawler:
+    settings = {
+        "TELNETCONSOLE_ENABLED": True,
+        **(settings_dict or {}),
+    }
+    return get_crawler(settings_dict=settings)
 
-        # This function has some side effects we don't need for this test
-        console._get_telnet_vars = dict
 
-        console.start_listening()
-        protocol = console.protocol()
-        portal = protocol.protocolArgs[0]
+@contextmanager
+def _get_console_and_portal(
+    settings: dict[str, Any] | None = None,
+) -> Generator[tuple[TelnetConsole, Any]]:
+    crawler = _get_crawler(settings_dict=settings)
+    console = TelnetConsole(crawler)
 
-        return console, portal
+    # This function has some side effects we don't need for this test
+    console._get_telnet_vars = dict  # type: ignore[method-assign]
 
-    @inline_callbacks_test
-    def test_bad_credentials(self):
-        console, portal = self._get_console_and_portal()
+    console.start_listening()
+    protocol = console.protocol()
+    portal = protocol.protocolArgs[0]
+
+    try:
+        yield console, portal
+    finally:
+        console.stop_listening()
+
+
+@coroutine_test
+async def test_bad_credentials() -> None:
+    with _get_console_and_portal() as (_, portal):
         creds = credentials.UsernamePassword(b"username", b"password")
         d = portal.login(creds, None, ITelnetProtocol)
         with pytest.raises(ValueError, match="Invalid credentials"):
-            yield d
-        console.stop_listening()
+            await maybe_deferred_to_future(d)
 
-    @inline_callbacks_test
-    def test_good_credentials(self):
-        console, portal = self._get_console_and_portal()
+
+@coroutine_test
+async def test_good_credentials() -> None:
+    with _get_console_and_portal() as (console, portal):
         creds = credentials.UsernamePassword(
             console.username.encode("utf8"), console.password.encode("utf8")
         )
         d = portal.login(creds, None, ITelnetProtocol)
-        yield d
-        console.stop_listening()
+        await maybe_deferred_to_future(d)
 
-    @inline_callbacks_test
-    def test_custom_credentials(self):
-        settings = {
-            "TELNETCONSOLE_USERNAME": "user",
-            "TELNETCONSOLE_PASSWORD": "pass",
-        }
-        console, portal = self._get_console_and_portal(settings=settings)
+
+@coroutine_test
+async def test_custom_credentials() -> None:
+    settings = {
+        "TELNETCONSOLE_USERNAME": "user",
+        "TELNETCONSOLE_PASSWORD": "pass",
+    }
+    with _get_console_and_portal(settings=settings) as (_, portal):
         creds = credentials.UsernamePassword(b"user", b"pass")
         d = portal.login(creds, None, ITelnetProtocol)
-        yield d
-        console.stop_listening()
+        await maybe_deferred_to_future(d)
 
-    def test_invalid_reversed_portrange(self):
-        settings = {"TELNETCONSOLE_PORT": [2, 1]}
-        console = TelnetConsole(get_crawler(settings_dict=settings))
-        with pytest.raises(ValueError, match=r"invalid portrange: \[2, 1\]"):
-            console.start_listening()
+
+def test_invalid_reversed_portrange() -> None:
+    settings = {"TELNETCONSOLE_PORT": [2, 1]}
+    console = TelnetConsole(_get_crawler(settings_dict=settings))
+    with pytest.raises(ValueError, match=r"invalid portrange: \[2, 1\]"):
+        console.start_listening()

tests/test_feedexport.py

@@ -87,6 +87,7 @@ class DummyBlockingFeedStorage(BlockingFeedStorage):
 
 class FailingBlockingFeedStorage(DummyBlockingFeedStorage):
     def _store_in_thread(self, file):
+        file.close()
         raise OSError("Cannot store")
 
 
@@ -477,9 +478,14 @@ class TestFeedExport(TestFeedExportBase):
             },
         }
         crawler = get_crawler(ItemSpider, settings)
+
+        def store(file: IO[bytes]) -> None:
+            file.close()
+            raise KeyError("foo")
+
         with mock.patch(
             "scrapy.extensions.feedexport.FileFeedStorage.store",
-            side_effect=KeyError("foo"),
+            side_effect=store,
         ):
             yield crawler.crawl(mockserver=self.mockserver)
         assert "feedexport/failed_count/FileFeedStorage" in crawler.stats.get_stats()

tests/test_squeues.py

@@ -42,6 +42,7 @@ def nonserializable_object_test(self):
         ValueError, match=r"unmarshallable object|can't pickle Selector objects"
     ):
         q.push(sel)
+    q.close()
 
 
 class FifoDiskQueueTestMixin:
@@ -53,6 +54,7 @@ class FifoDiskQueueTestMixin:
         assert q.pop() == "a"
         assert q.pop() == 123
         assert q.pop() == {"a": "dict"}
+        q.close()
 
     test_nonserializable_object = nonserializable_object_test
 
@@ -93,6 +95,7 @@ class PickleFifoDiskQueueTest(t.FifoDiskQueueTest, FifoDiskQueueTestMixin):
         i2 = q.pop()
         assert isinstance(i2, MyItem)
         assert i == i2
+        q.close()
 
     def test_serialize_loader(self):
         q = self.queue()
@@ -102,6 +105,7 @@ class PickleFifoDiskQueueTest(t.FifoDiskQueueTest, FifoDiskQueueTestMixin):
         assert isinstance(loader2, MyLoader)
         assert loader2.default_item_class is MyItem
         assert loader2.name_out("x") == "xx"
+        q.close()
 
     def test_serialize_request_recursive(self):
         q = self.queue()
@@ -112,6 +116,7 @@ class PickleFifoDiskQueueTest(t.FifoDiskQueueTest, FifoDiskQueueTestMixin):
         assert isinstance(r2, Request)
         assert r.url == r2.url
         assert r2.meta["request"] is r2
+        q.close()
 
     def test_non_pickable_object(self):
         q = self.queue()
@@ -158,6 +163,7 @@ class LifoDiskQueueTestMixin:
         assert q.pop() == {"a": "dict"}
         assert q.pop() == 123
         assert q.pop() == "a"
+        q.close()
 
     test_nonserializable_object = nonserializable_object_test
 
@@ -178,6 +184,7 @@ class PickleLifoDiskQueueTest(t.LifoDiskQueueTest, LifoDiskQueueTestMixin):
         i2 = q.pop()
         assert isinstance(i2, MyItem)
         assert i == i2
+        q.close()
 
     def test_serialize_loader(self):
         q = self.queue()
@@ -187,6 +194,7 @@ class PickleLifoDiskQueueTest(t.LifoDiskQueueTest, LifoDiskQueueTestMixin):
         assert isinstance(loader2, MyLoader)
         assert loader2.default_item_class is MyItem
         assert loader2.name_out("x") == "xx"
+        q.close()
 
     def test_serialize_request_recursive(self):
         q = self.queue()
@@ -197,3 +205,4 @@ class PickleLifoDiskQueueTest(t.LifoDiskQueueTest, LifoDiskQueueTestMixin):
         assert isinstance(r2, Request)
         assert r.url == r2.url
         assert r2.meta["request"] is r2
+        q.close()

tests/test_squeues_request.py

@@ -70,7 +70,6 @@ class TestRequestQueueBase(ABC):
         if test_peek:
             assert q.peek() is None
         assert q.pop() is None
-        q.close()
 
     @pytest.mark.parametrize("test_peek", [True, False])
     def test_order(self, q: queuelib.queue.BaseQueue, test_peek: bool):
@@ -108,7 +107,6 @@ class TestRequestQueueBase(ABC):
         if test_peek:
             assert q.peek() is None
         assert q.pop() is None
-        q.close()
 
 
 class TestPickleFifoDiskQueueRequest(TestRequestQueueBase):
@@ -116,9 +114,13 @@ class TestPickleFifoDiskQueueRequest(TestRequestQueueBase):
 
     @pytest.fixture
     def q(self, crawler, tmp_path):
-        return PickleFifoDiskQueue.from_crawler(
+        queue = PickleFifoDiskQueue.from_crawler(
             crawler=crawler, key=str(tmp_path / "pickle" / "fifo")
         )
+        try:
+            yield queue
+        finally:
+            queue.close()
 
 
 class TestPickleLifoDiskQueueRequest(TestRequestQueueBase):
@@ -126,9 +128,13 @@ class TestPickleLifoDiskQueueRequest(TestRequestQueueBase):
 
     @pytest.fixture
     def q(self, crawler, tmp_path):
-        return PickleLifoDiskQueue.from_crawler(
+        queue = PickleLifoDiskQueue.from_crawler(
             crawler=crawler, key=str(tmp_path / "pickle" / "lifo")
         )
+        try:
+            yield queue
+        finally:
+            queue.close()
 
 
 class TestMarshalFifoDiskQueueRequest(TestRequestQueueBase):
@@ -136,9 +142,13 @@ class TestMarshalFifoDiskQueueRequest(TestRequestQueueBase):
 
     @pytest.fixture
     def q(self, crawler, tmp_path):
-        return MarshalFifoDiskQueue.from_crawler(
+        queue = MarshalFifoDiskQueue.from_crawler(
             crawler=crawler, key=str(tmp_path / "marshal" / "fifo")
         )
+        try:
+            yield queue
+        finally:
+            queue.close()
 
 
 class TestMarshalLifoDiskQueueRequest(TestRequestQueueBase):
@@ -146,9 +156,13 @@ class TestMarshalLifoDiskQueueRequest(TestRequestQueueBase):
 
     @pytest.fixture
     def q(self, crawler, tmp_path):
-        return MarshalLifoDiskQueue.from_crawler(
+        queue = MarshalLifoDiskQueue.from_crawler(
             crawler=crawler, key=str(tmp_path / "marshal" / "lifo")
         )
+        try:
+            yield queue
+        finally:
+            queue.close()
 
 
 class TestFifoMemoryQueueRequest(TestRequestQueueBase):

tests/test_stats.py

@@ -131,6 +131,7 @@ class TestStatsCollector:
 
     @coroutine_test
     async def test_deprecated_spider_arg_custom_collector(self) -> None:
+        # the class reimplements many methods because those are called during the test crawl
         class CustomStatsCollector:
             def __init__(self, crawler):
                 self._stats = {}
@@ -141,10 +142,19 @@ class TestStatsCollector:
             def get_stats(self, spider=None):
                 return self._stats
 
+            def get_value(self, key, default=None, spider=None):
+                return self._stats.get(key, default)
+
+            def set_value(self, key, value, spider=None):
+                self._stats[key] = value
+
             def inc_value(self, key, count=1, start=0, spider=None):
                 d = self._stats
                 d[key] = d.setdefault(key, start) + count
 
+            def max_value(self, key, value, spider=None) -> None:
+                self._stats[key] = max(self._stats.setdefault(key, value), value)
+
             def close_spider(self, spider, reason):
                 pass
 

tests/test_utils_asyncio.py

@@ -149,3 +149,12 @@ class TestAsyncioLoopingCall:
         with pytest.raises(TypeError):
             looping_call.start(0.1)
         assert not looping_call.running
+
+    @coroutine_test
+    async def test_looping_function_raises(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        looping_call = AsyncioLoopingCall(lambda: 1 / 0)
+        looping_call.start(0.1)
+        assert not looping_call.running
+        assert "Error calling the AsyncioLoopingCall function" in caplog.text

tox.ini

@@ -53,6 +53,7 @@ deps =
     {[test-requirements]deps}
     pytest >= 8.4.1  # https://github.com/pytest-dev/pytest/pull/13502
 passenv =
+    PYTHONTRACEMALLOC
     PYTEST_ADDOPTS
     S3_TEST_FILE_URI
     AWS_ACCESS_KEY_ID