tests/test_contracts.py - master - githubmirror/scrapy

tests/test_contracts.py
783 строки24 KB
Adrian
Improve test coverage for contracts (#7677)
26 июн 2026, 12:47
26 июн 2026, 12:470ccddb4
from unittest import TextTestResult

import pytest
from twisted.python import failure

from scrapy.contracts import Contract, ContractsManager
from scrapy.contracts.default import (
    CallbackKeywordArgumentsContract,
    MetadataContract,
    ReturnsContract,
    ScrapesContract,
    UrlContract,
)
from scrapy.http import Request
from scrapy.item import Field, Item
from scrapy.spidermiddlewares.httperror import HttpError
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.utils.decorators import inline_callbacks_test


class DemoItem(Item):
    name = Field()
    url = Field()


class ResponseMock:
    url = "http://scrapy.org"


class ResponseMetaMock(ResponseMock):
    meta = None


class TaggedRequest(Request):
    def __init__(self, url, contract_tag=None, **kwargs):
        super().__init__(url, **kwargs)
        self.contract_tag = contract_tag


class CustomSuccessContract(Contract):
    name = "custom_success_contract"

    def adjust_request_args(self, args):
        args["url"] = "http://scrapy.org"
        return args


class CustomFailContract(Contract):
    name = "custom_fail_contract"

    def adjust_request_args(self, args):
        raise TypeError("Error in adjust_request_args")


class CustomTaggedRequestContract(Contract):
    name = "custom_tagged_request"
    request_cls = TaggedRequest

    def adjust_request_args(self, args):
        args["contract_tag"] = "custom"
        args["method"] = "POST"
        return args


class DemoSpider(Spider):
    name = "demo_spider"

    def returns_request(self, response):
        """method which returns request
        @url http://scrapy.org
        @returns requests 1
        """
        return Request("http://scrapy.org", callback=self.returns_item)

    async def returns_request_async(self, response):
        """async method which returns request
        @url http://scrapy.org
        @returns requests 1
        """
        return Request("http://scrapy.org", callback=self.returns_item)

    def returns_item(self, response):
        """method which returns item
        @url http://scrapy.org
        @returns items 1 1
        """
        return DemoItem(url=response.url)

    def returns_request_cb_kwargs(self, response, url):
        """method which returns request
        @url https://example.org
        @cb_kwargs {"url": "http://scrapy.org"}
        @returns requests 1
        """
        return Request(url, callback=self.returns_item_cb_kwargs)

    def returns_item_cb_kwargs(self, response, name):
        """method which returns item
        @url http://scrapy.org
        @cb_kwargs {"name": "Scrapy"}
        @returns items 1 1
        """
        return DemoItem(name=name, url=response.url)

    def returns_item_cb_kwargs_error_unexpected_keyword(self, response):
        """method which returns item
        @url http://scrapy.org
        @cb_kwargs {"arg": "value"}
        @returns items 1 1
        """
        return DemoItem(url=response.url)

    def returns_item_cb_kwargs_error_missing_argument(self, response, arg):
        """method which returns item
        @url http://scrapy.org
        @returns items 1 1
        """
        return DemoItem(url=response.url)

    def returns_dict_item(self, response):
        """method which returns item
        @url http://scrapy.org
        @returns items 1 1
        """
        return {"url": response.url}

    def returns_fail(self, response):
        """method which returns item
        @url http://scrapy.org
        @returns items 0 0
        """
        return DemoItem(url=response.url)

    def returns_request_range_fail(self, response):
        """method which returns fewer requests than the expected range
        @url http://scrapy.org
        @returns requests 2 3
        """
        return Request("http://scrapy.org", callback=self.returns_item)

    def yields_item_and_request(self, response):
        """yields one item and one request
        @url http://scrapy.org
        @returns items 1 1
        @scrapes name url
        """
        yield DemoItem(name="test", url=response.url)
        yield Request("http://scrapy.org", callback=self.returns_item)

    async def returns_async_gen(self, response):
        """async generator callback
        @url http://scrapy.org
        @returns items 1 1
        """
        yield DemoItem(url=response.url)

    def returns_dict_fail(self, response):
        """method which returns item
        @url http://scrapy.org
        @returns items 0 0
        """
        return {"url": response.url}

    def scrapes_item_ok(self, response):
        """returns item with name and url
        @url http://scrapy.org
        @returns items 1 1
        @scrapes name url
        """
        return DemoItem(name="test", url=response.url)

    def scrapes_dict_item_ok(self, response):
        """returns item with name and url
        @url http://scrapy.org
        @returns items 1 1
        @scrapes name url
        """
        return {"name": "test", "url": response.url}

    def scrapes_item_fail(self, response):
        """returns item with no name
        @url http://scrapy.org
        @returns items 1 1
        @scrapes name url
        """
        return DemoItem(url=response.url)

    def scrapes_dict_item_fail(self, response):
        """returns item with no name
        @url http://scrapy.org
        @returns items 1 1
        @scrapes name url
        """
        return {"url": response.url}

    def scrapes_multiple_missing_fields(self, response):
        """returns item with no name
        @url http://scrapy.org
        @returns items 1 1
        @scrapes name url
        """
        return {}

    def parse_no_url(self, response):
        """method with no url
        @returns items 1 1
        """

    def custom_tagged_request(self, response):
        """
        @url http://scrapy.org
        @custom_tagged_request
        """

    def invalid_regex(self, response):
        """method with invalid regex
        @ Scrapy is awsome
        """

    def invalid_regex_with_valid_contract(self, response):
        """method with invalid regex
        @ scrapy is awsome
        @url http://scrapy.org
        """

    def returns_request_meta(self, response):
        """method which returns request
        @url https://example.org
        @meta {"cookiejar": "session1"}
        @returns requests 1
        """
        return Request(
            "https://example.org", meta=response.meta, callback=self.returns_item_meta
        )

    def returns_item_meta(self, response):
        """method which returns item
        @url http://scrapy.org
        @meta {"key": "example"}
        @returns items 1 1
        """
        return DemoItem(name="example", url=response.url)

    def returns_error_missing_meta(self, response):
        """method which depends of metadata be defined

        @url http://scrapy.org
        @returns items 1
        """
        key = response.meta["key"]
        yield {key: "value"}


class CustomContractSuccessSpider(Spider):
    name = "custom_contract_success_spider"

    def parse(self, response):
        """
        @custom_success_contract
        """


class CustomContractFailSpider(Spider):
    name = "custom_contract_fail_spider"

    def parse(self, response):
        """
        @custom_fail_contract
        """


class InheritsDemoSpider(DemoSpider):
    name = "inherits_demo_spider"


class TestContractsManager:
    contracts = [
        UrlContract,
        CallbackKeywordArgumentsContract,
        MetadataContract,
        ReturnsContract,
        ScrapesContract,
        CustomTaggedRequestContract,
        CustomSuccessContract,
        CustomFailContract,
    ]

    def setup_method(self):
        self.conman = ContractsManager(self.contracts)
        self.results = TextTestResult(stream=None, descriptions=False, verbosity=0)

    def should_succeed(self):
        assert not self.results.failures
        assert not self.results.errors

    def should_fail(self):
        assert self.results.failures
        assert not self.results.errors

    def should_error(self):
        assert self.results.errors

    def test_contracts(self):
        spider = DemoSpider()

        # extract contracts correctly
        contracts = self.conman.extract_contracts(spider.returns_request)
        assert len(contracts) == 2
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, ReturnsContract]
        )

        # returns request for valid method
        request = self.conman.from_method(spider.returns_request, self.results)
        assert request is not None

        # no request for missing url
        request = self.conman.from_method(spider.parse_no_url, self.results)
        assert request is None

    def test_cb_kwargs(self):
        spider = DemoSpider()
        response = ResponseMock()

        # extract contracts correctly
        contracts = self.conman.extract_contracts(spider.returns_request_cb_kwargs)
        assert len(contracts) == 3
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, CallbackKeywordArgumentsContract, ReturnsContract]
        )

        contracts = self.conman.extract_contracts(spider.returns_item_cb_kwargs)
        assert len(contracts) == 3
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, CallbackKeywordArgumentsContract, ReturnsContract]
        )

        contracts = self.conman.extract_contracts(
            spider.returns_item_cb_kwargs_error_unexpected_keyword
        )
        assert len(contracts) == 3
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, CallbackKeywordArgumentsContract, ReturnsContract]
        )

        contracts = self.conman.extract_contracts(
            spider.returns_item_cb_kwargs_error_missing_argument
        )
        assert len(contracts) == 2
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, ReturnsContract]
        )

        # returns_request
        request = self.conman.from_method(
            spider.returns_request_cb_kwargs, self.results
        )
        request.callback(response, **request.cb_kwargs)
        self.should_succeed()

        # returns_item
        request = self.conman.from_method(spider.returns_item_cb_kwargs, self.results)
        request.callback(response, **request.cb_kwargs)
        self.should_succeed()

        # returns_item (error, callback doesn't take keyword arguments)
        request = self.conman.from_method(
            spider.returns_item_cb_kwargs_error_unexpected_keyword, self.results
        )
        request.callback(response, **request.cb_kwargs)
        self.should_error()

        # returns_item (error, contract doesn't provide keyword arguments)
        request = self.conman.from_method(
            spider.returns_item_cb_kwargs_error_missing_argument, self.results
        )
        request.callback(response, **request.cb_kwargs)
        self.should_error()

    def test_meta(self):
        spider = DemoSpider()

        # extract contracts correctly
        contracts = self.conman.extract_contracts(spider.returns_request_meta)
        assert len(contracts) == 3
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, MetadataContract, ReturnsContract]
        )

        contracts = self.conman.extract_contracts(spider.returns_item_meta)
        assert len(contracts) == 3
        assert frozenset(type(x) for x in contracts) == frozenset(
            [UrlContract, MetadataContract, ReturnsContract]
        )

        response = ResponseMetaMock()

        # returns_request
        request = self.conman.from_method(spider.returns_request_meta, self.results)
        assert request.meta["cookiejar"] == "session1"
        response.meta = request.meta
        request.callback(response)
        assert response.meta["cookiejar"] == "session1"
        self.should_succeed()

        response = ResponseMetaMock()

        # returns_item
        request = self.conman.from_method(spider.returns_item_meta, self.results)
        assert request.meta["key"] == "example"
        response.meta = request.meta
        request.callback(response)
        assert response.meta["key"] == "example"
        self.should_succeed()

        response = ResponseMetaMock()

        request = self.conman.from_method(
            spider.returns_error_missing_meta, self.results
        )
        request.callback(response)
        self.should_error()

    def test_returns(self):
        spider = DemoSpider()
        response = ResponseMock()

        # returns_item
        request = self.conman.from_method(spider.returns_item, self.results)
        request.callback(response)
        self.should_succeed()

        # returns_dict_item
        request = self.conman.from_method(spider.returns_dict_item, self.results)
        request.callback(response)
        self.should_succeed()

        # returns_request
        request = self.conman.from_method(spider.returns_request, self.results)
        request.callback(response)
        self.should_succeed()

        # returns_fail
        request = self.conman.from_method(spider.returns_fail, self.results)
        request.callback(response)
        self.should_fail()

        # returns_dict_fail
        request = self.conman.from_method(spider.returns_dict_fail, self.results)
        request.callback(response)
        self.should_fail()

    def test_returns_async(self):
        spider = DemoSpider()
        response = ResponseMock()

        request = self.conman.from_method(spider.returns_request_async, self.results)
        request.callback(response)
        self.should_error()

    def test_returns_invalid_argument_count(self):
        spider = DemoSpider()
        with pytest.raises(ValueError, match="expected 1, 2 or 3, got 0"):
            ReturnsContract(spider.returns_item)
        with pytest.raises(ValueError, match="expected 1, 2 or 3, got 4"):
            ReturnsContract(spider.returns_item, "items", "1", "2", "3")

    def test_returns_default_bounds(self):
        spider = DemoSpider()
        contract = ReturnsContract(spider.returns_item, "items")
        assert contract.min_bound == 1
        assert contract.max_bound == float("inf")

    def test_returns_range_fail(self):
        spider = DemoSpider()
        response = ResponseMock()

        request = self.conman.from_method(
            spider.returns_request_range_fail, self.results
        )
        request.callback(response)
        self.should_fail()
        assert "expected 2..3" in self.results.failures[-1][-1]

    def test_returns_and_scrapes_ignore_other_types(self):
        spider = DemoSpider()
        response = ResponseMock()

        # @returns and @scrapes only count matching output objects and skip
        # the request that is also yielded.
        request = self.conman.from_method(spider.yields_item_and_request, self.results)
        request.callback(response)
        self.should_succeed()

    def test_testcase_str(self):
        spider = DemoSpider()
        contract = UrlContract(spider.returns_request, "http://scrapy.org")
        assert (
            str(contract.testcase_pre)
            == "[demo_spider] returns_request (@url pre-hook)"
        )

    def test_scrapes(self):
        spider = DemoSpider()
        response = ResponseMock()

        # scrapes_item_ok
        request = self.conman.from_method(spider.scrapes_item_ok, self.results)
        request.callback(response)
        self.should_succeed()

        # scrapes_dict_item_ok
        request = self.conman.from_method(spider.scrapes_dict_item_ok, self.results)
        request.callback(response)
        self.should_succeed()

        # scrapes_item_fail
        request = self.conman.from_method(spider.scrapes_item_fail, self.results)
        request.callback(response)
        self.should_fail()

        # scrapes_dict_item_fail
        request = self.conman.from_method(spider.scrapes_dict_item_fail, self.results)
        request.callback(response)
        self.should_fail()

        # scrapes_multiple_missing_fields
        request = self.conman.from_method(
            spider.scrapes_multiple_missing_fields, self.results
        )
        request.callback(response)
        self.should_fail()
        message = "ContractFail: Missing fields: name, url"
        assert message in self.results.failures[-1][-1]

    def test_regex(self):
        spider = DemoSpider()
        response = ResponseMock()

        # invalid regex
        request = self.conman.from_method(spider.invalid_regex, self.results)
        assert request is None

        # invalid regex with valid contract
        request = self.conman.from_method(
            spider.invalid_regex_with_valid_contract, self.results
        )
        request.callback(response)
        self.should_succeed()

    def test_custom_contracts(self):
        self.conman.from_spider(CustomContractSuccessSpider(), self.results)
        self.should_succeed()

        self.conman.from_spider(CustomContractFailSpider(), self.results)
        self.should_error()

    def test_errback(self):
        spider = DemoSpider()
        response = ResponseMock()

        try:
            raise HttpError(response, "Ignoring non-200 response")
        except HttpError:
            failure_mock = failure.Failure()

        request = self.conman.from_method(spider.returns_request, self.results)
        request.errback(failure_mock)

        assert not self.results.failures
        assert self.results.errors

    @inline_callbacks_test
    def test_same_url(self):
        class TestSameUrlSpider(Spider):
            name = "test_same_url"

            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self.visited = 0

            async def start(self_):  # pylint: disable=no-self-argument
                for item_or_request in self.conman.from_spider(self_, self.results):
                    yield item_or_request

            def parse_first(self, response):
                self.visited += 1
                return DemoItem()

            def parse_second(self, response):
                self.visited += 1
                return DemoItem()

        with MockServer() as mockserver:
            contract_doc = f"@url {mockserver.url('/status?n=200')}"

            TestSameUrlSpider.parse_first.__doc__ = contract_doc
            TestSameUrlSpider.parse_second.__doc__ = contract_doc

            crawler = get_crawler(TestSameUrlSpider)
            yield crawler.crawl()

        assert crawler.spider.visited == 2

    def test_custom_tagged_request_contract(self):
        spider = DemoSpider()
        request = self.conman.from_method(spider.custom_tagged_request, self.results)
        assert request.method == "POST"
        assert isinstance(request, TaggedRequest)
        assert request.contract_tag == "custom"

    def test_inherited_contracts(self):
        spider = InheritsDemoSpider()

        requests = self.conman.from_spider(spider, self.results)
        assert requests
        assert any(
            isinstance(request, TaggedRequest) for request in requests if request
        )


