<s> import unittest <EOL> from google . appengine . api import taskqueue <EOL> from google . appengine . ext import testbed <EOL> from mock import Mock <EOL> from mock import patch <EOL> from furious . extras . insert_task_handlers import insert_tasks_ignore_duplicate_names <EOL> class TestInsertTasksIgnoreDuplicateNames ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> harness = testbed . Testbed ( ) <EOL> harness . activate ( ) <EOL> harness . init_taskqueue_stub ( ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_insert_tasks_ignore_duplicate_names_relays_parms ( self , <EOL> mock_tasks ) : <EOL> """<STR_LIT>""" <EOL> insert_tasks_ignore_duplicate_names ( [ '<STR_LIT:1>' ] , '<STR_LIT>' , '<STR_LIT>' , foo = '<STR_LIT:bar>' ) <EOL> mock_tasks . assert_called_once_with ( [ '<STR_LIT:1>' ] , '<STR_LIT>' , '<STR_LIT>' , foo = '<STR_LIT:bar>' ) <EOL> @ patch ( '<STR_LIT>' , auto_spec = True ) <EOL> def test_insert_tasks_ignore_duplicate_names_raises ( self , mock_add ) : <EOL> """<STR_LIT>""" <EOL> mock_add . side_effect = ( taskqueue . DuplicateTaskNameError , <EOL> taskqueue . TaskAlreadyExistsError , <EOL> taskqueue . TaskAlreadyExistsError , <EOL> taskqueue . TaskAlreadyExistsError ) <EOL> tasks = [ taskqueue . Task ( '<STR_LIT:1>' ) , taskqueue . Task ( '<STR_LIT:2>' ) , taskqueue . Task ( '<STR_LIT:3>' ) ] <EOL> inserted = insert_tasks_ignore_duplicate_names ( tasks , '<STR_LIT>' ) <EOL> self . assertEqual ( inserted , <NUM_LIT:0> ) <EOL> self . assertEqual ( mock_add . call_count , <NUM_LIT:4> ) <EOL> mock_add . assert_any_call ( tasks , transactional = False ) <EOL> for task in tasks : <EOL> mock_add . assert_any_call ( [ task ] , transactional = False ) <EOL> @ patch ( '<STR_LIT>' , auto_spec = True ) <EOL> def test_insert_tasks_ignore_duplicate_names_partial_errors ( self , mock_add ) : <EOL> """<STR_LIT>""" <EOL> mock_add . side_effect = ( taskqueue . DuplicateTaskNameError , <EOL> None , <EOL> taskqueue . TaskAlreadyExistsError , <EOL> None ) <EOL> tasks = [ taskqueue . Task ( '<STR_LIT:1>' ) , taskqueue . Task ( '<STR_LIT:2>' ) , taskqueue . Task ( '<STR_LIT:3>' ) ] <EOL> inserted = insert_tasks_ignore_duplicate_names ( tasks , '<STR_LIT>' ) <EOL> self . assertEqual ( inserted , <NUM_LIT:2> ) <EOL> self . assertEqual ( mock_add . call_count , <NUM_LIT:4> ) <EOL> mock_add . assert_any_call ( tasks , transactional = False ) <EOL> for task in tasks : <EOL> mock_add . assert_any_call ( [ task ] , transactional = False ) <EOL> @ patch ( '<STR_LIT>' , auto_spec = True ) <EOL> def test_insert_tasks_ignore_duplicate_names_was_enqueued ( self , mock_add ) : <EOL> """<STR_LIT>""" <EOL> mock_add . side_effect = ( taskqueue . DuplicateTaskNameError , <EOL> taskqueue . TaskAlreadyExistsError , <EOL> None ) <EOL> tasks = [ Mock ( ) , taskqueue . Task ( '<STR_LIT:2>' ) , taskqueue . Task ( '<STR_LIT:3>' ) ] <EOL> tasks [ <NUM_LIT:0> ] . was_enqueued = True <EOL> inserted = insert_tasks_ignore_duplicate_names ( tasks , '<STR_LIT>' ) <EOL> self . assertEqual ( inserted , <NUM_LIT:2> ) <EOL> self . assertEqual ( mock_add . call_count , <NUM_LIT:3> ) <EOL> mock_add . assert_any_call ( tasks , transactional = False ) <EOL> mock_add . assert_any_call ( [ tasks [ <NUM_LIT:1> ] ] , transactional = False ) <EOL> mock_add . assert_any_call ( [ tasks [ <NUM_LIT:2> ] ] , transactional = False ) </s>
