Merge branch 'georgekoltsov/group-mig-etl-inline-methods' into 'master'

Update BulkImports::Pipeline to make use of DSL methods optional

See merge request gitlab-org/gitlab!54368

ee/lib/ee/bulk_imports/groups/loaders/epic_award_emoji_loader.rb

@@ -7,10 +7,6 @@ module EE
         class EpicAwardEmojiLoader
           NotAllowedError = Class.new(StandardError)
 
-          def initialize(options = {})
-            @options = options
-          end
-
           # rubocop: disable CodeReuse/ActiveRecord
           def load(context, data)
             return unless data

ee/lib/ee/bulk_imports/groups/loaders/epics_loader.rb

@@ -7,10 +7,6 @@ module EE
         class EpicsLoader
           NotAllowedError = Class.new(StandardError)
 
-          def initialize(options = {})
-            @options = options
-          end
-
           def load(context, data)
             raise NotAllowedError unless authorized?(context)
 

ee/lib/ee/bulk_imports/groups/transformers/epic_attributes_transformer.rb

@@ -5,8 +5,6 @@ module EE
     module Groups
       module Transformers
         class EpicAttributesTransformer
-          def initialize(*args); end
-
           def transform(context, data)
             data
               .then { |data| add_group_id(context, data) }

lib/bulk_imports/common/loaders/entity_loader.rb

-# frozen_string_literal: true
-
-module BulkImports
-  module Common
-    module Loaders
-      class EntityLoader
-        def initialize(*args); end
-
-        def load(context, entity)
-          context.bulk_import.entities.create!(entity)
-        end
-      end
-    end
-  end
-end

lib/bulk_imports/common/transformers/award_emoji_transformer.rb

@@ -4,8 +4,6 @@ module BulkImports
   module Common
     module Transformers
       class AwardEmojiTransformer
-        def initialize(*args); end
-
         def transform(context, data)
           user = find_user(context, data&.dig('user', 'public_email')) || context.current_user
 

lib/bulk_imports/common/transformers/prohibited_attributes_transformer.rb

@@ -14,10 +14,6 @@ module BulkImports
           /\Aremote_\w+_(url|urls|request_header)\Z/ # carrierwave automatically creates these attribute methods for uploads
         ).freeze
 
-        def initialize(options = {})
-          @options = options
-        end
-
         def transform(context, data)
           data.each_with_object({}) do |(key, value), result|
             prohibited = prohibited_key?(key)

lib/bulk_imports/groups/extractors/subgroups_extractor.rb

@@ -4,8 +4,6 @@ module BulkImports
   module Groups
     module Extractors
       class SubgroupsExtractor
-        def initialize(*args); end
-
         def extract(context)
           encoded_parent_path = ERB::Util.url_encode(context.entity.source_full_path)
 

lib/bulk_imports/groups/loaders/group_loader.rb

@@ -4,10 +4,6 @@ module BulkImports
   module Groups
     module Loaders
       class GroupLoader
-        def initialize(options = {})
-          @options = options
-        end
-
         def load(context, data)
           return unless user_can_create_group?(context.current_user, data)
 

lib/bulk_imports/groups/loaders/labels_loader.rb

-# frozen_string_literal: true
-
-module BulkImports
-  module Groups
-    module Loaders
-      class LabelsLoader
-        def initialize(*); end
-
-        def load(context, data)
-          Labels::CreateService.new(data).execute(group: context.group)
-        end
-      end
-    end
-  end
-end

lib/bulk_imports/groups/loaders/members_loader.rb

-# frozen_string_literal: true
-
-module BulkImports
-  module Groups
-    module Loaders
-      class MembersLoader
-        def initialize(*); end
-
-        def load(context, data)
-          return unless data
-
-          context.group.members.create!(data)
-        end
-      end
-    end
-  end
-end

lib/bulk_imports/groups/pipelines/labels_pipeline.rb

@@ -11,7 +11,9 @@ module BulkImports
 
         transformer Common::Transformers::ProhibitedAttributesTransformer
 
