Ingest vulnerability flags with the new ingestion flow

This change was intentionally postponed to demonstrate the benefit of
using the new abstration. With the new ingestion tasks abstraction we
can introduce new tasks easily and test them individually.

ee/app/services/security/ingestion/ingest_report_slice_service.rb

@@ -16,6 +16,7 @@ module Security
         IngestFindingIdentifiers
         IngestFindingLinks
         IngestFindingSignatures
+        IngestVulnerabilityFlags
         IngestRemediations
       ].freeze
 

ee/app/services/security/ingestion/tasks/ingest_vulnerability_flags.rb

+# frozen_string_literal: true
+
+module Security
+  module Ingestion
+    module Tasks
+      # Ingests the data for the `Vulnerabilities`::Flag` model.
+      class IngestVulnerabilityFlags < AbstractTask
+        include BulkInsertableTask
+
+        self.model = Vulnerabilities::Flag
+        self.unique_by = :index_vulnerability_flags_on_unique_columns
+        self.uses = :id
+
+        private
+
+        def attributes
+          finding_maps.flat_map { |finding_map| flag_attributes_for(finding_map) }
+        end
+
+        def after_ingest
+          model.where(finding: finding_ids).where.not(id: return_data).delete_all # rubocop:disable CodeReuse/ActiveRecord
+        end
+
+        def finding_ids
+          finding_maps.map(&:finding_id)
+        end
+
+        def flag_attributes_for(finding_map)
+          finding_map.report_finding.flags.map do |vulnerability_flag|
+            vulnerability_flag.to_h.merge!(vulnerability_occurrence_id: finding_map.finding_id)
+          end
+        end
+      end
+    end
+  end
+end

ee/spec/services/security/ingestion/ingest_report_slice_service_spec.rb

@@ -29,6 +29,7 @@ RSpec.describe Security::Ingestion::IngestReportSliceService do
       expect(Security::Ingestion::Tasks::IngestFindingIdentifiers).to have_received(:execute).ordered.with(pipeline, finding_maps)
       expect(Security::Ingestion::Tasks::IngestFindingLinks).to have_received(:execute).ordered.with(pipeline, finding_maps)
       expect(Security::Ingestion::Tasks::IngestFindingSignatures).to have_received(:execute).ordered.with(pipeline, finding_maps)
+      expect(Security::Ingestion::Tasks::IngestVulnerabilityFlags).to have_received(:execute).ordered.with(pipeline, finding_maps)
       expect(Security::Ingestion::Tasks::IngestRemediations).to have_received(:execute).ordered.with(pipeline, finding_maps)
     end
 

ee/spec/services/security/ingestion/tasks/ingest_vulnerability_flags_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::Ingestion::Tasks::IngestVulnerabilityFlags do
+  describe '#execute' do
+    let_it_be(:pipeline) { create(:ci_pipeline) }
+
+    let(:vulnerability_finding) { create(:vulnerabilities_finding) }
+    let!(:existing_vulnerability_flag_1) { create(:vulnerabilities_flag, origin: 'post analyzer 1', finding: vulnerability_finding) }
+    let!(:existing_vulnerability_flag_2) { create(:vulnerabilities_flag, origin: 'post analyzer 2', finding: vulnerability_finding, description: 'foo') }
+
+    let(:report_finding_flag_2) { create(:ci_reports_security_flag, origin: 'post analyzer 2', description: 'bar') }
+    let(:report_finding_flag_3) { create(:ci_reports_security_flag, origin: 'post analyzer 3') }
+    let(:report_finding_flag_4) { create(:ci_reports_security_flag, origin: 'post analyzer 4') }
+
+    let(:report_finding) { create(:ci_reports_security_finding, flags: [report_finding_flag_2, report_finding_flag_3, report_finding_flag_4]) }
+    let(:finding_map) { create(:finding_map, finding: vulnerability_finding, report_finding: report_finding) }
+    let(:service_object) { described_class.new(pipeline, [finding_map]) }
+
+    subject(:ingest_vulnerability_flags) { service_object.execute }
+
+    it 'creates new records and updates the existing ones' do
+      expect { ingest_vulnerability_flags }.to change { vulnerability_finding.vulnerability_flags.count }.from(2).to(3)
+                                           .and change { existing_vulnerability_flag_2.reload.description }.from('foo').to('bar')
+    end
+
+    it 'deletes the old records' do
+      expect { ingest_vulnerability_flags }.to change { Vulnerabilities::Flag.find_by_id(existing_vulnerability_flag_1.id) }.to(nil)
+    end
+  end
+end