<s> from __future__ import absolute_import <EOL> import json <EOL> import mock <EOL> import re <EOL> import responses <EOL> from datetime import datetime , timedelta <EOL> from flask import current_app <EOL> from mock import patch <EOL> from changes . config import db <EOL> from changes . constants import Result , Status <EOL> from changes . db . types . filestorage import FileData <EOL> from changes . jobs . sync_job_step import ( <EOL> sync_job_step , is_missing_tests , has_timed_out , <EOL> _SNAPSHOT_TIMEOUT_BONUS_MINUTES , <EOL> _get_artifacts_to_sync , <EOL> _sync_from_artifact_store , <EOL> _sync_artifacts_for_jobstep , <EOL> ) <EOL> from changes . models import ( <EOL> ItemOption , ItemStat , JobStep , HistoricalImmutableStep , Task , FileCoverage , <EOL> TestCase , FailureReason , Artifact , LogSource <EOL> ) <EOL> from changes . testutils import TestCase as BaseTestCase <EOL> class HasTimedOutTest ( BaseTestCase ) : <EOL> def test_simple ( self ) : <EOL> project = self . create_project ( ) <EOL> plan = self . create_plan ( project ) <EOL> step = self . create_step ( plan ) <EOL> option = ItemOption ( <EOL> item_id = step . id , <EOL> name = '<STR_LIT>' , <EOL> value = '<STR_LIT:5>' , <EOL> ) <EOL> db . session . add ( option ) <EOL> db . session . flush ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build , status = Status . queued ) <EOL> jobplan = self . create_job_plan ( job , plan ) <EOL> db . session . commit ( ) <EOL> default_always , default_never = <NUM_LIT:0> , <NUM_LIT> <EOL> jobphase = self . create_jobphase ( job ) <EOL> jobstep = self . create_jobstep ( jobphase ) <EOL> assert not has_timed_out ( jobstep , jobplan , default_always ) <EOL> jobstep . status = Status . allocated <EOL> jobstep . date_created = datetime . utcnow ( ) - timedelta ( minutes = <NUM_LIT:6> ) <EOL> db . session . add ( jobstep ) <EOL> db . session . commit ( ) <EOL> assert has_timed_out ( jobstep , jobplan , default_never ) <EOL> jobstep . status = Status . in_progress <EOL> jobstep . date_started = datetime . utcnow ( ) <EOL> db . session . add ( jobstep ) <EOL> db . session . commit ( ) <EOL> assert not has_timed_out ( jobstep , jobplan , default_always ) <EOL> jobstep . date_started = datetime . utcnow ( ) - timedelta ( minutes = <NUM_LIT:6> ) <EOL> db . session . add ( jobstep ) <EOL> db . session . commit ( ) <EOL> assert has_timed_out ( jobstep , jobplan , default_never ) <EOL> jobstep . status = Status . allocated <EOL> db . session . add ( jobstep ) <EOL> db . session . commit ( ) <EOL> assert has_timed_out ( jobstep , jobplan , default_never ) <EOL> jobplan . data [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ option . name ] = '<STR_LIT:0>' <EOL> db . session . add ( jobplan ) <EOL> db . session . commit ( ) <EOL> assert has_timed_out ( jobstep , jobplan , <NUM_LIT:4> ) <EOL> assert not has_timed_out ( jobstep , jobplan , <NUM_LIT:7> ) <EOL> assert has_timed_out ( jobstep , jobplan , <NUM_LIT:0> ) <EOL> jobplan . data [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ option . name ] = '<STR_LIT>' <EOL> db . session . add ( jobplan ) <EOL> db . session . commit ( ) <EOL> assert not has_timed_out ( jobstep , jobplan , default_always ) <EOL> def test_snapshot ( self ) : <EOL> project = self . create_project ( ) <EOL> plan = self . create_plan ( project ) <EOL> step = self . create_step ( plan ) <EOL> option = ItemOption ( <EOL> item_id = step . id , <EOL> name = '<STR_LIT>' , <EOL> value = '<STR_LIT:5>' , <EOL> ) <EOL> db . session . add ( option ) <EOL> db . session . flush ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build , status = Status . in_progress ) <EOL> jobplan = self . create_job_plan ( job , plan ) <EOL> db . session . commit ( ) <EOL> jobphase = self . create_jobphase ( job ) <EOL> jobstep = self . create_jobstep ( jobphase , <EOL> status = Status . in_progress , <EOL> date_started = datetime . utcnow ( ) - timedelta ( minutes = <NUM_LIT:4> + _SNAPSHOT_TIMEOUT_BONUS_MINUTES ) ) <EOL> with patch ( '<STR_LIT>' , return_value = False ) : <EOL> assert has_timed_out ( jobstep , jobplan , <NUM_LIT:0> ) <EOL> with patch ( '<STR_LIT>' , return_value = True ) : <EOL> assert not has_timed_out ( jobstep , jobplan , <NUM_LIT:0> ) <EOL> class IsMissingTestsTest ( BaseTestCase ) : <EOL> def test_single_phase ( self ) : <EOL> project = self . create_project ( ) <EOL> plan = self . create_plan ( project ) <EOL> option = ItemOption ( <EOL> item_id = plan . id , <EOL> name = '<STR_LIT>' , <EOL> value = '<STR_LIT:0>' , <EOL> ) <EOL> db . session . add ( option ) <EOL> db . session . commit ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> jobplan = self . create_job_plan ( job , plan ) <EOL> jobphase = self . create_jobphase ( <EOL> job = job , <EOL> date_started = datetime ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:15> , <NUM_LIT> ) , <EOL> ) <EOL> jobstep = self . create_jobstep ( jobphase ) <EOL> jobstep2 = self . create_jobstep ( jobphase ) <EOL> assert not is_missing_tests ( jobstep , jobplan ) <EOL> jobplan . data [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ option . name ] = '<STR_LIT:1>' <EOL> db . session . add ( jobplan ) <EOL> db . session . commit ( ) <EOL> assert is_missing_tests ( jobstep , jobplan ) <EOL> testcase = TestCase ( <EOL> project_id = project . id , <EOL> job_id = job . id , <EOL> step_id = jobstep2 . id , <EOL> name = '<STR_LIT:test>' , <EOL> ) <EOL> db . session . add ( testcase ) <EOL> db . session . commit ( ) <EOL> assert is_missing_tests ( jobstep , jobplan ) <EOL> testcase = TestCase ( <EOL> project_id = project . id , <EOL> job_id = job . id , <EOL> step_id = jobstep . id , <EOL> name = '<STR_LIT>' , <EOL> ) <EOL> db . session . add ( testcase ) <EOL> db . session . commit ( ) <EOL> assert not is_missing_tests ( jobstep , jobplan ) <EOL> def test_multi_phase ( self ) : <EOL> project = self . create_project ( ) <EOL> plan = self . create_plan ( project ) <EOL> option = ItemOption ( <EOL> item_id = plan . id , <EOL> name = '<STR_LIT>' , <EOL> value = '<STR_LIT:1>' , <EOL> ) <EOL> db . session . add ( option ) <EOL> db . session . commit ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> jobplan = self . create_job_plan ( job , plan ) <EOL> jobphase = self . create_jobphase ( <EOL> job = job , <EOL> label = '<STR_LIT>' , <EOL> date_created = datetime ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:15> , <NUM_LIT> ) , <EOL> ) <EOL> jobphase2 = self . create_jobphase ( <EOL> job = job , <EOL> label = '<STR_LIT:test>' , <EOL> date_created = datetime ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:16> , <NUM_LIT> ) , <EOL> ) <EOL> jobstep = self . create_jobstep ( jobphase ) <EOL> jobstep2 = self . create_jobstep ( jobphase2 ) <EOL> job2 = self . create_job ( build = build ) <EOL> self . create_job_plan ( job2 , plan ) <EOL> self . create_jobphase ( <EOL> job = job2 , <EOL> label = '<STR_LIT>' , <EOL> date_created = datetime ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) , <EOL> ) <EOL> assert not is_missing_tests ( jobstep , jobplan ) <EOL> assert is_missing_tests ( jobstep2 , jobplan ) <EOL> testcase = TestCase ( <EOL> project_id = project . id , <EOL> job_id = job . id , <EOL> step_id = jobstep . id , <EOL> name = '<STR_LIT:test>' , <EOL> ) <EOL> db . session . add ( testcase ) <EOL> db . session . commit ( ) <EOL> assert not is_missing_tests ( jobstep , jobplan ) <EOL> assert is_missing_tests ( jobstep2 , jobplan ) <EOL> testcase = TestCase ( <EOL> project_id = project . id , <EOL> job_id = job . id , <EOL> step_id = jobstep2 . id , <EOL> name = '<STR_LIT>' , <EOL> ) <EOL> db . session . add ( testcase ) <EOL> db . session . commit ( ) <EOL> assert not is_missing_tests ( jobstep2 , jobplan ) <EOL> def test_snapshot_build ( self ) : <EOL> """<STR_LIT>""" <EOL> project = self . create_project ( ) <EOL> plan = self . create_plan ( project ) <EOL> option = ItemOption ( <EOL> item_id = plan . id , <EOL> name = '<STR_LIT>' , <EOL> value = '<STR_LIT:1>' , <EOL> ) <EOL> db . session . add ( option ) <EOL> db . session . commit ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> jobplan = self . create_job_plan ( job , plan ) <EOL> jobphase = self . create_jobphase ( <EOL> job = job , <EOL> date_started = datetime ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:15> , <NUM_LIT> ) , <EOL> ) <EOL> jobstep = self . create_jobstep ( jobphase ) <EOL> assert is_missing_tests ( jobstep , jobplan ) <EOL> snapshot = self . create_snapshot ( project ) <EOL> snapshot_image = self . create_snapshot_image ( snapshot , plan , job_id = job . id ) <EOL> assert not is_missing_tests ( jobstep , jobplan ) <EOL> class SyncJobStepTest ( BaseTestCase ) : <EOL> ARTIFACTSTORE_REQUEST_RE = re . compile ( r'<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> @ responses . activate <EOL> def test_in_progress ( self , get_implementation , queue_delay ) : <EOL> responses . add ( responses . GET , SyncJobStepTest . ARTIFACTSTORE_REQUEST_RE , body = '<STR_LIT>' , status = <NUM_LIT> ) <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> def mark_in_progress ( step ) : <EOL> step . status = Status . in_progress <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase ) <EOL> task = self . create_task ( <EOL> parent_id = job . id , <EOL> task_id = step . id , <EOL> task_name = '<STR_LIT>' , <EOL> ) <EOL> db . session . add ( ItemStat ( item_id = job . id , name = '<STR_LIT>' , value = <NUM_LIT:1> ) ) <EOL> db . session . commit ( ) <EOL> implementation . update_step . side_effect = mark_in_progress <EOL> sync_job_step ( <EOL> step_id = step . id . hex , <EOL> task_id = step . id . hex , <EOL> parent_task_id = job . id . hex , <EOL> ) <EOL> implementation . update_step . assert_called_once_with ( <EOL> step = step <EOL> ) <EOL> db . session . expire ( step ) <EOL> db . session . expire ( task ) <EOL> step = JobStep . query . get ( step . id ) <EOL> assert step . status == Status . in_progress <EOL> task = Task . query . get ( task . id ) <EOL> assert task . status == Status . in_progress <EOL> queue_delay . assert_any_call ( '<STR_LIT>' , kwargs = { <EOL> '<STR_LIT>' : step . id . hex , <EOL> '<STR_LIT>' : step . id . hex , <EOL> '<STR_LIT>' : job . id . hex , <EOL> } , countdown = <NUM_LIT:5> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> @ responses . activate <EOL> def test_finished ( self , get_implementation , queue_delay ) : <EOL> responses . add ( responses . GET , SyncJobStepTest . ARTIFACTSTORE_REQUEST_RE , body = '<STR_LIT>' , status = <NUM_LIT> ) <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> def mark_finished ( step ) : <EOL> step . status = Status . finished <EOL> step . result = Result . failed <EOL> implementation . update_step . side_effect = mark_finished <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase ) <EOL> task = self . create_task ( <EOL> parent_id = job . id , <EOL> task_id = step . id , <EOL> task_name = '<STR_LIT>' , <EOL> status = Status . finished , <EOL> ) <EOL> db . session . add ( TestCase ( <EOL> name = '<STR_LIT:test>' , <EOL> step_id = step . id , <EOL> job_id = job . id , <EOL> project_id = project . id , <EOL> result = Result . failed , <EOL> ) ) <EOL> db . session . add ( FileCoverage ( <EOL> job = job , step = step , project = job . project , <EOL> filename = '<STR_LIT>' , data = '<STR_LIT>' , <EOL> lines_covered = <NUM_LIT:6> , <EOL> lines_uncovered = <NUM_LIT:5> , <EOL> diff_lines_covered = <NUM_LIT:3> , <EOL> diff_lines_uncovered = <NUM_LIT:2> , <EOL> ) ) <EOL> db . session . commit ( ) <EOL> sync_job_step ( <EOL> step_id = step . id . hex , <EOL> task_id = step . id . hex , <EOL> parent_task_id = job . id . hex , <EOL> ) <EOL> implementation . update_step . assert_called_once_with ( <EOL> step = step <EOL> ) <EOL> db . session . expire ( step ) <EOL> db . session . expire ( task ) <EOL> step = JobStep . query . get ( step . id ) <EOL> assert step . status == Status . finished <EOL> task = Task . query . get ( task . id ) <EOL> assert task . status == Status . finished <EOL> assert len ( queue_delay . mock_calls ) == <NUM_LIT:0> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == step . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:0> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == step . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:6> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == step . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:5> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == step . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:3> <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == step . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:2> <EOL> assert FailureReason . query . filter ( <EOL> FailureReason . step_id == step . id , <EOL> FailureReason . reason == '<STR_LIT>' , <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> @ responses . activate <EOL> def test_missing_test_results_and_expected ( self , get_implementation , queue_delay ) : <EOL> responses . add ( responses . GET , SyncJobStepTest . ARTIFACTSTORE_REQUEST_RE , body = '<STR_LIT>' , status = <NUM_LIT> ) <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> def mark_finished ( step ) : <EOL> step . status = Status . finished <EOL> step . result = Result . passed <EOL> implementation . update_step . side_effect = mark_finished <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> db . session . add ( ItemOption ( <EOL> item_id = plan . id , <EOL> name = '<STR_LIT>' , <EOL> value = '<STR_LIT:1>' <EOL> ) ) <EOL> db . session . commit ( ) <EOL> self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( <EOL> job = job , <EOL> date_started = datetime ( <NUM_LIT> , <NUM_LIT:9> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:15> , <NUM_LIT> ) , <EOL> ) <EOL> step = self . create_jobstep ( phase ) <EOL> with mock . patch . object ( sync_job_step , '<STR_LIT>' , True ) : <EOL> sync_job_step ( <EOL> step_id = step . id . hex , <EOL> task_id = step . id . hex , <EOL> parent_task_id = job . id . hex , <EOL> ) <EOL> db . session . expire ( step ) <EOL> step = JobStep . query . get ( step . id ) <EOL> assert step . status == Status . finished <EOL> assert step . result == Result . failed <EOL> stat = ItemStat . query . filter ( <EOL> ItemStat . item_id == step . id , <EOL> ItemStat . name == '<STR_LIT>' , <EOL> ) . first ( ) <EOL> assert stat . value == <NUM_LIT:1> <EOL> assert FailureReason . query . filter ( <EOL> FailureReason . step_id == step . id , <EOL> FailureReason . reason == '<STR_LIT>' , <EOL> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> @ responses . activate <EOL> def test_timed_out ( self , get_implementation , mock_has_timed_out ) : <EOL> responses . add ( responses . GET , SyncJobStepTest . ARTIFACTSTORE_REQUEST_RE , body = '<STR_LIT>' , status = <NUM_LIT> ) <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> jobplan = self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase , status = Status . in_progress ) <EOL> mock_has_timed_out . return_value = True <EOL> current_app . config [ '<STR_LIT>' ] = <NUM_LIT> <EOL> with mock . patch . object ( sync_job_step , '<STR_LIT>' , True ) : <EOL> sync_job_step ( <EOL> step_id = step . id . hex , <EOL> task_id = step . id . hex , <EOL> parent_task_id = job . id . hex <EOL> ) <EOL> mock_has_timed_out . assert_called_once_with ( step , jobplan , default_timeout = <NUM_LIT> ) <EOL> implementation . cancel_step . assert_called_once_with ( <EOL> step = step , <EOL> ) <EOL> assert step . result == Result . failed <EOL> assert step . status == Status . finished <EOL> assert FailureReason . query . filter ( <EOL> FailureReason . step_id == step . id , <EOL> FailureReason . reason == '<STR_LIT>' , <EOL> ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> @ responses . activate <EOL> def test_failure_reasons ( self , get_implementation ) : <EOL> responses . add ( responses . GET , SyncJobStepTest . ARTIFACTSTORE_REQUEST_RE , body = '<STR_LIT>' , status = <NUM_LIT> ) <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase , status = Status . finished , result = Result . passed ) <EOL> db . session . add ( FailureReason ( <EOL> step_id = step . id , <EOL> job_id = job . id , <EOL> build_id = build . id , <EOL> project_id = project . id , <EOL> reason = '<STR_LIT>' <EOL> ) ) <EOL> db . session . commit ( ) <EOL> with mock . patch . object ( sync_job_step , '<STR_LIT>' , True ) : <EOL> sync_job_step ( <EOL> step_id = step . id . hex , <EOL> task_id = step . id . hex , <EOL> parent_task_id = job . id . hex <EOL> ) <EOL> assert step . result == Result . infra_failed <EOL> @ mock . patch . object ( Artifact , '<STR_LIT:file>' ) <EOL> @ responses . activate <EOL> def test_sync_from_artifact_store ( self , artifact_file ) : <EOL> artifacts = [ { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> responses . add ( responses . GET , SyncJobStepTest . ARTIFACTSTORE_REQUEST_RE , body = json . dumps ( artifacts ) ) <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project ) <EOL> job = self . create_job ( build ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase , status = Status . finished , result = Result . passed ) <EOL> _sync_from_artifact_store ( step ) <EOL> logsources = LogSource . query . filter ( LogSource . step_id == step . id ) . all ( ) <EOL> assert len ( logsources ) == <NUM_LIT:1> <EOL> assert logsources [ <NUM_LIT:0> ] . name == '<STR_LIT>' <EOL> assert logsources [ <NUM_LIT:0> ] . in_artifact_store <EOL> db_artifacts = Artifact . query . filter ( Artifact . step_id == step . id ) . all ( ) <EOL> assert len ( db_artifacts ) == <NUM_LIT:1> <EOL> assert db_artifacts [ <NUM_LIT:0> ] . name == '<STR_LIT>' <EOL> assert artifact_file . save . call_count == <NUM_LIT:1> <EOL> def test_get_artifacts_to_sync ( self ) : <EOL> artifact_manager = mock . Mock ( ) <EOL> artifact_manager . can_process . side_effect = lambda name : not name . endswith ( '<STR_LIT>' ) <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase , status = Status . finished , result = Result . passed ) <EOL> def make_AS_file ( ) : <EOL> return FileData ( { <EOL> '<STR_LIT:filename>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> artstore_junit = self . create_artifact ( step , '<STR_LIT>' , <EOL> file = make_AS_file ( ) ) <EOL> artstore_junit2 = self . create_artifact ( step , '<STR_LIT>' , <EOL> file = make_AS_file ( ) ) <EOL> artstore_coverage = self . create_artifact ( step , '<STR_LIT>' , <EOL> file = make_AS_file ( ) ) <EOL> other_junit = self . create_artifact ( step , '<STR_LIT>' ) <EOL> other_manifest = self . create_artifact ( step , '<STR_LIT>' ) <EOL> ignored_art = self . create_artifact ( step , '<STR_LIT>' ) <EOL> arts = Artifact . query . filter ( Artifact . step_id == step . id ) . all ( ) <EOL> assert ( sorted ( _get_artifacts_to_sync ( arts , artifact_manager , prefer_artifactstore = True ) ) == <EOL> sorted ( [ artstore_junit , artstore_junit2 , artstore_coverage , other_manifest ] ) ) <EOL> assert ( sorted ( _get_artifacts_to_sync ( arts , artifact_manager , prefer_artifactstore = False ) ) == <EOL> sorted ( [ artstore_coverage , other_junit , other_manifest ] ) ) <EOL> @ mock . patch . object ( HistoricalImmutableStep , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_sync_artifacts_for_jobstep ( self , _get_artifacts_to_sync , get_implementation ) : <EOL> implementation = mock . Mock ( ) <EOL> get_implementation . return_value = implementation <EOL> implementation . prefer_artifactstore . return_value = False <EOL> project = self . create_project ( ) <EOL> build = self . create_build ( project = project ) <EOL> job = self . create_job ( build = build ) <EOL> plan = self . create_plan ( project ) <EOL> self . create_step ( plan , implementation = '<STR_LIT:test>' , order = <NUM_LIT:0> ) <EOL> self . create_job_plan ( job , plan ) <EOL> phase = self . create_jobphase ( job ) <EOL> step = self . create_jobstep ( phase , status = Status . finished , result = Result . passed ) <EOL> artifact = self . create_artifact ( step , '<STR_LIT>' ) <EOL> to_sync = [ artifact ] <EOL> _get_artifacts_to_sync . return_value = to_sync <EOL> _sync_artifacts_for_jobstep ( step ) <EOL> assert Task . query . filter ( Task . task_id == artifact . id ) . first ( ) <EOL> implementation . verify_final_artifacts . assert_called_once_with ( step , to_sync ) <EOL> _sync_artifacts_for_jobstep ( step ) <EOL> implementation . verify_final_artifacts . assert_called_once_with ( step , to_sync ) </s>
