import unittest

from testfixtures import LogCapture
from twisted.internet import defer
from twisted.python.failure import Failure
from twisted.trial.unittest import TestCase as TwistedTestCase

from scrapy.exceptions import DropItem
from scrapy.utils.test import get_crawler
from scrapy.http import Request, Response
from scrapy.item import Item, Field
from scrapy.logformatter import LogFormatter
from scrapy.spiders import Spider
from tests.mockserver import MockServer
from tests.spiders import ItemSpider


class CustomItem(Item):

    name = Field()

    def __str__(self):
        return f"name: {self['name']}"


class LogFormatterTestCase(unittest.TestCase):
    def setUp(self):
        self.formatter = LogFormatter()
        self.spider = Spider("default")

    def test_crawled_with_referer(self):
        req = Request("http://www.example.com")
        res = Response("http://www.example.com")
        logkws = self.formatter.crawled(req, res, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline, "Crawled (200) <GET http://www.example.com> (referer: None)"
        )

    def test_crawled_without_referer(self):
        req = Request(
            "http://www.example.com", headers={"referer": "http://example.com"}
        )
        res = Response("http://www.example.com", flags=["cached"])
        logkws = self.formatter.crawled(req, res, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline,
            "Crawled (200) <GET http://www.example.com> (referer: http://example.com) ['cached']",
        )

    def test_flags_in_request(self):
        req = Request("http://www.example.com", flags=["test", "flag"])
        res = Response("http://www.example.com")
        logkws = self.formatter.crawled(req, res, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline,
            "Crawled (200) <GET http://www.example.com> ['test', 'flag'] (referer: None)",
        )

    def test_dropped(self):
        item = {}
        exception = Exception("\u2018")
        response = Response("http://www.example.com")
        logkws = self.formatter.dropped(item, exception, response, self.spider)
        logline = logkws["msg"] % logkws["args"]
        lines = logline.splitlines()
        assert all(isinstance(x, str) for x in lines)
        self.assertEqual(lines, ["Dropped: \u2018", "{}"])

    def test_item_error(self):
        # In practice, the complete traceback is shown by passing the
        # 'exc_info' argument to the logging function
        item = {"key": "value"}
        exception = Exception()
        response = Response("http://www.example.com")
        logkws = self.formatter.item_error(item, exception, response, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(logline, "Error processing {'key': 'value'}")

    def test_spider_error(self):
        # In practice, the complete traceback is shown by passing the
        # 'exc_info' argument to the logging function
        failure = Failure(Exception())
        request = Request(
            "http://www.example.com", headers={"Referer": "http://example.org"}
        )
        response = Response("http://www.example.com", request=request)
        logkws = self.formatter.spider_error(failure, request, response, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline,
            "Spider error processing <GET http://www.example.com> (referer: http://example.org)",
        )

    def test_download_error_short(self):
        # In practice, the complete traceback is shown by passing the
        # 'exc_info' argument to the logging function
        failure = Failure(Exception())
        request = Request("http://www.example.com")
        logkws = self.formatter.download_error(failure, request, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(logline, "Error downloading <GET http://www.example.com>")

    def test_download_error_long(self):
        # In practice, the complete traceback is shown by passing the
        # 'exc_info' argument to the logging function
        failure = Failure(Exception())
        request = Request("http://www.example.com")
        logkws = self.formatter.download_error(
            failure, request, self.spider, "Some message"
        )
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline, "Error downloading <GET http://www.example.com>: Some message"
        )

    def test_scraped(self):
        item = CustomItem()
        item["name"] = "\xa3"
        response = Response("http://www.example.com")
        logkws = self.formatter.scraped(item, response, self.spider)
        logline = logkws["msg"] % logkws["args"]
        lines = logline.splitlines()
        assert all(isinstance(x, str) for x in lines)
        self.assertEqual(
            lines, ["Scraped from <200 http://www.example.com>", "name: \xa3"]
        )


class LogFormatterSubclass(LogFormatter):
    def crawled(self, request, response, spider):
        kwargs = super().crawled(request, response, spider)
        CRAWLEDMSG = "Crawled (%(status)s) %(request)s (referer: %(referer)s) %(flags)s"
        log_args = kwargs["args"]
        log_args["flags"] = str(request.flags)
        return {
            "level": kwargs["level"],
            "msg": CRAWLEDMSG,
            "args": log_args,
        }


class LogformatterSubclassTest(LogFormatterTestCase):
    def setUp(self):
        self.formatter = LogFormatterSubclass()
        self.spider = Spider("default")

    def test_crawled_with_referer(self):
        req = Request("http://www.example.com")
        res = Response("http://www.example.com")
        logkws = self.formatter.crawled(req, res, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline, "Crawled (200) <GET http://www.example.com> (referer: None) []"
        )

    def test_crawled_without_referer(self):
        req = Request(
            "http://www.example.com",
            headers={"referer": "http://example.com"},
            flags=["cached"],
        )
        res = Response("http://www.example.com")
        logkws = self.formatter.crawled(req, res, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline,
            "Crawled (200) <GET http://www.example.com> (referer: http://example.com) ['cached']",
        )

    def test_flags_in_request(self):
        req = Request("http://www.example.com", flags=["test", "flag"])
        res = Response("http://www.example.com")
        logkws = self.formatter.crawled(req, res, self.spider)
        logline = logkws["msg"] % logkws["args"]
        self.assertEqual(
            logline,
            "Crawled (200) <GET http://www.example.com> (referer: None) ['test', 'flag']",
        )


class SkipMessagesLogFormatter(LogFormatter):
    def crawled(self, *args, **kwargs):
        return None

    def scraped(self, *args, **kwargs):
        return None

    def dropped(self, *args, **kwargs):
        return None


class DropSomeItemsPipeline:
    drop = True

    def process_item(self, item, spider):
        if self.drop:
            self.drop = False
            raise DropItem("Ignoring item")
        else:
            self.drop = True


class ShowOrSkipMessagesTestCase(TwistedTestCase):
    def setUp(self):
        self.mockserver = MockServer()
        self.mockserver.__enter__()
        self.base_settings = {
            "LOG_LEVEL": "DEBUG",
            "ITEM_PIPELINES": {
                DropSomeItemsPipeline: 300,
            },
        }

    def tearDown(self):
        self.mockserver.__exit__(None, None, None)

    @defer.inlineCallbacks
    def test_show_messages(self):
        crawler = get_crawler(ItemSpider, self.base_settings)
        with LogCapture() as lc:
            yield crawler.crawl(mockserver=self.mockserver)
        self.assertIn("Scraped from <200 http://127.0.0.1:", str(lc))
        self.assertIn("Crawled (200) <GET http://127.0.0.1:", str(lc))
        self.assertIn("Dropped: Ignoring item", str(lc))

    @defer.inlineCallbacks
    def test_skip_messages(self):
        settings = self.base_settings.copy()
        settings["LOG_FORMATTER"] = SkipMessagesLogFormatter
        crawler = get_crawler(ItemSpider, settings)
        with LogCapture() as lc:
            yield crawler.crawl(mockserver=self.mockserver)
        self.assertNotIn("Scraped from <200 http://127.0.0.1:", str(lc))
        self.assertNotIn("Crawled (200) <GET http://127.0.0.1:", str(lc))
        self.assertNotIn("Dropped: Ignoring item", str(lc))


if __name__ == "__main__":
    unittest.main()