1 files changed, 94 insertions, 68 deletions

diff --git a/spec/lib/bulk_imports/pipeline/runner_spec.rb b/spec/lib/bulk_imports/pipeline/runner_spec.rb
index 76e4e64a7d6..59f01c9caaa 100644
--- a/spec/lib/bulk_imports/pipeline/runner_spec.rb
+++ b/spec/lib/bulk_imports/pipeline/runner_spec.rb
@@ -27,29 +27,31 @@ RSpec.describe BulkImports::Pipeline::Runner do
     end
   end
 
-  describe 'pipeline runner' do
-    before do
-      stub_const('BulkImports::Extractor', extractor)
-      stub_const('BulkImports::Transformer', transformer)
-      stub_const('BulkImports::Loader', loader)
-
-      pipeline = Class.new do
-        include BulkImports::Pipeline
+  before do
+    stub_const('BulkImports::Extractor', extractor)
+    stub_const('BulkImports::Transformer', transformer)
+    stub_const('BulkImports::Loader', loader)
 
-        extractor BulkImports::Extractor
-        transformer BulkImports::Transformer
-        loader BulkImports::Loader
+    pipeline = Class.new do
+      include BulkImports::Pipeline
 
-        def after_run(_); end
-      end
+      extractor BulkImports::Extractor
+      transformer BulkImports::Transformer
+      loader BulkImports::Loader
 
-      stub_const('BulkImports::MyPipeline', pipeline)
+      def after_run(_); end
     end
 
-    context 'when entity is not marked as failed' do
-      let(:entity) { create(:bulk_import_entity) }
-      let(:context) { BulkImports::Pipeline::Context.new(entity) }
+    stub_const('BulkImports::MyPipeline', pipeline)
+  end
 
+  let_it_be_with_refind(:entity) { create(:bulk_import_entity) }
+  let(:context) { BulkImports::Pipeline::Context.new(entity, extra: :data) }
+
+  subject { BulkImports::MyPipeline.new(context) }
+
+  describe 'pipeline runner' do
+    context 'when entity is not marked as failed' do
       it 'runs pipeline extractor, transformer, loader' do
         extracted_data = BulkImports::Pipeline::ExtractedData.new(data: { foo: :bar })
 
@@ -76,58 +78,61 @@ RSpec.describe BulkImports::Pipeline::Runner do
         expect_next_instance_of(Gitlab::Import::Logger) do |logger|
           expect(logger).to receive(:info)
             .with(
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity',
-              message: 'Pipeline started',
-              pipeline_class: 'BulkImports::MyPipeline'
+              log_params(
+                context,
+                message: 'Pipeline started',
+                pipeline_class: 'BulkImports::MyPipeline'
+              )
             )
           expect(logger).to receive(:info)
             .with(
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity',
-              pipeline_class: 'BulkImports::MyPipeline',
-              pipeline_step: :extractor,
-              step_class: 'BulkImports::Extractor'
+              log_params(
+                context,
+                pipeline_class: 'BulkImports::MyPipeline',
+                pipeline_step: :extractor,
+                step_class: 'BulkImports::Extractor'
+              )
             )
           expect(logger).to receive(:info)
             .with(
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity',
-              pipeline_class: 'BulkImports::MyPipeline',
-              pipeline_step: :transformer,
-              step_class: 'BulkImports::Transformer'
+              log_params(
+                context,
+                pipeline_class: 'BulkImports::MyPipeline',
+                pipeline_step: :transformer,
+                step_class: 'BulkImports::Transformer'
+              )
             )
           expect(logger).to receive(:info)
             .with(
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity',
-              pipeline_class: 'BulkImports::MyPipeline',
-              pipeline_step: :loader,
-              step_class: 'BulkImports::Loader'
+              log_params(
+                context,
+                pipeline_class: 'BulkImports::MyPipeline',
+                pipeline_step: :loader,
+                step_class: 'BulkImports::Loader'
+              )
             )
           expect(logger).to receive(:info)
             .with(
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity',
-              pipeline_class: 'BulkImports::MyPipeline',
-              pipeline_step: :after_run
+              log_params(
+                context,
+                pipeline_class: 'BulkImports::MyPipeline',
+                pipeline_step: :after_run
+              )
             )
           expect(logger).to receive(:info)
             .with(
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity',
-              message: 'Pipeline finished',
-              pipeline_class: 'BulkImports::MyPipeline'
+              log_params(
+                context,
+                message: 'Pipeline finished',
+                pipeline_class: 'BulkImports::MyPipeline'
+              )
             )
         end
 
