Add typing to scrapy/utils/test.py, fix a FTP test.

scrapy/crawler.py

@@ -4,11 +4,13 @@ import logging
 import pprint
 import signal
 import warnings
-from typing import TYPE_CHECKING, Optional, Type, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Set, Type, Union
 
 from twisted.internet import defer
 from zope.interface.exceptions import DoesNotImplement
 
+from scrapy.spiderloader import SpiderLoader
+
 try:
     # zope >= 5.0 only supports MultipleInvalid
     from zope.interface.exceptions import MultipleInvalid
@@ -171,7 +173,7 @@ class CrawlerRunner:
     )
 
     @staticmethod
-    def _get_spider_loader(settings):
+    def _get_spider_loader(settings) -> SpiderLoader:
         """Get SpiderLoader instance from settings"""
         cls_path = settings.get("SPIDER_LOADER_CLASS")
         loader_cls = load_object(cls_path)
@@ -190,13 +192,13 @@ class CrawlerRunner:
             )
         return loader_cls.from_settings(settings.frozencopy())
 
-    def __init__(self, settings=None):
+    def __init__(self, settings: Union[Dict[str, Any], Settings, None] = None):
         if isinstance(settings, dict) or settings is None:
             settings = Settings(settings)
         self.settings = settings
         self.spider_loader = self._get_spider_loader(settings)
-        self._crawlers = set()
-        self._active = set()
+        self._crawlers: Set[Crawler] = set()
+        self._active: Set[defer.Deferred] = set()
         self.bootstrap_failed = False
 
     @property
@@ -252,7 +254,9 @@ class CrawlerRunner:
 
         return d.addBoth(_done)
 
-    def create_crawler(self, crawler_or_spidercls):
+    def create_crawler(
+        self, crawler_or_spidercls: Union[Type[Spider], str, Crawler]
+    ) -> Crawler:
         """
         Return a :class:`~scrapy.crawler.Crawler` object.
 
@@ -272,7 +276,7 @@ class CrawlerRunner:
             return crawler_or_spidercls
         return self._create_crawler(crawler_or_spidercls)
 
-    def _create_crawler(self, spidercls):
+    def _create_crawler(self, spidercls: Union[str, Type[Spider]]) -> Crawler:
         if isinstance(spidercls, str):
             spidercls = self.spider_loader.load(spidercls)
         return Crawler(spidercls, self.settings)

scrapy/spiderloader.py

@@ -1,10 +1,13 @@
 import traceback
 import warnings
 from collections import defaultdict
+from typing import DefaultDict, Dict, List, Tuple, Type
 
 from zope.interface import implementer
 
+from scrapy import Spider
 from scrapy.interfaces import ISpiderLoader
+from scrapy.settings import BaseSettings
 from scrapy.utils.misc import walk_modules
 from scrapy.utils.spider import iter_spider_classes
 
@@ -16,11 +19,11 @@ class SpiderLoader:
     in a Scrapy project.
     """
 
-    def __init__(self, settings):
+    def __init__(self, settings: BaseSettings):
         self.spider_modules = settings.getlist("SPIDER_MODULES")
         self.warn_only = settings.getbool("SPIDER_LOADER_WARN_ONLY")
-        self._spiders = {}
-        self._found = defaultdict(list)
+        self._spiders: Dict[str, Type[Spider]] = {}
+        self._found: DefaultDict[str, List[Tuple[str, str]]] = defaultdict(list)
         self._load_all_spiders()
 
     def _check_name_duplicates(self):
@@ -68,7 +71,7 @@ class SpiderLoader:
     def from_settings(cls, settings):
         return cls(settings)
 
-    def load(self, spider_name):
+    def load(self, spider_name: str) -> Type[Spider]:
         """
         Return the Spider class for the given spider name. If the spider
         name is not found, raise a KeyError.

scrapy/utils/test.py

@@ -7,24 +7,30 @@ import os
 from importlib import import_module
 from pathlib import Path
 from posixpath import split
-from unittest import mock
+from typing import Any, Coroutine, Dict, List, Optional, Tuple, Type
+from unittest import TestCase, mock
 
+from twisted.internet.defer import Deferred
 from twisted.trial.unittest import SkipTest
 
+from scrapy import Spider
+from scrapy.crawler import Crawler
 from scrapy.utils.boto import is_botocore_available
 
 
-def assert_gcs_environ():
+def assert_gcs_environ() -> None:
     if "GCS_PROJECT_ID" not in os.environ:
         raise SkipTest("GCS_PROJECT_ID not found")
 
 
-def skip_if_no_boto():
+def skip_if_no_boto() -> None:
     if not is_botocore_available():
         raise SkipTest("missing botocore library")
 
 
-def get_gcs_content_and_delete(bucket, path):
+def get_gcs_content_and_delete(
+    bucket: Any, path: str
+) -> Tuple[bytes, List[Dict[str, str]], Any]:
     from google.cloud import storage
 
     client = storage.Client(project=os.environ.get("GCS_PROJECT_ID"))
@@ -37,8 +43,13 @@ def get_gcs_content_and_delete(bucket, path):
 
 
 def get_ftp_content_and_delete(
-    path, host, port, username, password, use_active_mode=False
-):
+    path: str,
+    host: str,
+    port: int,
+    username: str,
+    password: str,
+    use_active_mode: bool = False,
+) -> bytes:
     from ftplib import FTP
 
     ftp = FTP()
@@ -46,19 +57,23 @@ def get_ftp_content_and_delete(
     ftp.login(username, password)
     if use_active_mode:
         ftp.set_pasv(False)
-    ftp_data = []
+    ftp_data: List[bytes] = []
 
-    def buffer_data(data):
+    def buffer_data(data: bytes) -> None:
         ftp_data.append(data)
 
     ftp.retrbinary(f"RETR {path}", buffer_data)
     dirname, filename = split(path)
     ftp.cwd(dirname)
     ftp.delete(filename)
-    return "".join(ftp_data)
+    return b"".join(ftp_data)
 
 
-def get_crawler(spidercls=None, settings_dict=None, prevent_warnings=True):
+def get_crawler(
+    spidercls: Optional[Type[Spider]] = None,
+    settings_dict: Optional[Dict[str, Any]] = None,
+    prevent_warnings: bool = True,
+) -> Crawler:
     """Return an unconfigured Crawler object. If settings_dict is given, it
     will be used to populate the crawler settings with a project level
     priority.
