<s> import unittest <EOL> from twisted . internet import defer <EOL> from twisted . internet . error import TimeoutError , DNSLookupError , ConnectionRefusedError , ConnectionDone , ConnectError , ConnectionLost , TCPTimedOutError <EOL> from scrapy import twisted_version <EOL> from scrapy . downloadermiddlewares . retry import RetryMiddleware <EOL> from scrapy . xlib . tx import ResponseFailed <EOL> from scrapy . spiders import Spider <EOL> from scrapy . http import Request , Response <EOL> from scrapy . utils . test import get_crawler <EOL> class RetryTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> crawler = get_crawler ( Spider ) <EOL> self . spider = crawler . _create_spider ( '<STR_LIT:foo>' ) <EOL> self . mw = RetryMiddleware . from_crawler ( crawler ) <EOL> self . mw . max_retry_times = <NUM_LIT:2> <EOL> def test_priority_adjust ( self ) : <EOL> req = Request ( '<STR_LIT>' ) <EOL> rsp = Response ( '<STR_LIT>' , body = b'<STR_LIT>' , status = <NUM_LIT> ) <EOL> req2 = self . mw . process_response ( req , rsp , self . spider ) <EOL> assert req2 . priority < req . priority <EOL> def test_404 ( self ) : <EOL> req = Request ( '<STR_LIT>' ) <EOL> rsp = Response ( '<STR_LIT>' , body = b'<STR_LIT>' , status = <NUM_LIT> ) <EOL> assert self . mw . process_response ( req , rsp , self . spider ) is rsp <EOL> def test_dont_retry ( self ) : <EOL> req = Request ( '<STR_LIT>' , meta = { '<STR_LIT>' : True } ) <EOL> rsp = Response ( '<STR_LIT>' , body = b'<STR_LIT>' , status = <NUM_LIT> ) <EOL> r = self . mw . process_response ( req , rsp , self . spider ) <EOL> assert r is rsp <EOL> req = Request ( '<STR_LIT>' , meta = { '<STR_LIT>' : False } ) <EOL> rsp = Response ( '<STR_LIT>' ) <EOL> r = self . mw . process_response ( req , rsp , self . spider ) <EOL> assert r is rsp <EOL> def test_dont_retry_exc ( self ) : <EOL> req = Request ( '<STR_LIT>' , meta = { '<STR_LIT>' : True } ) <EOL> r = self . mw . process_exception ( req , DNSLookupError ( ) , self . spider ) <EOL> assert r is None <EOL> def test_503 ( self ) : <EOL> req = Request ( '<STR_LIT>' ) <EOL> rsp = Response ( '<STR_LIT>' , body = b'<STR_LIT>' , status = <NUM_LIT> ) <EOL> req = self . mw . process_response ( req , rsp , self . spider ) <EOL> assert isinstance ( req , Request ) <EOL> self . assertEqual ( req . meta [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> req = self . mw . process_response ( req , rsp , self . spider ) <EOL> assert isinstance ( req , Request ) <EOL> self . assertEqual ( req . meta [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> assert self . mw . process_response ( req , rsp , self . spider ) is rsp <EOL> def test_twistederrors ( self ) : <EOL> exceptions = [ defer . TimeoutError , TCPTimedOutError , TimeoutError , <EOL> DNSLookupError , ConnectionRefusedError , ConnectionDone , <EOL> ConnectError , ConnectionLost ] <EOL> if twisted_version >= ( <NUM_LIT:11> , <NUM_LIT:1> , <NUM_LIT:0> ) : <EOL> exceptions . append ( ResponseFailed ) <EOL> for exc in exceptions : <EOL> req = Request ( '<STR_LIT>' % exc . __name__ ) <EOL> self . _test_retry_exception ( req , exc ( '<STR_LIT:foo>' ) ) <EOL> def _test_retry_exception ( self , req , exception ) : <EOL> req = self . mw . process_exception ( req , exception , self . spider ) <EOL> assert isinstance ( req , Request ) <EOL> self . assertEqual ( req . meta [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> req = self . mw . process_exception ( req , exception , self . spider ) <EOL> assert isinstance ( req , Request ) <EOL> self . assertEqual ( req . meta [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> req = self . mw . process_exception ( req , exception , self . spider ) <EOL> self . assertEqual ( req , None ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