-        BulkImports::MyPipeline.new(context).run
+        subject.run
       end
 
       context 'when exception is raised' do
-        let(:entity) { create(:bulk_import_entity, :created) }
-        let(:context) { BulkImports::Pipeline::Context.new(entity) }
-
         before do
           allow_next_instance_of(BulkImports::Extractor) do |extractor|
             allow(extractor).to receive(:extract).with(context).and_raise(StandardError, 'Error!')
@@ -135,7 +140,21 @@ RSpec.describe BulkImports::Pipeline::Runner do
         end
 
         it 'logs import failure' do
-          BulkImports::MyPipeline.new(context).run
+          expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+            expect(logger).to receive(:error)
+              .with(
+                log_params(
+                  context,
+                  pipeline_step: :extractor,
+                  pipeline_class: 'BulkImports::MyPipeline',
+                  exception_class: 'StandardError',
+                  exception_message: 'Error!'
+                )
+              )
+          end
+
+          expect { subject.run }
+            .to change(entity.failures, :count).by(1)
 
           failure = entity.failures.first
 
@@ -152,29 +171,29 @@ RSpec.describe BulkImports::Pipeline::Runner do
           end
 
           it 'marks entity as failed' do
-            BulkImports::MyPipeline.new(context).run
-
-            expect(entity.failed?).to eq(true)
+            expect { subject.run }
+              .to change(entity, :status_name).to(:failed)
           end
 
           it 'logs warn message' do
             expect_next_instance_of(Gitlab::Import::Logger) do |logger|
               expect(logger).to receive(:warn)
                 .with(
-                  message: 'Pipeline failed',
-                  pipeline_class: 'BulkImports::MyPipeline',
-                  bulk_import_entity_id: entity.id,
-                  bulk_import_entity_type: entity.source_type
+                  log_params(
+                    context,
+                    message: 'Pipeline failed',
+                    pipeline_class: 'BulkImports::MyPipeline'
+                  )
                 )
             end
 
-            BulkImports::MyPipeline.new(context).run
+            subject.run
           end
         end
 
         context 'when pipeline is not marked to abort on failure' do
-          it 'marks entity as failed' do
-            BulkImports::MyPipeline.new(context).run
+          it 'does not mark entity as failed' do
+            subject.run
 
             expect(entity.failed?).to eq(false)
           end
@@ -183,24 +202,31 @@ RSpec.describe BulkImports::Pipeline::Runner do
     end
 
     context 'when entity is marked as failed' do
-      let(:entity) { create(:bulk_import_entity) }
-      let(:context) { BulkImports::Pipeline::Context.new(entity) }
-
       it 'logs and returns without execution' do
-        allow(entity).to receive(:failed?).and_return(true)
+        entity.fail_op!
 
         expect_next_instance_of(Gitlab::Import::Logger) do |logger|
           expect(logger).to receive(:info)
             .with(
-              message: 'Skipping due to failed pipeline status',
-              pipeline_class: 'BulkImports::MyPipeline',
-              bulk_import_entity_id: entity.id,
-              bulk_import_entity_type: 'group_entity'
+              log_params(
+                context,
+                message: 'Skipping due to failed pipeline status',
+                pipeline_class: 'BulkImports::MyPipeline'
+              )
             )
         end
 
-        BulkImports::MyPipeline.new(context).run
+        subject.run
       end
     end
   end
+
+  def log_params(context, extra = {})
+    {
+      bulk_import_id: context.bulk_import.id,
+      bulk_import_entity_id: context.entity.id,
+      bulk_import_entity_type: context.entity.source_type,
+      context_extra: context.extra
+    }.merge(extra)
+  end
 end