Add Spec for ProcessPipelineService

app/models/ci/pipeline.rb

@@ -80,7 +80,7 @@ module Ci
 
     state_machine :status, initial: :created do
       event :enqueue do
-        transition [:created, :skipped] => :pending
+        transition [:created, :skipped, :scheduled] => :pending
         transition [:success, :failed, :canceled] => :running
       end
 

app/services/ci/process_build_service.rb

@@ -2,7 +2,8 @@
 
 module Ci
   class ProcessBuildService < BaseService
-    def execute(build)
+    def execute(build, current_status)
+      if valid_statuses_for_when(build.when).include?(current_status)
         if build.schedulable?
           build.schedule!
         elsif build.action?
@@ -10,6 +11,30 @@ module Ci
         else
           build.enqueue
         end
+        true
+      else
+        build.skip
+        false
+      end
+    end
+
+    private
+
+    def valid_statuses_for_when(value)
+      case value
+      when 'on_success'
+        %w[success skipped]
+      when 'on_failure'
+        %w[failed]
+      when 'always'
+        %w[success failed skipped]
+      when 'manual'
+        %w[success skipped]
+      when 'delayed'
+        %w[success skipped]
+      else
+        []
+      end
     end
   end
 end

app/services/ci/process_pipeline_service.rb

@@ -29,39 +29,13 @@ module Ci
       if HasStatus::COMPLETED_STATUSES.include?(current_status)
         created_builds_in_stage(index).select do |build|
           Gitlab::OptimisticLocking.retry_lock(build) do |subject|
-            process_build(subject, current_status)
+            Ci::ProcessBuildService.new(project, @user)
+              .execute(build, current_status)
           end
         end
       end
     end
 
-    def process_build(build, current_status)
-      if valid_statuses_for_when(build.when).include?(current_status)
-        Ci::ProcessBuildService.new(project, @user).execute(build)
-        true
-      else
-        build.skip
-        false
-      end
-    end
-
-    def valid_statuses_for_when(value)
-      case value
-      when 'on_success'
-        %w[success skipped]
-      when 'on_failure'
-        %w[failed]
-      when 'always'
-        %w[success failed skipped]
-      when 'manual'
-        %w[success skipped]
-      when 'delayed'
-        %w[success skipped]
-      else
-        []
-      end
-    end
-
     # rubocop: disable CodeReuse/ActiveRecord
     def status_for_prior_stages(index)
       pipeline.builds.where('stage_idx < ?', index).latest.status || 'success'

spec/services/ci/process_pipeline_service_spec.rb

@@ -242,6 +242,187 @@ describe Ci::ProcessPipelineService, '#execute' do
     end
   end
 