@@ -82,7 +97,7 @@ def get_pythonpath() -> str:
     return str(Path(scrapy_path).parent) + os.pathsep + os.environ.get("PYTHONPATH", "")
 
 
-def get_testenv():
+def get_testenv() -> Dict[str, str]:
     """Return a OS environment dict suitable to fork processes that need to import
     this installation of Scrapy, instead of a system installed one.
     """
@@ -91,21 +106,23 @@ def get_testenv():
     return env
 
 
-def assert_samelines(testcase, text1, text2, msg=None):
+def assert_samelines(
+    testcase: TestCase, text1: str, text2: str, msg: Optional[str] = None
+) -> None:
     """Asserts text1 and text2 have the same lines, ignoring differences in
     line endings between platforms
     """
     testcase.assertEqual(text1.splitlines(), text2.splitlines(), msg)
 
 
-def get_from_asyncio_queue(value):
-    q = asyncio.Queue()
+def get_from_asyncio_queue(value: Any) -> Coroutine:
+    q: asyncio.Queue = asyncio.Queue()
     getter = q.get()
     q.put_nowait(value)
     return getter
 
 
-def mock_google_cloud_storage():
+def mock_google_cloud_storage() -> Tuple[Any, Any, Any]:
     """Creates autospec mocks for google-cloud-storage Client, Bucket and Blob
     classes and set their proper return values.
     """
@@ -122,7 +139,7 @@ def mock_google_cloud_storage():
     return (client_mock, bucket_mock, blob_mock)
 
 
-def get_web_client_agent_req(url):
+def get_web_client_agent_req(url: str) -> Deferred:
     from twisted.internet import reactor
     from twisted.web.client import Agent  # imports twisted.internet.reactor
 

tests/test_pipeline_files.py

@@ -32,6 +32,7 @@ from scrapy.utils.test import (
     get_gcs_content_and_delete,
     skip_if_no_boto,
 )
+from tests.mockserver import MockFTPServer
 
 from .test_pipeline_media import _mocked_download_func
 
@@ -639,31 +640,29 @@ class TestGCSFilesStore(unittest.TestCase):
 class TestFTPFileStore(unittest.TestCase):
     @defer.inlineCallbacks
     def test_persist(self):
-        uri = os.environ.get("FTP_TEST_FILE_URI")
-        if not uri:
-            raise unittest.SkipTest("No FTP URI available for testing")
         data = b"TestFTPFilesStore: \xe2\x98\x83"
         buf = BytesIO(data)
         meta = {"foo": "bar"}
         path = "full/filename"
-        store = FTPFilesStore(uri)
-        empty_dict = yield store.stat_file(path, info=None)
-        self.assertEqual(empty_dict, {})
-        yield store.persist_file(path, buf, info=None, meta=meta, headers=None)
-        stat = yield store.stat_file(path, info=None)
-        self.assertIn("last_modified", stat)
-        self.assertIn("checksum", stat)
-        self.assertEqual(stat["checksum"], "d113d66b2ec7258724a268bd88eef6b6")
-        path = f"{store.basedir}/{path}"
-        content = get_ftp_content_and_delete(
-            path,
-            store.host,
-            store.port,
-            store.username,
-            store.password,
-            store.USE_ACTIVE_MODE,
-        )
-        self.assertEqual(data.decode(), content)
+        with MockFTPServer() as ftp_server:
+            store = FTPFilesStore(ftp_server.url("/"))
+            empty_dict = yield store.stat_file(path, info=None)
+            self.assertEqual(empty_dict, {})
+            yield store.persist_file(path, buf, info=None, meta=meta, headers=None)
+            stat = yield store.stat_file(path, info=None)
+            self.assertIn("last_modified", stat)
+            self.assertIn("checksum", stat)
+            self.assertEqual(stat["checksum"], "d113d66b2ec7258724a268bd88eef6b6")
+            path = f"{store.basedir}/{path}"
+            content = get_ftp_content_and_delete(
+                path,
+                store.host,
+                store.port,
+                store.username,
+                store.password,
+                store.USE_ACTIVE_MODE,
+            )
+        self.assertEqual(data, content)
 
 
 class ItemWithFiles(Item):