class CustomFailContractPreProcess(Contract):
    name = "test_contract"

    def pre_process(self, response):
        raise KeyboardInterrupt("Pre-process exception")


class CustomFailContractPostProcess(Contract):
    name = "test_contract"

    def post_process(self, response):
        raise KeyboardInterrupt("Post-process exception")


class PreProcessSuccessContract(Contract):
    name = "pre_success"

    def pre_process(self, response):
        return


class PreProcessAssertionFailContract(Contract):
    name = "pre_assertion_fail"

    def pre_process(self, response):
        raise AssertionError("pre-process assertion")


class PreProcessErrorContract(Contract):
    name = "pre_error"

    def pre_process(self, response):
        raise ValueError("pre-process error")


class PostProcessSuccessContract(Contract):
    name = "post_success"

    def post_process(self, output):
        return


class PostProcessErrorContract(Contract):
    name = "post_error"

    def post_process(self, output):
        raise ValueError("post-process error")


class TestCustomContractPrePostProcess:
    def setup_method(self):
        self.results = TextTestResult(stream=None, descriptions=False, verbosity=0)

    def test_pre_hook_keyboard_interrupt(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = CustomFailContractPreProcess(spider.returns_request)
        conman = ContractsManager([UrlContract, ReturnsContract, contract])

        request = conman.from_method(spider.returns_request, self.results)
        contract.add_pre_hook(request, self.results)
        with pytest.raises(KeyboardInterrupt, match="Pre-process exception"):
            request.callback(response, **request.cb_kwargs)

        assert not self.results.failures
        assert not self.results.errors

    def test_post_hook_keyboard_interrupt(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = CustomFailContractPostProcess(spider.returns_request)
        conman = ContractsManager([UrlContract, ReturnsContract, contract])

        request = conman.from_method(spider.returns_request, self.results)
        contract.add_post_hook(request, self.results)
        with pytest.raises(KeyboardInterrupt, match="Post-process exception"):
            request.callback(response, **request.cb_kwargs)

        assert not self.results.failures
        assert not self.results.errors

    def test_pre_hook_success(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PreProcessSuccessContract(spider.returns_request)
        conman = ContractsManager([UrlContract, ReturnsContract, contract])

        request = conman.from_method(spider.returns_request, self.results)
        contract.add_pre_hook(request, self.results)
        request.callback(response, **request.cb_kwargs)

        assert not self.results.failures
        assert not self.results.errors

    def test_pre_hook_assertion_failure(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PreProcessAssertionFailContract(spider.returns_request)
        conman = ContractsManager([UrlContract, ReturnsContract, contract])

        request = conman.from_method(spider.returns_request, self.results)
        contract.add_pre_hook(request, self.results)
        request.callback(response, **request.cb_kwargs)

        assert self.results.failures
        assert not self.results.errors

    def test_pre_hook_error(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PreProcessErrorContract(spider.returns_request)
        conman = ContractsManager([UrlContract, ReturnsContract, contract])

        request = conman.from_method(spider.returns_request, self.results)
        contract.add_pre_hook(request, self.results)
        request.callback(response, **request.cb_kwargs)

        assert self.results.errors

    def test_pre_hook_async_callback(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PreProcessSuccessContract(spider.returns_request_async)
        request = Request("http://scrapy.org", callback=spider.returns_request_async)
        contract.add_pre_hook(request, self.results)

        with pytest.raises(TypeError, match="async callbacks"):
            request.callback(response)

    def test_pre_hook_async_generator(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PreProcessSuccessContract(spider.returns_async_gen)
        request = Request("http://scrapy.org", callback=spider.returns_async_gen)
        contract.add_pre_hook(request, self.results)

        with pytest.raises(TypeError, match="async callbacks"):
            request.callback(response)

    def test_post_hook_async_generator(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PostProcessSuccessContract(spider.returns_async_gen)
        request = Request("http://scrapy.org", callback=spider.returns_async_gen)
        contract.add_post_hook(request, self.results)

        with pytest.raises(TypeError, match="async callbacks"):
            request.callback(response)

    def test_post_hook_error(self):
        spider = DemoSpider()
        response = ResponseMock()
        contract = PostProcessErrorContract(spider.returns_request)
        conman = ContractsManager([UrlContract, ReturnsContract, contract])

        request = conman.from_method(spider.returns_request, self.results)
        contract.add_post_hook(request, self.results)
        request.callback(response, **request.cb_kwargs)

        assert self.results.errors