+  context 'when delayed jobs are defined' do
+    context 'when the scene is timed incremental rollout' do
+      before do
+        create_build('build', stage_idx: 0)
+        create_build('rollout10%', **delayed_options, stage_idx: 1)
+        create_build('rollout100%', **delayed_options, stage_idx: 2)
+        create_build('cleanup', stage_idx: 3)
+
+        allow(Ci::BuildScheduleWorker).to receive(:perform_at)
+      end
+
+      context 'when builds are successful' do
+        it 'properly processes the pipeline' do
+          expect(process_pipeline).to be_truthy
+          expect(builds_names_and_statuses).to eq({ 'build': 'pending' })
+  
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'scheduled' })
+
+          enqueue_scheduled('rollout10%')
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'success', 'rollout100%': 'scheduled' })
+
+          enqueue_scheduled('rollout100%')
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'success', 'rollout100%': 'success', 'cleanup': 'pending' })
+
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'success', 'rollout100%': 'success', 'cleanup': 'success' })
+          expect(pipeline.reload.status).to eq 'success'
+        end
+      end
+
+      context 'when build job fails' do
+        it 'properly processes the pipeline' do
+          expect(process_pipeline).to be_truthy
+          expect(builds_names_and_statuses).to eq({ 'build': 'pending' })
+  
+          fail_running_or_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'failed' })
+          expect(pipeline.reload.status).to eq 'failed'
+        end
+      end
+
+      context 'when rollout 10% is unscheduled' do
+        it 'properly processes the pipeline' do
+          expect(process_pipeline).to be_truthy
+          expect(builds_names_and_statuses).to eq({ 'build': 'pending' })
+  
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'scheduled' })
+  
+          unschedule
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'manual' })
+          expect(pipeline.reload.status).to eq 'manual'
+        end
+
+        context 'when user plays rollout 10%' do
+          it 'schedules rollout100%' do
+            process_pipeline
+            succeed_pending
+            unschedule
+            play_manual_action('rollout10%')
+            succeed_pending
+
+            expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'success', 'rollout100%': 'scheduled' })
+            expect(pipeline.reload.status).to eq 'scheduled'
+          end
+        end
+      end
+
+      context 'when rollout 10% fails' do
+        it 'properly processes the pipeline' do
+          expect(process_pipeline).to be_truthy
+          expect(builds_names_and_statuses).to eq({ 'build': 'pending' })
+  
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'scheduled' })
+  
+          enqueue_scheduled('rollout10%')
+          fail_running_or_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'failed' })
+          expect(pipeline.reload.status).to eq 'failed'
+        end
+
+        context 'when user retries rollout 10%' do
+          it 'does not schedule rollout10% again' do
+            process_pipeline
+            succeed_pending
+            enqueue_scheduled('rollout10%')
+            fail_running_or_pending
+            retry_build('rollout10%')
+
+            expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'pending' })
+            expect(pipeline.reload.status).to eq 'running'
+          end
+        end
+      end
+
+      context 'when rollout 10% is played immidiately' do
+        it 'properly processes the pipeline' do
+          expect(process_pipeline).to be_truthy
+          expect(builds_names_and_statuses).to eq({ 'build': 'pending' })
+  
+          succeed_pending
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'scheduled' })
+  
+          play_manual_action('rollout10%')
+
+          expect(builds_names_and_statuses).to eq({ 'build': 'success', 'rollout10%': 'pending' })
+          expect(pipeline.reload.status).to eq 'running'
+        end
+      end
+    end
+
+    context 'when only one scheduled job exists in a pipeline' do
+      before do
+        create_build('delayed', **delayed_options, stage_idx: 0)
+
+        allow(Ci::BuildScheduleWorker).to receive(:perform_at)
+      end
+
+      it 'properly processes the pipeline' do
+        expect(process_pipeline).to be_truthy
+        expect(builds_names_and_statuses).to eq({ 'delayed': 'scheduled' })
+
+        expect(pipeline.reload.status).to eq 'scheduled'
+      end
+    end
+
+    context 'when there are two delayed jobs in a stage' do
+      before do
+        create_build('delayed1', **delayed_options, stage_idx: 0)
+        create_build('delayed2', **delayed_options, stage_idx: 0)
+        create_build('job', stage_idx: 1)
+
+        allow(Ci::BuildScheduleWorker).to receive(:perform_at)
+      end
+
+      it 'blocks the stage until all scheduled jobs finished' do
+        expect(process_pipeline).to be_truthy
+        expect(builds_names_and_statuses).to eq({ 'delayed1': 'scheduled', 'delayed2': 'scheduled' })
+
+        enqueue_scheduled('delayed1')
+
+        expect(builds_names_and_statuses).to eq({ 'delayed1': 'pending', 'delayed2': 'scheduled' })
+        expect(pipeline.reload.status).to eq 'scheduled'
+      end
+    end
+
+    context 'when a delayed job is allowed to fail' do
+      before do
+        create_build('delayed', **delayed_options, allow_failure: true, stage_idx: 0)
+        create_build('job', stage_idx: 1)
+
+        allow(Ci::BuildScheduleWorker).to receive(:perform_at)
+      end
+
+      it 'blocks the stage and continues after it failed' do
+        expect(process_pipeline).to be_truthy
+        expect(builds_names_and_statuses).to eq({ 'delayed': 'scheduled' })
+
+        enqueue_scheduled('delayed')
+        fail_running_or_pending
+
+        expect(builds_names_and_statuses).to eq({ 'delayed': 'failed', 'job': 'pending' })
+        expect(pipeline.reload.status).to eq 'pending'
+      end
+    end
+  end
+
   context 'when there are manual action in earlier stages' do
     context 'when first stage has only optional manual actions' do
       before do
@@ -536,6 +717,10 @@ describe Ci::ProcessPipelineService, '#execute' do
     builds.pluck(:name)
   end
 
+  def builds_names_and_statuses
+    builds.inject({}) { |h, b| h[b.name.to_sym] = b.status; h }
+  end
+
   def all_builds_names
     all_builds.pluck(:name)
   end
@@ -549,7 +734,7 @@ describe Ci::ProcessPipelineService, '#execute' do
   end
 
   def succeed_pending
-    builds.pending.update_all(status: 'success')
+    builds.pending.map(&:success)
   end
 
   def succeed_running_or_pending
@@ -568,6 +753,14 @@ describe Ci::ProcessPipelineService, '#execute' do
     builds.find_by(name: name).play(user)
   end
 
+  def enqueue_scheduled(name)
+    builds.scheduled.find_by(name: name).enqueue
+  end
+
+  def retry_build(name)
+    Ci::Build.retry(builds.find_by(name: name), user)
+  end
+
   def manual_actions
     pipeline.manual_actions(true)
   end
@@ -575,4 +768,12 @@ describe Ci::ProcessPipelineService, '#execute' do
   def create_build(name, **opts)
     create(:ci_build, :created, pipeline: pipeline, name: name, **opts)
   end
+
+  def delayed_options
+    { when: 'delayed', options: { start_in: '1 minute' } }
+  end
+
+  def unschedule
+    pipeline.builds.scheduled.map(&:unschedule)
+  end
 end