<s> from __future__ import absolute_import <EOL> from datetime import datetime <EOL> from flask import current_app <EOL> import mock <EOL> from changes . constants import Status , Result <EOL> from changes . config import db <EOL> from changes . jobs . sync_job import sync_job , _should_retry_jobstep , _find_and_retry_jobsteps <EOL> import changes . jobs . sync_job <EOL> from changes . models import ItemStat , Job , HistoricalImmutableStep , Task <EOL> from changes . testutils import TestCase <EOL> class SyncJobTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SyncJobTest , self ) . setUp ( ) <EOL> self . project = self . create_project ( ) <EOL> self . build = self . create_build ( project = self . project ) <EOL> self . job = self . create_job ( build = self . build ) <EOL> self . jobphase = self . create_jobphase ( self . job ) <EOL> self . jobstep = self . create_jobstep ( self . jobphase ) <EOL> self . task = self . create_task ( <EOL> parent_id = self . build . id , <EOL> task_id = self . job . id , <EOL> task_name = '<STR_LIT>' , <EOL> ) <EOL> self . plan = self . create_plan ( self . project ) <EOL> self . step = self . create_step ( self . plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> self . jobplan = self . create_job_plan ( self . job , self . plan ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> def test_in_progress ( self , get_implementation , <EOL> queue_delay ) : <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> build , job , task = self . build , self . job , self . task <EOL> self . create_task ( <EOL> task_name = '<STR_LIT>' , <EOL> task_id = job . phases [ <NUM_LIT:0> ] . steps [ <NUM_LIT:0> ] . id , <EOL> parent_id = job . id , <EOL> status = Status . in_progress , <EOL> ) <EOL> def mark_in_progress ( job ) : <EOL> job . status = Status . in_progress <EOL> implementation . update . side_effect = mark_in_progress <EOL> sync_job ( <EOL> job_id = job . id . hex , <EOL> task_id = job . id . hex , <EOL> parent_task_id = build . id . hex <EOL> ) <EOL> get_implementation . assert_called_once_with ( ) <EOL> implementation . update . assert_called_once_with ( <EOL> job = self . job , <EOL> ) <EOL> assert implementation . validate_phase . call_count == <NUM_LIT:0> <EOL> assert implementation . validate . call_count == <NUM_LIT:0> <EOL> queue_delay . assert_any_call ( '<STR_LIT>' , kwargs = { <EOL> '<STR_LIT>' : job . id . hex , <EOL> '<STR_LIT>' : job . id . hex , <EOL> '<STR_LIT>' : build . id . hex , <EOL> } , countdown = <NUM_LIT:5> ) <EOL> task = Task . query . get ( task . id ) <EOL> assert task . status == Status . in_progress <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> def test_finished ( self , get_implementation , queue_delay , <EOL> mock_fire_signal ) : <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> assert self . jobplan <EOL> build , job , task = self . build , self . job , self . task <EOL> step = job . phases [ <NUM_LIT:0> ] . steps [ <NUM_LIT:0> ] <EOL> self . create_task ( <EOL> task_name = '<STR_LIT>' , <EOL> task_id = step . id , <EOL> parent_id = job . id , <EOL> status = Status . finished , <EOL> ) <EOL> self . create_test ( job ) <EOL> self . create_test ( job ) <EOL> step2 = self . create_jobstep ( self . jobphase , status = Status . finished , <EOL> replacement_id = step . id ) <EOL> self . jobstep . status = Status . finished <EOL> db . session . add ( self . jobstep ) <EOL> db . session . add ( ItemStat ( item_id = step . id , name = '<STR_LIT>' , value = <NUM_LIT:1> ) ) <EOL> db . session . add ( ItemStat ( item_id = step . id , name = '<STR_LIT>' , value = <NUM_LIT:10> ) ) <EOL> db . session . add ( ItemStat ( item_id = step . id , name = '<STR_LIT>' , value = <NUM_LIT> ) ) <EOL> db . session . add ( ItemStat ( item_id = step2 . id , name = '<STR_LIT>' , value = <NUM_LIT:10> ) ) <EOL> db . session . commit ( ) <EOL> sync_job ( <EOL> job_id = job . id . hex , <EOL> task_id = job . id . hex , <EOL> parent_task_id = build . id . hex , <EOL> ) <EOL> implementation . validate_phase . assert_called_once_with ( phase = self . job . phases [ <NUM_LIT:0> ] ) <EOL> implementation . validate . assert_called_once_with ( job = self . job ) <EOL> job = Job . query . get ( job . id ) <EOL> assert job . status == Status . finished <EOL> queue_delay . assert_any_call ( '<STR_LIT>' , kwargs = { <EOL> '<STR_LIT>' : self . project . id . hex , <EOL> '<STR_LIT>' : self . plan . id . hex , <EOL> } , countdown = <NUM_LIT:1> ) <EOL> mock_fire_signal . delay . assert_any_call ( <EOL> signal = '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : job . id . hex } , <EOL> ) <EOL> task = Task . query . get ( task . id ) <EOL> assert task . status == Status . finished <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == job . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:1> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == job . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:10> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == job . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT> <EOL> def test_should_retry_jobstep ( self ) : <EOL> self . jobstep . result = Result . infra_failed <EOL> assert _should_retry_jobstep ( self . jobstep ) <EOL> self . jobstep . result = Result . failed <EOL> assert not _should_retry_jobstep ( self . jobstep ) <EOL> self . jobstep . result = Result . infra_failed <EOL> self . jobstep . replacement_id = <NUM_LIT:0> <EOL> assert not _should_retry_jobstep ( self . jobstep ) <EOL> self . jobstep . replacement_id = None <EOL> ts = <NUM_LIT> <EOL> changes . jobs . sync_job . MAX_DURATION_FOR_RETRY_SECS = <NUM_LIT:10> <EOL> self . jobstep . date_started = datetime . utcfromtimestamp ( ts ) <EOL> self . jobstep . date_finished = datetime . utcfromtimestamp ( ts + <NUM_LIT:11> ) <EOL> assert not _should_retry_jobstep ( self . jobstep ) <EOL> self . jobstep . date_started = self . jobstep . date_finished = None <EOL> self . create_jobstep ( self . jobphase , status = Status . finished , <EOL> replacement_id = self . jobstep . id ) <EOL> assert not _should_retry_jobstep ( self . jobstep ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_find_and_retry_jobsteps ( self , should_retry_jobstep ) : <EOL> should_retry_jobstep . return_value = True <EOL> implementation = mock . Mock ( ) <EOL> phase = self . job . phases [ <NUM_LIT:0> ] <EOL> _find_and_retry_jobsteps ( phase , implementation ) <EOL> implementation . create_replacement_jobstep . assert_called_once_with ( self . jobstep ) <EOL> implementation . reset_mock ( ) <EOL> current_app . config [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> _find_and_retry_jobsteps ( phase , implementation ) <EOL> implementation . create_replacement_jobstep . assert_not_called ( ) <EOL> current_app . config [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> self . create_jobstep ( phase , status = Status . finished , <EOL> replacement_id = self . jobstep . id ) <EOL> _find_and_retry_jobsteps ( phase , implementation ) <EOL> implementation . create_replacement_jobstep . assert_not_called ( ) </s>
