Ensure that we test all possible combinations

We need to cartesian test legacy and atomic processing
with different feature flags

This introduces a HACK to align behaviour
of atomic processing with legacy processing

Fix some YAML configs for Pipeline Processing

This ensures that all statuses are covered
for Pipeline and Stage

app/models/ci/pipeline.rb

@@ -115,8 +115,11 @@ module Ci
 
     state_machine :status, initial: :created do
       event :enqueue do
-        transition [:created, :waiting_for_resource, :preparing, :skipped, :scheduled] => :pending
+        transition [:created, :manual, :waiting_for_resource, :preparing, :skipped, :scheduled] => :pending
         transition [:success, :failed, :canceled] => :running
+
+        # this is needed to ensure tests to be covered
+        transition [:running] => :running
       end
 
       event :request_resource do

app/models/ci/stage.rb

@@ -42,8 +42,7 @@ module Ci
 
     state_machine :status, initial: :created do
       event :enqueue do
-        transition [:created, :waiting_for_resource, :preparing] => :pending
-        transition [:success, :failed, :canceled, :skipped] => :running
+        transition any - [:pending] => :pending
       end
 
       event :request_resource do

app/services/ci/pipeline_processing/atomic_processing_service.rb

@@ -95,7 +95,7 @@ module Ci
       def processable_status(processable)
         if processable.scheduling_type_dag?
           # Processable uses DAG, get status of all dependent needs
-          @collection.status_for_names(processable.aggregated_needs_names.to_a)
+          @collection.status_for_names(processable.aggregated_needs_names.to_a, dag: true)
         else
           # Processable uses Stages, get status of prior stage
           @collection.status_for_prior_stage_position(processable.stage_idx.to_i)

app/services/ci/pipeline_processing/atomic_processing_service/status_collection.rb

@@ -32,14 +32,14 @@ module Ci
 
         # This methods gets composite status of all processables
         def status_of_all
-          status_for_array(all_statuses)
+          status_for_array(all_statuses, dag: false)
         end
 
         # This methods gets composite status for processables with given names
-        def status_for_names(names)
+        def status_for_names(names, dag:)
           name_statuses = all_statuses_by_name.slice(*names)
 
-          status_for_array(name_statuses.values)
+          status_for_array(name_statuses.values, dag: dag)
         end
 
         # This methods gets composite status for processables before given stage
@@ -48,7 +48,7 @@ module Ci
             stage_statuses = all_statuses_grouped_by_stage_position
               .select { |stage_position, _| stage_position < position }
 
-            status_for_array(stage_statuses.values.flatten)
+            status_for_array(stage_statuses.values.flatten, dag: false)
           end
         end
 
@@ -65,7 +65,7 @@ module Ci
           strong_memoize("status_for_stage_position_#{current_position}") do
             stage_statuses = all_statuses_grouped_by_stage_position[current_position].to_a
 
-            status_for_array(stage_statuses.flatten)
+            status_for_array(stage_statuses.flatten, dag: false)
           end
         end
 
@@ -76,7 +76,14 @@ module Ci
 
         private
 
-        def status_for_array(statuses)
+        def status_for_array(statuses, dag:)
+          # TODO: This is hack to support
+          # the same exact behaviour for Atomic and Legacy processing
+          # that DAG is blocked from executing if dependent is not "complete"
+          if dag && statuses.any? { |status| HasStatus::COMPLETED_STATUSES.exclude?(status[:status]) }
+            return 'pending'
+          end
+
           result = Gitlab::Ci::Status::Composite
             .new(statuses)
             .status

spec/models/ci/pipeline_spec.rb

@@ -53,6 +53,29 @@ describe Ci::Pipeline, :mailer do
     end
   end
 
+  describe '#set_status' do
+    where(:from_status, :to_status) do
+      from_status_names = described_class.state_machines[:status].states.map(&:name)
+      to_status_names = from_status_names - [:created] # we never want to transition into created
+
+      from_status_names.product(to_status_names)
+    end
+
+    with_them do
+      it do
+        pipeline.status = from_status.to_s
+
+        if from_status != to_status
+          expect(pipeline.set_status(to_status.to_s))
+            .to eq(true)
+        else
+          expect(pipeline.set_status(to_status.to_s))
+            .to eq(false), "loopback transitions are not allowed"
+        end
+      end
+    end
+  end
+
   describe '.processables' do
     before do
       create(:ci_build, name: 'build', pipeline: pipeline)

spec/models/ci/stage_spec.rb

@@ -3,7 +3,8 @@
 require 'spec_helper'
 
 describe Ci::Stage, :models do
-  let(:stage) { create(:ci_stage_entity) }
+  let_it_be(:pipeline) { create(:ci_empty_pipeline) }
+  let(:stage) { create(:ci_stage_entity, pipeline: pipeline, project: pipeline.project) }
 
   it_behaves_like 'having unique enum values'
 
@@ -55,6 +56,29 @@ describe Ci::Stage, :models do
     end
   end
 