-        loader BulkImports::Groups::Loaders::LabelsLoader
+        def load(context, data)
+          Labels::CreateService.new(data).execute(group: context.group)
+        end
 
         def after_run(extracted_data)
           context.entity.update_tracker_for(

lib/bulk_imports/groups/pipelines/members_pipeline.rb

@@ -12,7 +12,11 @@ module BulkImports
         transformer Common::Transformers::ProhibitedAttributesTransformer
         transformer BulkImports::Groups::Transformers::MemberAttributesTransformer
 
-        loader BulkImports::Groups::Loaders::MembersLoader
+        def load(context, data)
+          return unless data
+
+          context.group.members.create!(data)
+        end
 
         def after_run(extracted_data)
           context.entity.update_tracker_for(

lib/bulk_imports/groups/pipelines/subgroup_entities_pipeline.rb

@@ -9,7 +9,10 @@ module BulkImports
         extractor BulkImports::Groups::Extractors::SubgroupsExtractor
         transformer Common::Transformers::ProhibitedAttributesTransformer
         transformer BulkImports::Groups::Transformers::SubgroupToEntityTransformer
-        loader BulkImports::Common::Loaders::EntityLoader
+
+        def load(context, data)
+          context.bulk_import.entities.create!(data)
+        end
       end
     end
   end

lib/bulk_imports/groups/transformers/group_attributes_transformer.rb

@@ -4,10 +4,6 @@ module BulkImports
   module Groups
     module Transformers
       class GroupAttributesTransformer
-        def initialize(options = {})
-          @options = options
-        end
-
         def transform(context, data)
           import_entity = context.entity
 

lib/bulk_imports/groups/transformers/member_attributes_transformer.rb

@@ -4,8 +4,6 @@ module BulkImports
   module Groups
     module Transformers
       class MemberAttributesTransformer
-        def initialize(*); end
-
         def transform(context, data)
           data
             .then { |data| add_user(data) }

lib/bulk_imports/groups/transformers/subgroup_to_entity_transformer.rb

@@ -4,8 +4,6 @@ module BulkImports
   module Groups
     module Transformers
       class SubgroupToEntityTransformer
-        def initialize(*args); end
-
         def transform(context, entry)
           {
             source_type: :group_entity,

lib/bulk_imports/pipeline.rb

@@ -15,16 +15,78 @@ module BulkImports
 
       attr_reader :context
 
+      # Fetch pipeline extractor.
+      # An extractor is defined either by instance `#extract(context)` method
+      # or by using `extractor` DSL.
+      #
+      # @example
+      # class MyPipeline
+      #   extractor MyExtractor, foo: :bar
+      # end
+      #
+      # class MyPipeline
+      #   def extract(context)
+      #     puts 'Fetch some data'
+      #   end
+      # end
+      #
+      # If pipeline implements instance method `extract` - use it
+      # and ignore class `extractor` method implementation.
       def extractor
-        @extractor ||= instantiate(self.class.get_extractor)
-      end
-
+        @extractor ||= self.respond_to?(:extract) ? self : instantiate(self.class.get_extractor)
+      end
+
+      # Fetch pipeline transformers.
+      #
+      # A transformer can be defined using:
+      #   - `transformer` class method
+      #   - `transform` instance method
+      #
+      # Multiple transformers can be defined within a single
+      # pipeline and run sequentially for each record in the
+      # following order:
+      #   - Transformers defined using `transformer` class method
+      #   - Instance method `transform`
+      #
+      # Instance method `transform` is always the last to run.
+      #
+      # @example
+      # class MyPipeline
+      #   transformer MyTransformerOne, foo: :bar
+      #   transformer MyTransformerTwo, foo: :bar
+      #
+      #   def transform(context, data)
+      #     # perform transformation here
+      #   end
+      # end
+      #
+      # In the example above `MyTransformerOne` is the first to run and
+      # the instance `#transform` method is the last.
       def transformers
         @transformers ||= self.class.transformers.map(&method(:instantiate))
-      end
-
+        @transformers << self if respond_to?(:transform) && @transformers.exclude?(self)
+        @transformers
+      end
+
+      # Fetch pipeline loader.
+      # A loader is defined either by instance method `#load(context, data)`
+      # or by using `loader` DSL.
+      #
+      # @example
+      # class MyPipeline
+      #   loader MyLoader, foo: :bar
+      # end
+      #
+      # class MyPipeline
+      #   def load(context, data)
+      #     puts 'Load some data'
+      #   end
+      # end
+      #
+      # If pipeline implements instance method `load` - use it
+      # and ignore class `loader` method implementation.
       def loader
-        @loaders ||= instantiate(self.class.get_loader)
+        @loader ||= self.respond_to?(:load) ? self : instantiate(self.class.get_loader)
       end
 
       def pipeline
@@ -32,7 +94,13 @@ module BulkImports
       end
 
       def instantiate(class_config)
+        options = class_config[:options]
+
+        if options
           class_config[:klass].new(class_config[:options])
+        else
+          class_config[:klass].new
+        end
       end
 
       def abort_on_failure?

spec/lib/bulk_imports/common/loaders/entity_loader_spec.rb

-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe BulkImports::Common::Loaders::EntityLoader do
-  describe '#load' do
-    it "creates entities for the given data" do
-      group = create(:group, path: "imported-group")
-      parent_entity = create(:bulk_import_entity, group: group, bulk_import: create(:bulk_import))
-      context = BulkImports::Pipeline::Context.new(parent_entity)
-
-      data = {
-        source_type: :group_entity,
-        source_full_path: "parent/subgroup",
-        destination_name: "subgroup",
-        destination_namespace: parent_entity.group.full_path,
-        parent_id: parent_entity.id
-      }
-
-      expect { subject.load(context, data) }.to change(BulkImports::Entity, :count).by(1)
-
-      subgroup_entity = BulkImports::Entity.last
-
-      expect(subgroup_entity.source_full_path).to eq 'parent/subgroup'
-      expect(subgroup_entity.destination_namespace).to eq 'imported-group'
-      expect(subgroup_entity.destination_name).to eq 'subgroup'
-      expect(subgroup_entity.parent_id).to eq parent_entity.id
-    end
-  end
-end

spec/lib/bulk_imports/groups/loaders/labels_loader_spec.rb

-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe BulkImports::Groups::Loaders::LabelsLoader do
-  describe '#load' do
-    let(:user) { create(:user) }
-    let(:group) { create(:group) }
-    let(:entity) { create(:bulk_import_entity, group: group) }
-    let(:context) { BulkImports::Pipeline::Context.new(entity) }
-
-    let(:data) do
-      {
-        'title' => 'label',
-        'description' => 'description',
-        'color' => '#FFFFFF'
-      }
-    end
-
-    it 'creates the label' do
-      expect { subject.load(context, data) }.to change(Label, :count).by(1)
-
-      label = group.labels.first
-
-      expect(label.title).to eq(data['title'])
-      expect(label.description).to eq(data['description'])
-      expect(label.color).to eq(data['color'])
-    end
-  end
-end

spec/lib/bulk_imports/groups/loaders/members_loader_spec.rb

-# frozen_string_literal: true
-
-require 'spec_helper'
-
-RSpec.describe BulkImports::Groups::Loaders::MembersLoader do
-  describe '#load' do
-    let_it_be(:user_importer) { create(:user) }
-    let_it_be(:user_member) { create(:user) }
-    let_it_be(:group) { create(:group) }
-    let_it_be(:bulk_import) { create(:bulk_import, user: user_importer) }
-    let_it_be(:entity) { create(:bulk_import_entity, bulk_import: bulk_import, group: group) }
-    let_it_be(:context) { BulkImports::Pipeline::Context.new(entity) }
-
-    let_it_be(:data) do
-      {
-        'user_id' => user_member.id,
-        'created_by_id' => user_importer.id,
-        'access_level' => 30,
-        'created_at' => '2020-01-01T00:00:00Z',
-        'updated_at' => '2020-01-01T00:00:00Z',
-        'expires_at' => nil
-      }
-    end
-
-    it 'does nothing when there is no data' do
-      expect { subject.load(context, nil) }.not_to change(GroupMember, :count)
-    end
-
-    it 'creates the member' do
-      expect { subject.load(context, data) }.to change(GroupMember, :count).by(1)
-
-      member = group.members.last
-
-      expect(member.user).to eq(user_member)
-      expect(member.created_by).to eq(user_importer)
-      expect(member.access_level).to eq(30)
-      expect(member.created_at).to eq('2020-01-01T00:00:00Z')
-      expect(member.updated_at).to eq('2020-01-01T00:00:00Z')
-      expect(member.expires_at).to eq(nil)
-    end
-  end
-end

spec/lib/bulk_imports/groups/pipelines/labels_pipeline_spec.rb

@@ -90,6 +90,24 @@ RSpec.describe BulkImports::Groups::Pipelines::LabelsPipeline do
     end
   end
 
+  describe '#load' do
+    it 'creates the label' do
+      data = {
+        'title' => 'label',
+        'description' => 'description',
+        'color' => '#FFFFFF'
+      }
+
+      expect { subject.load(context, data) }.to change(Label, :count).by(1)
+
+      label = group.labels.first
+
+      expect(label.title).to eq(data['title'])
+      expect(label.description).to eq(data['description'])
+      expect(label.color).to eq(data['color'])
+    end
+  end
+
   describe 'pipeline parts' do
     it { expect(described_class).to include_module(BulkImports::Pipeline) }
     it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
@@ -110,9 +128,5 @@ RSpec.describe BulkImports::Groups::Pipelines::LabelsPipeline do
           { klass: BulkImports::Common::Transformers::ProhibitedAttributesTransformer, options: nil }
         )
     end
-
-    it 'has loaders' do
-      expect(described_class.get_loader).to eq(klass: BulkImports::Groups::Loaders::LabelsLoader, options: nil)
-    end
   end
 end

spec/lib/bulk_imports/groups/pipelines/members_pipeline_spec.rb

@@ -37,6 +37,34 @@ RSpec.describe BulkImports::Groups::Pipelines::MembersPipeline do
     end
   end
 
+  describe '#load' do
+    it 'does nothing when there is no data' do
+      expect { subject.load(context, nil) }.not_to change(GroupMember, :count)
+    end
+
+    it 'creates the member' do
+      data = {
+        'user_id' => member_user1.id,
+        'created_by_id' => member_user2.id,
+        'access_level' => 30,
+        'created_at' => '2020-01-01T00:00:00Z',
+        'updated_at' => '2020-01-01T00:00:00Z',
+        'expires_at' => nil
+      }
+
+      expect { subject.load(context, data) }.to change(GroupMember, :count).by(1)
+
+      member = group.members.last
+
+      expect(member.user).to eq(member_user1)
+      expect(member.created_by).to eq(member_user2)
+      expect(member.access_level).to eq(30)
+      expect(member.created_at).to eq('2020-01-01T00:00:00Z')
+      expect(member.updated_at).to eq('2020-01-01T00:00:00Z')
+      expect(member.expires_at).to eq(nil)
+    end
+  end
+
   describe 'pipeline parts' do
     it { expect(described_class).to include_module(BulkImports::Pipeline) }
     it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
@@ -58,10 +86,6 @@ RSpec.describe BulkImports::Groups::Pipelines::MembersPipeline do
           { klass: BulkImports::Groups::Transformers::MemberAttributesTransformer, options: nil }
         )
     end
-
-    it 'has loaders' do
-      expect(described_class.get_loader).to eq(klass: BulkImports::Groups::Loaders::MembersLoader, options: nil)
-    end
   end
 
   def member_data(email:, has_next_page:, cursor: nil)

spec/lib/bulk_imports/groups/pipelines/subgroup_entities_pipeline_spec.rb

@@ -3,9 +3,14 @@
 require 'spec_helper'
 
 RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
-  describe '#run' do
   let_it_be(:user) { create(:user) }
-    let(:parent) { create(:group, name: 'imported-group', path: 'imported-group') }
+  let_it_be(:group) { create(:group, path: 'group') }
+  let_it_be(:parent) { create(:group, name: 'imported-group', path: 'imported-group') }
+  let(:context) { BulkImports::Pipeline::Context.new(parent_entity) }
+
+  subject { described_class.new(context) }
+
+  describe '#run' do
     let!(:parent_entity) do
       create(
         :bulk_import_entity,
@@ -14,8 +19,6 @@ RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
       )
     end
 
-    let(:context) { BulkImports::Pipeline::Context.new(parent_entity) }
-
     let(:subgroup_data) do
       [
         {
@@ -25,8 +28,6 @@ RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
       ]
     end
 
-    subject { described_class.new(context) }
-
     before do
       allow_next_instance_of(BulkImports::Groups::Extractors::SubgroupsExtractor) do |extractor|
         allow(extractor).to receive(:extract).and_return(subgroup_data)
@@ -47,6 +48,29 @@ RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
     end
   end
 
+  describe '#load' do
+    let(:parent_entity) { create(:bulk_import_entity, group: group, bulk_import: create(:bulk_import)) }
+
+    it 'creates entities for the given data' do
+      data = {
+        source_type: :group_entity,
+        source_full_path: 'parent/subgroup',
+        destination_name: 'subgroup',
+        destination_namespace: parent_entity.group.full_path,
+        parent_id: parent_entity.id
+      }
+
+      expect { subject.load(context, data) }.to change(BulkImports::Entity, :count).by(1)
+
+      subgroup_entity = BulkImports::Entity.last
+
+      expect(subgroup_entity.source_full_path).to eq 'parent/subgroup'
+      expect(subgroup_entity.destination_namespace).to eq 'group'
+      expect(subgroup_entity.destination_name).to eq 'subgroup'
+      expect(subgroup_entity.parent_id).to eq parent_entity.id
+    end
+  end
+
   describe 'pipeline parts' do
     it { expect(described_class).to include_module(BulkImports::Pipeline) }
     it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
@@ -61,9 +85,5 @@ RSpec.describe BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline do
         { klass: BulkImports::Groups::Transformers::SubgroupToEntityTransformer, options: nil }
       )
     end
-
-    it 'has loaders' do
-      expect(described_class.get_loader).to eq(klass: BulkImports::Common::Loaders::EntityLoader, options: nil)
-    end
   end
 end

spec/lib/bulk_imports/pipeline_spec.rb

@@ -3,7 +3,6 @@
 require 'spec_helper'
 
 RSpec.describe BulkImports::Pipeline do
-  describe 'pipeline attributes' do
   before do
     stub_const('BulkImports::Extractor', Class.new)
     stub_const('BulkImports::Transformer', Class.new)
@@ -14,14 +13,15 @@ RSpec.describe BulkImports::Pipeline do
 
       abort_on_failure!
 
-        extractor BulkImports::Extractor, { foo: :bar }
-        transformer BulkImports::Transformer, { foo: :bar }
-        loader BulkImports::Loader, { foo: :bar }
+      extractor BulkImports::Extractor, foo: :bar
+      transformer BulkImports::Transformer, foo: :bar
+      loader BulkImports::Loader, foo: :bar
     end
 
     stub_const('BulkImports::MyPipeline', klass)
   end
 
+  describe 'pipeline attributes' do
     describe 'getters' do
       it 'retrieves class attributes' do
         expect(BulkImports::MyPipeline.get_extractor).to eq({ klass: BulkImports::Extractor, options: { foo: :bar } })
@@ -29,6 +29,27 @@ RSpec.describe BulkImports::Pipeline do
         expect(BulkImports::MyPipeline.get_loader).to eq({ klass: BulkImports::Loader, options: { foo: :bar } })
         expect(BulkImports::MyPipeline.abort_on_failure?).to eq(true)
       end
+
+      context 'when extractor and loader are defined within the pipeline' do
+        before do
+          klass = Class.new do
+            include BulkImports::Pipeline
+
+            def extract; end
+
+            def load; end
+          end
+
+          stub_const('BulkImports::AnotherPipeline', klass)
+        end
+
+        it 'returns itself when retrieving extractor & loader' do
+          pipeline = BulkImports::AnotherPipeline.new(nil)
+
+          expect(pipeline.send(:extractor)).to eq(pipeline)
+          expect(pipeline.send(:loader)).to eq(pipeline)
+        end
+      end
     end
 
     describe 'setters' do
@@ -54,4 +75,46 @@ RSpec.describe BulkImports::Pipeline do
       end
     end
   end
+
+  describe '#instantiate' do
+    context 'when options are present' do
+      it 'instantiates new object with options' do
+        expect(BulkImports::Extractor).to receive(:new).with(foo: :bar)
+        expect(BulkImports::Transformer).to receive(:new).with(foo: :bar)
+        expect(BulkImports::Loader).to receive(:new).with(foo: :bar)
+
+        pipeline = BulkImports::MyPipeline.new(nil)
+
+        pipeline.send(:extractor)
+        pipeline.send(:transformers)
+        pipeline.send(:loader)
+      end
+    end
+
+    context 'when options are missing' do
+      before do
+        klass = Class.new do
+          include BulkImports::Pipeline
+
+          extractor BulkImports::Extractor
+          transformer BulkImports::Transformer
+          loader BulkImports::Loader
+        end
+
+        stub_const('BulkImports::NoOptionsPipeline', klass)
+      end
+
+      it 'instantiates new object without options' do
+        expect(BulkImports::Extractor).to receive(:new).with(no_args)
+        expect(BulkImports::Transformer).to receive(:new).with(no_args)
+        expect(BulkImports::Loader).to receive(:new).with(no_args)
+
+        pipeline = BulkImports::NoOptionsPipeline.new(nil)
+
+        pipeline.send(:extractor)
+        pipeline.send(:transformers)
+        pipeline.send(:loader)
+      end
+    end
+  end
 end