3 files changed, 78 insertions, 34 deletions

diff --git a/spec/lib/bulk_imports/pipeline/context_spec.rb b/spec/lib/bulk_imports/pipeline/context_spec.rb
index c8c3fe3a861..5b7711ad5d7 100644
--- a/spec/lib/bulk_imports/pipeline/context_spec.rb
+++ b/spec/lib/bulk_imports/pipeline/context_spec.rb
@@ -3,29 +3,52 @@
 require 'spec_helper'
 
 RSpec.describe BulkImports::Pipeline::Context do
-  let(:group) { instance_double(Group) }
-  let(:user) { instance_double(User) }
-  let(:bulk_import) { instance_double(BulkImport, user: user, configuration: :config) }
-
-  let(:entity) do
-    instance_double(
-      BulkImports::Entity,
-      bulk_import: bulk_import,
-      group: group
+  let_it_be(:user) { create(:user) }
+  let_it_be(:group) { create(:group) }
+  let_it_be(:bulk_import) { create(:bulk_import, user: user) }
+
+  let_it_be(:entity) do
+    create(
+      :bulk_import_entity,
+      source_full_path: 'source/full/path',
+      destination_name: 'My Destination Group',
+      destination_namespace: group.full_path,
+      group: group,
+      bulk_import: bulk_import
+    )
+  end
+
+  let_it_be(:tracker) do
+    create(
+      :bulk_import_tracker,
+      entity: entity,
+      pipeline_name: described_class.name
     )
   end
 
-  subject { described_class.new(entity) }
+  subject { described_class.new(tracker, extra: :data) }
+
+  describe '#entity' do
+    it { expect(subject.entity).to eq(entity) }
+  end
 
   describe '#group' do
     it { expect(subject.group).to eq(group) }
   end
 
+  describe '#bulk_import' do
+    it { expect(subject.bulk_import).to eq(bulk_import) }
+  end
+
   describe '#current_user' do
     it { expect(subject.current_user).to eq(user) }
   end
 
-  describe '#current_user' do
+  describe '#configuration' do
     it { expect(subject.configuration).to eq(bulk_import.configuration) }
   end
+
+  describe '#extra' do
+    it { expect(subject.extra).to eq(extra: :data) }
+  end
 end
diff --git a/spec/lib/bulk_imports/pipeline/extracted_data_spec.rb b/spec/lib/bulk_imports/pipeline/extracted_data_spec.rb
index 25c5178227a..9c79b3f4c9e 100644
--- a/spec/lib/bulk_imports/pipeline/extracted_data_spec.rb
+++ b/spec/lib/bulk_imports/pipeline/extracted_data_spec.rb
@@ -9,7 +9,7 @@ RSpec.describe BulkImports::Pipeline::ExtractedData do
   let(:page_info) do
     {
       'has_next_page' => has_next_page,
-      'end_cursor' => cursor
+      'next_page' => cursor
     }
   end
 
diff --git a/spec/lib/bulk_imports/pipeline/runner_spec.rb b/spec/lib/bulk_imports/pipeline/runner_spec.rb
index 59f01c9caaa..7235b7c95cd 100644
--- a/spec/lib/bulk_imports/pipeline/runner_spec.rb
+++ b/spec/lib/bulk_imports/pipeline/runner_spec.rb
@@ -38,23 +38,20 @@ RSpec.describe BulkImports::Pipeline::Runner do
       extractor BulkImports::Extractor
       transformer BulkImports::Transformer
       loader BulkImports::Loader
-
-      def after_run(_); end
     end
 
     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) }
+  let_it_be_with_reload(:entity) { create(:bulk_import_entity) }
+  let_it_be(:tracker) { create(:bulk_import_tracker, entity: entity) }
+  let_it_be(:context) { BulkImports::Pipeline::Context.new(tracker, 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 })
-
         expect_next_instance_of(BulkImports::Extractor) do |extractor|
           expect(extractor)
             .to receive(:extract)
@@ -132,6 +129,22 @@ RSpec.describe BulkImports::Pipeline::Runner do
         subject.run
       end
 
+      context 'when extracted data has multiple pages' do
+        it 'updates tracker information and runs pipeline again' do
+          first_page = extracted_data(has_next_page: true)
+          last_page = extracted_data
+
+          expect_next_instance_of(BulkImports::Extractor) do |extractor|
+            expect(extractor)
+              .to receive(:extract)
+              .with(context)
+              .and_return(first_page, last_page)
+          end
+
+          subject.run
+        end
+      end
+
       context 'when exception is raised' do
         before do
           allow_next_instance_of(BulkImports::Extractor) do |extractor|
@@ -170,12 +183,7 @@ RSpec.describe BulkImports::Pipeline::Runner do
             BulkImports::MyPipeline.abort_on_failure!
           end
 
-          it 'marks entity as failed' do
-            expect { subject.run }
-              .to change(entity, :status_name).to(:failed)
-          end
-
-          it 'logs warn message' do
+          it 'logs a warn message and marks entity as failed' do
             expect_next_instance_of(Gitlab::Import::Logger) do |logger|
               expect(logger).to receive(:warn)
                 .with(
@@ -188,6 +196,9 @@ RSpec.describe BulkImports::Pipeline::Runner do
             end
 
             subject.run
+
+            expect(entity.status_name).to eq(:failed)
+            expect(tracker.status_name).to eq(:failed)
           end
         end
 
@@ -206,11 +217,11 @@ RSpec.describe BulkImports::Pipeline::Runner do
         entity.fail_op!
 
         expect_next_instance_of(Gitlab::Import::Logger) do |logger|
-          expect(logger).to receive(:info)
+          expect(logger).to receive(:warn)
             .with(
               log_params(
                 context,
-                message: 'Skipping due to failed pipeline status',
+                message: 'Skipping pipeline due to failed entity',
                 pipeline_class: 'BulkImports::MyPipeline'
               )
             )
@@ -219,14 +230,24 @@ RSpec.describe BulkImports::Pipeline::Runner do
         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)
+    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
+
+    def extracted_data(has_next_page: false)
+      BulkImports::Pipeline::ExtractedData.new(
+        data: { foo: :bar },
+        page_info: {
+          'has_next_page' => has_next_page,
+          'next_page' => has_next_page ? 'cursor' : nil
+        }
+      )
+    end
   end
 end