+  describe '#set_status' do
+    where(:from_status, :to_status) do
+      from_status_names = described_class.state_machines[:status].states.map(&:name)
+      to_status_names = from_status_names - [:created] # we never want to transition into created
+
+      from_status_names.product(to_status_names)
+    end
+
+    with_them do
+      it do
+        stage.status = from_status.to_s
+
+        if from_status != to_status
+          expect(stage.set_status(to_status.to_s))
+            .to eq(true)
+        else
+          expect(stage.set_status(to_status.to_s))
+            .to eq(false), "loopback transitions are not allowed"
+        end
+      end
+    end
+  end
+
   describe '#update_status' do
     context 'when stage objects needs to be updated' do
       before do

spec/services/ci/pipeline_processing/atomic_processing_service/status_collection_spec.rb

@@ -18,7 +18,7 @@ describe Ci::PipelineProcessing::AtomicProcessingService::StatusCollection do
     it 'does update existing status of processable' do
       collection.set_processable_status(test_a.id, 'success', 100)
 
-      expect(collection.status_for_names(['test-a'])).to eq('success')
+      expect(collection.status_for_names(['test-a'], dag: false)).to eq('success')
     end
 
     it 'ignores a missing processable' do
@@ -33,15 +33,18 @@ describe Ci::PipelineProcessing::AtomicProcessingService::StatusCollection do
   end
 
   describe '#status_for_names' do
-    where(:names, :status) do
-      %w[build-a]         | 'success'
-      %w[build-a build-b] | 'failed'
-      %w[build-a test-a]  | 'running'
+    where(:names, :status, :dag) do
+      %w[build-a]         | 'success' | false
+      %w[build-a build-b] | 'failed'  | false
+      %w[build-a test-a]  | 'running' | false
+      %w[build-a]         | 'success' | true
+      %w[build-a build-b] | 'failed'  | true
+      %w[build-a test-a]  | 'pending' | true
     end
 
     with_them do
       it 'returns composite status of given names' do
-        expect(collection.status_for_names(names)).to eq(status)
+        expect(collection.status_for_names(names, dag: dag)).to eq(status)
       end
     end
   end

spec/services/ci/pipeline_processing/atomic_processing_service_spec.rb

@@ -2,17 +2,19 @@
 
 require 'spec_helper'
 require_relative 'shared_processing_service.rb'
-# require_relative 'shared_processing_service_tests_with_yaml.rb'
+require_relative 'shared_processing_service_tests_with_yaml.rb'
 
 describe Ci::PipelineProcessing::AtomicProcessingService do
   before do
     stub_feature_flags(ci_atomic_processing: true)
+
+    # This feature flag is implicit
+    # Atomic Processing does not process statuses differently
+    stub_feature_flags(ci_composite_status: true)
   end
 
   it_behaves_like 'Pipeline Processing Service'
-  # TODO: This needs to be enabled. There is a different behavior when using `needs` depending on
-  # a `manual` job. More info: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/29405#note_327520605
-  # it_behaves_like 'Pipeline Processing Service Tests With Yaml'
+  it_behaves_like 'Pipeline Processing Service Tests With Yaml'
 
   private
 

spec/services/ci/pipeline_processing/legacy_processing_service_spec.rb

@@ -7,11 +7,25 @@ require_relative 'shared_processing_service_tests_with_yaml.rb'
 describe Ci::PipelineProcessing::LegacyProcessingService do
   before do
     stub_feature_flags(ci_atomic_processing: false)
-    stub_feature_flags(ci_composite_status: false)
   end
 
-  it_behaves_like 'Pipeline Processing Service'
-  it_behaves_like 'Pipeline Processing Service Tests With Yaml'
+  context 'when ci_composite_status is enabled' do
+    before do
+      stub_feature_flags(ci_composite_status: true)
+    end
+
+    it_behaves_like 'Pipeline Processing Service'
+    it_behaves_like 'Pipeline Processing Service Tests With Yaml'
+  end
+
+  context 'when ci_composite_status is disabled' do
+    before do
+      stub_feature_flags(ci_composite_status: false)
+    end
+
+    it_behaves_like 'Pipeline Processing Service'
+    it_behaves_like 'Pipeline Processing Service Tests With Yaml'
+  end
 
   private
 

spec/services/ci/pipeline_processing/shared_processing_service.rb

@@ -816,10 +816,10 @@ shared_examples 'Pipeline Processing Service' do
   context 'when a needed job is skipped', :sidekiq_inline do
     let!(:linux_build) { create_build('linux:build', stage: 'build', stage_idx: 0) }
     let!(:linux_rspec) { create_build('linux:rspec', stage: 'test', stage_idx: 1) }
-    let!(:deploy) do
-      create_build('deploy', stage: 'deploy', stage_idx: 2, scheduling_type: :dag, needs: [
-        create(:ci_build_need, name: 'linux:rspec')
-      ])
+    let!(:deploy) { create_build('deploy', stage: 'deploy', stage_idx: 2, scheduling_type: :dag) }
+
+    before do
+      create(:ci_build_need, build: deploy, name: 'linux:build')
     end
 
     it 'skips the jobs depending on it' do
