1 files changed, 96 insertions, 74 deletions

diff --git a/spec/workers/bulk_imports/entity_worker_spec.rb b/spec/workers/bulk_imports/entity_worker_spec.rb
index deae15a3ca2..ce45299c7f7 100644
--- a/spec/workers/bulk_imports/entity_worker_spec.rb
+++ b/spec/workers/bulk_imports/entity_worker_spec.rb
@@ -14,96 +14,118 @@ RSpec.describe BulkImports::EntityWorker do
     )
   end
 
-  it 'enqueues the first stage pipelines work' do
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
-        .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: nil
-        )
-    end
+  let(:job_args) { entity.id }
 
-    expect(BulkImports::PipelineWorker)
-      .to receive(:perform_async)
-      .with(
-        pipeline_tracker.id,
-        pipeline_tracker.stage,
-        entity.id
-      )
+  it 'updates pipeline trackers to enqueued state when selected' do
+    worker = BulkImports::EntityWorker.new
 
-    subject.perform(entity.id)
-  end
+    next_tracker = worker.send(:next_pipeline_trackers_for, entity.id).first
 
-  it 'do not enqueue a new pipeline job if the current stage still running' do
-    expect(BulkImports::PipelineWorker)
-      .not_to receive(:perform_async)
+    next_tracker.reload
 
-    subject.perform(entity.id, 0)
-  end
-
-  it 'enqueues the next stage pipelines when the current stage is finished' do
-    next_stage_pipeline_tracker = create(
-      :bulk_import_tracker,
-      entity: entity,
-      pipeline_name: 'Stage1::Pipeline',
-      stage: 1
-    )
+    expect(next_tracker.enqueued?).to be_truthy
 
-    pipeline_tracker.fail_op!
+    expect(worker.send(:next_pipeline_trackers_for, entity.id))
+      .not_to include(next_tracker)
+  end
 
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
+  include_examples 'an idempotent worker' do
+    it 'enqueues the first stage pipelines work' do
+      expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+        # the worker runs twice but only executes once
+        expect(logger)
+          .to receive(:info).twice
+          .with(
+            worker: described_class.name,
+            entity_id: entity.id,
+            current_stage: nil
+          )
+      end
+
+      expect(BulkImports::PipelineWorker)
+        .to receive(:perform_async)
         .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: 0
+          pipeline_tracker.id,
+          pipeline_tracker.stage,
+          entity.id
         )
+
+      subject
     end
 
-    expect(BulkImports::PipelineWorker)
-      .to receive(:perform_async)
-      .with(
-        next_stage_pipeline_tracker.id,
-        next_stage_pipeline_tracker.stage,
-        entity.id
-      )
+    it 'logs and tracks the raised exceptions' do
+      exception = StandardError.new('Error!')
+
+      expect(BulkImports::PipelineWorker)
+        .to receive(:perform_async)
+              .and_raise(exception)
+
+      expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+        expect(logger)
+          .to receive(:info).twice
+          .with(
+            worker: described_class.name,
+            entity_id: entity.id,
+            current_stage: nil
+          )
+
+        expect(logger)
+          .to receive(:error)
+          .with(
+            worker: described_class.name,
+            entity_id: entity.id,
+            current_stage: nil,
+            error_message: 'Error!'
+          )
+      end
+
+      expect(Gitlab::ErrorTracking)
+        .to receive(:track_exception)
+              .with(exception, entity_id: entity.id)
+
+      subject
+    end
 
-    subject.perform(entity.id, 0)
-  end
+    context 'in first stage' do
+      let(:job_args) { [entity.id, 0] }
 
-  it 'logs and tracks the raised exceptions' do
-    exception = StandardError.new('Error!')
+      it 'do not enqueue a new pipeline job if the current stage still running' do
+        expect(BulkImports::PipelineWorker)
+          .not_to receive(:perform_async)
 
-    expect(BulkImports::PipelineWorker)
-      .to receive(:perform_async)
-      .and_raise(exception)
+        subject
+      end
 
-    expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-      expect(logger)
-        .to receive(:info)
-        .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: nil
+      it 'enqueues the next stage pipelines when the current stage is finished' do
+        next_stage_pipeline_tracker = create(
+          :bulk_import_tracker,
+          entity: entity,
+          pipeline_name: 'Stage1::Pipeline',
+          stage: 1
         )
 
-      expect(logger)
-        .to receive(:error)
-        .with(
-          worker: described_class.name,
-          entity_id: entity.id,
-          current_stage: nil,
-          error_message: 'Error!'
-        )
+        pipeline_tracker.fail_op!
+
+        expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+          expect(logger)
+            .to receive(:info).twice
+            .with(
+              worker: described_class.name,
+              entity_id: entity.id,
+              current_stage: 0
+            )
+        end
+
+        expect(BulkImports::PipelineWorker)
+          .to receive(:perform_async)
+            .with(
+              next_stage_pipeline_tracker.id,
+              next_stage_pipeline_tracker.stage,
+              entity.id
+            )
+
+        subject
+      end
     end
-
-    expect(Gitlab::ErrorTracking)
-      .to receive(:track_exception)
-      .with(exception, entity_id: entity.id)
-
-    subject.perform(entity.id)
   end
 end