<s> import httplib <EOL> import unittest <EOL> from mapreduce import base_handler <EOL> from mapreduce import errors <EOL> from mapreduce import parameters <EOL> from mapreduce import util <EOL> from google . appengine . ext . webapp import mock_webapp <EOL> try : <EOL> from cloudstorage import api_utils <EOL> enable_cloudstorage_tests = True <EOL> except ImportError : <EOL> enable_cloudstorage_tests = False <EOL> class TaskQueueHandlerTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . handler = base_handler . TaskQueueHandler ( ) <EOL> self . request = mock_webapp . MockRequest ( ) <EOL> self . request . headers [ "<STR_LIT>" ] = "<STR_LIT:default>" <EOL> self . request . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> def init ( self ) : <EOL> self . handler . initialize ( self . request , <EOL> mock_webapp . MockResponse ( ) ) <EOL> def testDefaultRetryParams ( self ) : <EOL> if not enable_cloudstorage_tests : <EOL> return <EOL> self . assertTrue ( api_utils . _get_default_retry_params ( ) . save_access_token ) <EOL> self . assertEquals ( "<STR_LIT>" , <EOL> api_utils . _get_default_retry_params ( ) . _user_agent ) <EOL> def testPostNoTaskQueueHeader ( self ) : <EOL> """<STR_LIT>""" <EOL> del self . request . headers [ "<STR_LIT>" ] <EOL> self . init ( ) <EOL> self . handler . post ( ) <EOL> self . assertEquals ( httplib . FORBIDDEN , self . handler . response . status ) <EOL> def testTaskRetryCount ( self ) : <EOL> self . init ( ) <EOL> self . assertEquals ( <NUM_LIT:0> , self . handler . task_retry_count ( ) ) <EOL> self . request . headers [ "<STR_LIT>" ] = <NUM_LIT:5> <EOL> self . init ( ) <EOL> self . assertEquals ( <NUM_LIT:5> , self . handler . task_retry_count ( ) ) <EOL> class FaultyTaskQueueHandler ( base_handler . TaskQueueHandler ) : <EOL> """<STR_LIT>""" <EOL> dropped = False <EOL> handled = False <EOL> def _preprocess ( self ) : <EOL> raise Exception ( ) <EOL> def _drop_gracefully ( self ) : <EOL> self . dropped = True <EOL> def handle ( self ) : <EOL> self . handled = True <EOL> @ classmethod <EOL> def reset ( cls ) : <EOL> cls . dropped = False <EOL> cls . handled = False <EOL> class FaultyTaskQueueHandlerTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> FaultyTaskQueueHandler . reset ( ) <EOL> self . handler = FaultyTaskQueueHandler ( ) <EOL> self . request = mock_webapp . MockRequest ( ) <EOL> self . request . headers [ "<STR_LIT>" ] = "<STR_LIT:default>" <EOL> self . request . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> self . request . headers [ util . _MR_ID_TASK_HEADER ] = "<STR_LIT>" <EOL> def init ( self ) : <EOL> self . handler . initialize ( self . request , <EOL> mock_webapp . MockResponse ( ) ) <EOL> def testSmoke ( self ) : <EOL> self . init ( ) <EOL> self . handler . post ( ) <EOL> self . assertTrue ( self . handler . dropped ) <EOL> self . assertFalse ( self . handler . handled ) <EOL> self . assertEqual ( httplib . OK , self . handler . response . status ) <EOL> def testTaskRetriedTooManyTimes ( self ) : <EOL> self . request . headers [ "<STR_LIT>" ] = ( <EOL> parameters . config . TASK_MAX_ATTEMPTS ) <EOL> self . init ( ) <EOL> self . handler . post ( ) <EOL> self . assertTrue ( self . handler . dropped ) <EOL> self . assertFalse ( self . handler . handled ) <EOL> self . assertEqual ( httplib . OK , self . handler . response . status ) <EOL> class JsonErrorHandler ( base_handler . JsonHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , error ) : <EOL> """<STR_LIT>""" <EOL> self . error = error <EOL> self . json_response = { } <EOL> def handle ( self ) : <EOL> """<STR_LIT>""" <EOL> raise self . error <EOL> class JsonHandlerTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . handler = base_handler . JsonHandler ( ) <EOL> self . handler . initialize ( mock_webapp . MockRequest ( ) , <EOL> mock_webapp . MockResponse ( ) ) <EOL> def testBasePath ( self ) : <EOL> """<STR_LIT>""" <EOL> self . handler . request . path = "<STR_LIT>" <EOL> self . assertRaises ( base_handler . BadRequestPathError , <EOL> self . handler . base_path ) <EOL> self . handler . request . path = "<STR_LIT>" <EOL> self . assertEquals ( "<STR_LIT>" , self . handler . base_path ( ) ) <EOL> self . handler . request . path = "<STR_LIT>" <EOL> self . assertEquals ( "<STR_LIT>" , self . handler . base_path ( ) ) <EOL> self . handler . request . path = "<STR_LIT>" <EOL> self . assertEquals ( "<STR_LIT>" , self . handler . base_path ( ) ) <EOL> def testMissingYamlError ( self ) : <EOL> """<STR_LIT>""" <EOL> handler = JsonErrorHandler ( errors . MissingYamlError ) <EOL> request = mock_webapp . MockRequest ( ) <EOL> response = mock_webapp . MockResponse ( ) <EOL> request . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> handler . initialize ( request , response ) <EOL> handler . _handle_wrapper ( ) <EOL> self . assertEquals ( "<STR_LIT>" , handler . json_response [ "<STR_LIT>" ] ) <EOL> self . assertEquals ( "<STR_LIT>" , <EOL> handler . json_response [ "<STR_LIT>" ] ) <EOL> def testError ( self ) : <EOL> """<STR_LIT>""" <EOL> handler = JsonErrorHandler ( Exception ( "<STR_LIT>" ) ) <EOL> request = mock_webapp . MockRequest ( ) <EOL> response = mock_webapp . MockResponse ( ) <EOL> request . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> handler . initialize ( request , response ) <EOL> handler . _handle_wrapper ( ) <EOL> self . assertEquals ( "<STR_LIT>" , handler . json_response [ "<STR_LIT>" ] ) <EOL> self . assertEquals ( "<STR_LIT>" , <EOL> handler . json_response [ "<STR_LIT>" ] ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