@@ -836,6 +836,23 @@ shared_examples 'Pipeline Processing Service' do
     end
   end
 
+  context 'when a needed job is manual', :sidekiq_inline do
+    let!(:linux_build) { create_build('linux:build', stage: 'build', stage_idx: 0, when: 'manual', allow_failure: true) }
+    let!(:deploy) { create_build('deploy', stage: 'deploy', stage_idx: 1, scheduling_type: :dag) }
+
+    before do
+      create(:ci_build_need, build: deploy, name: 'linux:build')
+    end
+
+    it 'makes deploy DAG to be waiting for optional manual to finish' do
+      expect(process_pipeline).to be_truthy
+
+      expect(stages).to eq(%w(skipped created))
+      expect(all_builds.manual).to contain_exactly(linux_build)
+      expect(all_builds.created).to contain_exactly(deploy)
+    end
+  end
+
   private
 
   def all_builds

spec/services/ci/pipeline_processing/shared_processing_service_tests_with_yaml.rb

@@ -18,36 +18,40 @@ shared_context 'Pipeline Processing Service Tests With Yaml' do
       project.add_developer(user)
     end
 
-    it 'follows transitions', :sidekiq_inline do
+    it 'follows transitions' do
       expect(pipeline).to be_persisted
-      check_expectation(test_file.dig('init', 'expect'))
+      Sidekiq::Worker.drain_all # ensure that all async jobs are executed
+      check_expectation(test_file.dig('init', 'expect'), "init")
 
-      test_file['transitions'].each do |transition|
+      test_file['transitions'].each_with_index do |transition, idx|
         event_on_jobs(transition['event'], transition['jobs'])
-        check_expectation(transition['expect'])
+        Sidekiq::Worker.drain_all # ensure that all async jobs are executed
+        check_expectation(transition['expect'], "transition:#{idx}")
       end
     end
 
     private
 
-    def check_expectation(expectation)
-      expectation.each do |key, value|
-        case key
-        when 'pipeline'
-          expect(pipeline.reload.status).to eq(value)
-        when 'stages'
-          expect(pipeline.stages.pluck(:name, :status).to_h).to eq(value)
-        when 'jobs'
-          expect(pipeline.builds.latest.pluck(:name, :status).to_h).to eq(value)
-        end
-      end
+    def check_expectation(expectation, message)
+      expect(current_state.deep_stringify_keys).to eq(expectation), message
+    end
+
+    def current_state
+      # reload pipeline and all relations
+      pipeline.reload
+
+      {
+        pipeline: pipeline.status,
+        stages: pipeline.ordered_stages.pluck(:name, :status).to_h,
+        jobs: pipeline.statuses.latest.pluck(:name, :status).to_h
+      }
     end
 
     def event_on_jobs(event, job_names)
-      builds = pipeline.builds.latest.where(name: job_names).to_a
-      expect(builds.count).to eq(job_names.count) # ensure that we have the same counts
+      statuses = pipeline.statuses.latest.by_name(job_names).to_a
+      expect(statuses.count).to eq(job_names.count) # ensure that we have the same counts
 
-      builds.each { |build| build.public_send("#{event}!") }
+      statuses.each { |status| status.public_send("#{event}!") }
     end
   end
 end

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_false.yml

@@ -24,9 +24,9 @@ transitions:
   - event: enqueue
     jobs: [test]
     expect:
-      pipeline: manual
+      pipeline: pending
       stages:
-        test: manual
+        test: pending
         deploy: created
       jobs:
         test: pending

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true.yml

@@ -26,7 +26,7 @@ transitions:
     expect:
       pipeline: pending
       stages:
-        test: running
+        test: pending
         deploy: created
       jobs:
         test: pending

spec/services/ci/pipeline_processing/test_cases/dag_test_manual_allow_failure_true_deploy_on_failure.yml

@@ -27,7 +27,7 @@ transitions:
     expect:
       pipeline: pending
       stages:
-        test: running
+        test: pending
         deploy: created
       jobs:
         test: pending

spec/services/ci/pipeline_processing/test_cases/stage_test_manual_allow_failure_false.yml

@@ -23,9 +23,9 @@ transitions:
   - event: enqueue
     jobs: [test]
     expect:
-      pipeline: manual
+      pipeline: pending
       stages:
-        test: manual
+        test: pending
         deploy: created
       jobs:
         test: pending

spec/services/ci/pipeline_processing/test_cases/stage_test_manual_allow_failure_true.yml

@@ -36,7 +36,7 @@ transitions:
     expect:
       pipeline: running
       stages:
-        test: running
+        test: pending
         deploy: success
       jobs:
         test: pending

spec/services/ci/pipeline_processing/test_cases/stage_test_manual_allow_failure_true_deploy_on_failure.yml

@@ -26,7 +26,7 @@ transitions:
     expect:
       pipeline: pending
       stages:
-        test: running
+        test: pending
         deploy: skipped
       jobs:
         test: pending