Add Group Import via GraphQL

- Introduce Group Import ETL pipeline to import
  groups from source GitLab instance using GraphQL
- Add sidekiq jobs & workers to execute group import
  in a distributed fashion

Gemfile

@@ -98,6 +98,7 @@ gem 'graphql', '~> 1.11.4'
 gem 'graphiql-rails', '~> 1.4.10'
 gem 'apollo_upload_server', '~> 2.0.2'
 gem 'graphql-docs', '~> 1.6.0', group: [:development, :test]
+gem 'graphlient', '~> 0.4.0' # Used by BulkImport feature (group::import)
 
 gem 'hashie'
 # Disable strong_params so that Mash does not respond to :permitted?

Gemfile.lock

@@ -506,7 +506,14 @@ GEM
     graphiql-rails (1.4.10)
       railties
       sprockets-rails
+    graphlient (0.4.0)
+      faraday (>= 1.0)
+      faraday_middleware
+      graphql-client
     graphql (1.11.4)
+    graphql-client (0.16.0)
+      activesupport (>= 3.0)
+      graphql (~> 1.8)
     graphql-docs (1.6.0)
       commonmarker (~> 0.16)
       escape_utils (~> 1.2)
@@ -1353,6 +1360,7 @@ DEPENDENCIES
   grape-path-helpers (~> 1.4)
   grape_logging (~> 1.7)
   graphiql-rails (~> 1.4.10)
+  graphlient (~> 0.4.0)
   graphql (~> 1.11.4)
   graphql-docs (~> 1.6.0)
   grpc (~> 1.30.2)

app/controllers/import/bulk_imports_controller.rb

@@ -6,13 +6,13 @@ class Import::BulkImportsController < ApplicationController
 
   feature_category :importers
 
-  rescue_from Gitlab::BulkImport::Client::ConnectionError, with: :bulk_import_connection_error
+  rescue_from BulkImports::Clients::Http::ConnectionError, with: :bulk_import_connection_error
 
   def configure
-    session[access_token_key] = params[access_token_key]&.strip
-    session[url_key] = params[url_key]
+    session[access_token_key] = configure_params[access_token_key]&.strip
+    session[url_key] = configure_params[url_key]
 
-    redirect_to status_import_bulk_import_url
+    redirect_to status_import_bulk_imports_url
   end
 
   def status
@@ -25,6 +25,12 @@ class Import::BulkImportsController < ApplicationController
     end
   end
 
+  def create
+    BulkImportService.new(current_user, create_params, credentials).execute
+
+    render json: :ok
+  end
+
   private
 
   def serialized_importable_data
@@ -40,16 +46,30 @@ class Import::BulkImportsController < ApplicationController
   end
 
   def client
-    @client ||= Gitlab::BulkImport::Client.new(
+    @client ||= BulkImports::Clients::Http.new(
       uri: session[url_key],
       token: session[access_token_key]
     )
   end
 
-  def import_params
+  def configure_params
     params.permit(access_token_key, url_key)
   end
 
+  def create_params
+    params.permit(:bulk_import, [*bulk_import_params])
+  end
+
+  def bulk_import_params
+    %i[
+      source_type
+      source_name
+      source_full_path
+      destination_name
+      destination_namespace
+    ]
+  end
+
   def ensure_group_import_enabled
     render_404 unless Feature.enabled?(:bulk_import)
   end
@@ -106,4 +126,11 @@ class Import::BulkImportsController < ApplicationController
     session[url_key] = nil
     session[access_token_key] = nil
   end
+
+  def credentials
+    {
+      url: session[url_key],
+      access_token: [access_token_key]
+    }
+  end
 end

app/models/bulk_import.rb

@@ -15,5 +15,20 @@ class BulkImport < ApplicationRecord
 
   state_machine :status, initial: :created do
     state :created, value: 0
+    state :started, value: 1
+    state :finished, value: 2
+    state :failed, value: -1
+
+    event :start do
+      transition created: :started
+    end
+
+    event :finish do
+      transition started: :finished
+    end
+
+    event :fail_op do
+      transition any => :failed
+    end
   end
 end

app/models/bulk_imports/entity.rb

@@ -38,6 +38,21 @@ class BulkImports::Entity < ApplicationRecord
 
   state_machine :status, initial: :created do
     state :created, value: 0
+    state :started, value: 1
+    state :finished, value: 2
+    state :failed, value: -1
+
+    event :start do
+      transition created: :started
+    end
+
+    event :finish do
+      transition started: :finished
+    end
+
+    event :fail_op do
+      transition any => :failed
+    end
   end
 
   private

app/services/bulk_import_service.rb

+# frozen_string_literal: true
+
+class BulkImportService
+  attr_reader :current_user, :params, :credentials
+
+  def initialize(current_user, params, credentials)
+    @current_user = current_user
+    @params = params
+    @credentials = credentials
+  end
+
+  def execute
+    bulk_import = create_bulk_import
+    bulk_import.start!
+
+    BulkImportWorker.perform_async(bulk_import.id)
+  end
+
+  private
+
+  def create_bulk_import
+    BulkImport.transaction do
+      bulk_import = BulkImport.create!(user: current_user, source_type: 'gitlab')
+      bulk_import.create_configuration!(credentials.slice(:url, :access_token))
+
+      params.each do |entity|
+        BulkImports::Entity.create!(
+          bulk_import: bulk_import,
+          source_type: entity[:source_type],
+          source_full_path: entity[:source_full_path],
+          destination_name: entity[:destination_name],
+          destination_namespace: entity[:destination_namespace]
+        )
+      end
+
+      bulk_import
+    end
+  end
+end

app/workers/all_queues.yml

@@ -1312,6 +1312,14 @@
   :weight: 1
   :idempotent:
   :tags: []
+- :name: bulk_import
+  :feature_category: :importers
+  :has_external_dependencies: true
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
 - :name: chat_notification
   :feature_category: :chatops
   :has_external_dependencies: true

app/workers/bulk_import_worker.rb

+# frozen_string_literal: true
+
+class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker
+  include ApplicationWorker
+
+  feature_category :importers
+
+  sidekiq_options retry: false, dead: false
+
+  worker_has_external_dependencies!
+
+  def perform(bulk_import_id)
+    bulk_import = BulkImport.find_by_id(bulk_import_id)
+
+    return unless bulk_import
+
+    bulk_import.entities.each do |entity|
+      entity.start!
+
+      BulkImports::Importers::GroupImporter.new(entity.id).execute
+
+      entity.finish!
+    end
+
+    bulk_import.finish!
+  end
+end

config/routes/import.rb

@@ -69,7 +69,7 @@ namespace :import do
     post :authorize
   end
 
-  resource :bulk_import, only: [:create] do
+  resource :bulk_imports, only: [:create] do
     post :configure
     get :status
   end

config/sidekiq_queues.yml

@@ -44,6 +44,8 @@
   - 3
 - - background_migration
   - 1
+- - bulk_import
+  - 1
 - - chaos
   - 2
 - - chat_notification

lib/bulk_imports/clients/graphql.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Clients
+    class Graphql
+      attr_reader :client
+
+      delegate :query, :parse, :execute, to: :client
+
+      def initialize(url: Gitlab::COM_URL, token: nil)
+        @url = Gitlab::Utils.append_path(url, '/api/graphql')
+        @token = token
+        @client = Graphlient::Client.new(
+          @url,
+          request_headers
+        )
+      end
+
+      def request_headers
+        return {} unless @token
+
+        {
+          headers: {
+            'Content-Type' => 'application/json',
+            'Authorization' => "Bearer #{@token}"
+          }
+        }
+      end
+    end
+  end
+end

lib/gitlab/bulk_import/client.rb → lib/bulk_imports/clients/http.rb

 # frozen_string_literal: true
 
-module Gitlab
-  module BulkImport
-    class Client
+module BulkImports
+  module Clients
+    class Http
       API_VERSION = 'v4'.freeze
       DEFAULT_PAGE = 1.freeze
       DEFAULT_PER_PAGE = 30.freeze

lib/bulk_imports/common/extractors/graphql_extractor.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Common
+    module Extractors
+      class GraphqlExtractor
+        def initialize(query)
+          @query = query[:query]
+          @query_string = @query.to_s
+          @variables = @query.variables
+        end
+
+        def extract(context)
+          @context = context
+
+          Enumerator.new do |yielder|
+            context.entities.each do |entity|
+              result = graphql_client.execute(parsed_query, query_variables(entity))
+
+              yielder << result.original_hash.deep_dup
+            end
+          end
+        end
+
+        private
+
+        def graphql_client
+          @graphql_client ||= BulkImports::Clients::Graphql.new(
+            url: @context.configuration.url,
+            token: @context.configuration.access_token
+          )
+        end
+
+        def parsed_query
+          @parsed_query ||= graphql_client.parse(@query.to_s)
+        end
+
+        def query_variables(entity)
+          return unless @variables
+
+          @variables.transform_values do |entity_attribute|
+            entity.public_send(entity_attribute) # rubocop:disable GitlabSecurity/PublicSend
+          end
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/common/transformers/graphql_cleaner_transformer.rb

+# frozen_string_literal: true
+
+# Cleanup GraphQL original response hash from unnecessary nesting
+# 1. Remove ['data']['group'] or ['data']['project'] hash nesting
+# 2. Remove ['edges'] & ['nodes'] array wrappings
+# 3. Remove ['node'] hash wrapping
+#
+# @example
+#   data = {"data"=>{"group"=> {
+#     "name"=>"test",
+#     "fullName"=>"test",
+#     "description"=>"test",
+#     "labels"=>{"edges"=>[{"node"=>{"title"=>"label1"}}, {"node"=>{"title"=>"label2"}}, {"node"=>{"title"=>"label3"}}]}}}}
+#
+#  BulkImports::Common::Transformers::GraphqlCleanerTransformer.new.transform(nil, data)
+#
+#  {"name"=>"test", "fullName"=>"test", "description"=>"test", "labels"=>[{"title"=>"label1"}, {"title"=>"label2"}, {"title"=>"label3"}]}
+module BulkImports
+  module Common
+    module Transformers
+      class GraphqlCleanerTransformer
+        EDGES = 'edges'
+        NODE = 'node'
+
+        def initialize(options = {})
+          @options = options
+        end
+
+        def transform(_, data)
+          return data unless data.is_a?(Hash)
+
+          data = data.dig('data', 'group') || data.dig('data', 'project') || data
+
+          clean_edges_and_nodes(data)
+        end
+
+        def clean_edges_and_nodes(data)
+          case data
+          when Array
+            data.map(&method(:clean_edges_and_nodes))
+          when Hash
+            if data.key?(NODE)
+              clean_edges_and_nodes(data[NODE])
+            else
+              data.transform_values { |value| clean_edges_and_nodes(value.try(:fetch, EDGES, value) || value) }
+            end
+          else
+            data
+          end
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/common/transformers/underscorify_keys_transformer.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Common
+    module Transformers
+      class UnderscorifyKeysTransformer
+        def initialize(options = {})
+          @options = options
+        end
+
+        def transform(_, data)
+          data.deep_transform_keys do |key|
+            key.to_s.underscore
+          end
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/groups/graphql/get_group_query.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Groups
+    module Graphql
+      module GetGroupQuery
+        extend self
+
+        def to_s
+          <<-'GRAPHQL'
+          query($full_path: ID!) {
+            group(fullPath: $full_path) {
+              name
+              path
+              fullPath
+              description
+              visibility
+              emailsDisabled
+              lfsEnabled
+              mentionsDisabled
+              projectCreationLevel
+              requestAccessEnabled
+              requireTwoFactorAuthentication
+              shareWithGroupLock
+              subgroupCreationLevel
+              twoFactorGracePeriod
+            }
+          }
+          GRAPHQL
+        end
+
+        def variables
+          { full_path: :source_full_path }
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/groups/loaders/group_loader.rb

+# frozen_string_literal: true
+
+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)
+
+          ::Groups::CreateService.new(context.current_user, data).execute
+        end
+
+        private
+
+        def user_can_create_group?(current_user, data)
+          if data['parent_id']
+            parent = Namespace.find_by_id(data['parent_id'])
+
+            Ability.allowed?(current_user, :create_subgroup, parent)
+          else
+            Ability.allowed?(current_user, :create_group)
+          end
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/groups/pipelines/group_pipeline.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Groups
+    module Pipelines
+      class GroupPipeline
+        include Pipeline
+
+        extractor Common::Extractors::GraphqlExtractor, query: Graphql::GetGroupQuery
+
+        transformer Common::Transformers::GraphqlCleanerTransformer
+        transformer Common::Transformers::UnderscorifyKeysTransformer
+        transformer Groups::Transformers::GroupAttributesTransformer
+
+        loader Groups::Loaders::GroupLoader
+      end
+    end
+  end
+end

lib/bulk_imports/groups/transformers/group_attributes_transformer.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Groups
+    module Transformers
+      class GroupAttributesTransformer
+        def initialize(options = {})
+          @options = options
+        end
+
+        def transform(context, data)
+          import_entity = find_by_full_path(data['full_path'], context.entities)
+
+          data
+            .then { |data| transform_name(import_entity, data) }
+            .then { |data| transform_path(import_entity, data) }
+            .then { |data| transform_full_path(data) }
+            .then { |data| transform_parent(context, import_entity, data) }
+            .then { |data| transform_visibility_level(data) }
+            .then { |data| transform_project_creation_level(data) }
+            .then { |data| transform_subgroup_creation_level(data) }
+        end
+
+        private
+
+        def transform_name(import_entity, data)
+          data['name'] = import_entity.destination_name
+          data
+        end
+
+        def transform_path(import_entity, data)
+          data['path'] = import_entity.destination_name.parameterize
+          data
+        end
+
+        def transform_full_path(data)
+          data.delete('full_path')
+          data
+        end
+
+        def transform_parent(context, import_entity, data)
+          current_user = context.current_user
+          namespace = Namespace.find_by_full_path(import_entity.destination_namespace)
+
+          return data if namespace == current_user.namespace
+
+          data['parent_id'] = namespace.id
+          data
+        end
+
+        def transform_visibility_level(data)
+          visibility = data['visibility']
+
+          return data unless visibility.present?
+
+          data['visibility_level'] = Gitlab::VisibilityLevel.string_options[visibility]
+          data.delete('visibility')
+          data
+        end
+
+        def transform_project_creation_level(data)
+          project_creation_level = data['project_creation_level']
+
+          return data unless project_creation_level.present?
+
+          data['project_creation_level'] = Gitlab::Access.project_creation_string_options[project_creation_level]
+          data
+        end
+
+        def transform_subgroup_creation_level(data)
+          subgroup_creation_level = data['subgroup_creation_level']
+
+          return data unless subgroup_creation_level.present?
+
+          data['subgroup_creation_level'] = Gitlab::Access.subgroup_creation_string_options[subgroup_creation_level]
+          data
+        end
+
+        def find_by_full_path(full_path, entities)
+          entities.find { |entity| entity.source_full_path == full_path }
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/importers/group_importer.rb

+# frozen_string_literal: true
+
+# Imports a top level group into a destination
+# Optionally imports into parent group
+# Entity must be of type: 'group' & have parent_id: nil
+# Subgroups not handled yet
+module BulkImports
+  module Importers
+    class GroupImporter
+      def initialize(entity_id)
+        @entity_id = entity_id
+      end
+
+      def execute
+        return if entity.parent
+
+        bulk_import = entity.bulk_import
+        configuration = bulk_import.configuration
+
+        context = BulkImports::Pipeline::Context.new(
+          current_user: bulk_import.user,
+          entities: [entity],
+          configuration: configuration
+        )
+
+        BulkImports::Groups::Pipelines::GroupPipeline.new.run(context)
+      end
+
+      def entity
+        @entity ||= BulkImports::Entity.find(@entity_id)
+      end
+    end
+  end
+end

lib/bulk_imports/pipeline.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Pipeline
+    extend ActiveSupport::Concern
+
+    included do
+      include Attributes
+      include Runner
+    end
+  end
+end

lib/bulk_imports/pipeline/attributes.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Pipeline
+    module Attributes
+      extend ActiveSupport::Concern
+      include Gitlab::ClassAttributes
+
+      class_methods do
+        def extractor(klass, options = nil)
+          add_attribute(:extractors, klass, options)
+        end
+
+        def transformer(klass, options = nil)
+          add_attribute(:transformers, klass, options)
+        end
+
+        def loader(klass, options = nil)
+          add_attribute(:loaders, klass, options)
+        end
+
+        def add_attribute(sym, klass, options)
+          class_attributes[sym] ||= []
+          class_attributes[sym] << { klass: klass, options: options }
+        end
+
+        def extractors
+          class_attributes[:extractors]
+        end
+
+        def transformers
+          class_attributes[:transformers]
+        end
+
+        def loaders
+          class_attributes[:loaders]
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/pipeline/context.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Pipeline
+    class Context
+      include Gitlab::Utils::LazyAttributes
+
+      Attribute = Struct.new(:name, :type)
+
+      PIPELINE_ATTRIBUTES = [
+        Attribute.new(:current_user, User),
+        Attribute.new(:entities, Array),
+        Attribute.new(:configuration, ::BulkImports::Configuration)
+      ].freeze
+
+      def initialize(args)
+        assign_attributes(args)
+      end
+
+      private
+
+      PIPELINE_ATTRIBUTES.each do |attr|
+        lazy_attr_reader attr.name, type: attr.type
+      end
+
+      def assign_attributes(values)
+        values.slice(*PIPELINE_ATTRIBUTES.map(&:name)).each do |name, value|
+          instance_variable_set("@#{name}", value)
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/pipeline/runner.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Pipeline
+    module Runner
+      extend ActiveSupport::Concern
+
+      included do
+        attr_reader :extractors, :transformers, :loaders
+
+        def initialize
+          @extractors = self.class.extractors.map(&method(:instantiate))
+          @transformers = self.class.transformers.map(&method(:instantiate))
+          @loaders = self.class.loaders.map(&method(:instantiate))
+
+          super
+        end
+
+        def run(context)
+          extractors.each do |extractor|
+            extractor.extract(context).each do |entry|
+              transformers.each do |transformer|
+                entry = transformer.transform(context, entry)
+              end
+
+              loaders.each do |loader|
+                loader.load(context, entry)
+              end
+            end
+          end
+        end
+
+        def instantiate(class_config)
+          class_config[:klass].new(class_config[:options])
+        end
+      end
+    end
+  end
+end

spec/controllers/import/bulk_imports_controller_spec.rb

@@ -24,7 +24,7 @@ RSpec.describe Import::BulkImportsController do
             expect(session[:bulk_import_gitlab_url]).to be_nil
 
             expect(response).to have_gitlab_http_status(:found)
-            expect(response).to redirect_to(status_import_bulk_import_url)
+            expect(response).to redirect_to(status_import_bulk_imports_url)
           end
         end
 
@@ -37,7 +37,7 @@ RSpec.describe Import::BulkImportsController do
           expect(session[:bulk_import_gitlab_access_token]).to eq(token)
           expect(session[:bulk_import_gitlab_url]).to eq(url)
           expect(response).to have_gitlab_http_status(:found)
-          expect(response).to redirect_to(status_import_bulk_import_url)
+          expect(response).to redirect_to(status_import_bulk_imports_url)
         end
 
         it 'strips access token with spaces' do
@@ -46,12 +46,12 @@ RSpec.describe Import::BulkImportsController do
           post :configure, params: { bulk_import_gitlab_access_token: "  #{token} " }
 
           expect(session[:bulk_import_gitlab_access_token]).to eq(token)
-          expect(controller).to redirect_to(status_import_bulk_import_url)
+          expect(controller).to redirect_to(status_import_bulk_imports_url)
         end
       end
 
       describe 'GET status' do
-        let(:client) { Gitlab::BulkImport::Client.new(uri: 'http://gitlab.example', token: 'token') }
+        let(:client) { BulkImports::Clients::Http.new(uri: 'http://gitlab.example', token: 'token') }
 
         describe 'serialized group data' do
           let(:client_response) do
@@ -111,7 +111,7 @@ RSpec.describe Import::BulkImportsController do
         context 'when connection error occurs' do
           before do
             allow(controller).to receive(:client).and_return(client)
-            allow(client).to receive(:get).and_raise(Gitlab::BulkImport::Client::ConnectionError)
+            allow(client).to receive(:get).and_raise(BulkImports::Clients::Http::ConnectionError)
           end
 
           it 'returns 422' do
@@ -128,9 +128,21 @@ RSpec.describe Import::BulkImportsController do
           end
         end
       end
+
+      describe 'POST create' do
+        it 'executes BulkImportService' do
+          expect_next_instance_of(BulkImportService) do |service|
+            expect(service).to receive(:execute)
+          end
+
+          post :create
+
+          expect(response).to have_gitlab_http_status(:ok)
+        end
+      end
     end
 
-    context 'when gitlab_api_imports feature flag is disabled' do
+    context 'when bulk_import feature flag is disabled' do
       before do
         stub_feature_flags(bulk_import: false)
       end

spec/factories/bulk_import.rb

@@ -4,5 +4,21 @@ FactoryBot.define do
   factory :bulk_import, class: 'BulkImport' do
     user
     source_type { :gitlab }
+
+    trait :created do
+      status { 0 }
+    end
+
+    trait :started do
+      status { 1 }
+    end
+
+    trait :finished do
+      status { 2 }
+    end
+
+    trait :failed do
+      status { -1 }
+    end
   end
 end

spec/factories/bulk_import/entities.rb

@@ -17,5 +17,25 @@ FactoryBot.define do
     trait(:project_entity) do
       source_type { :project_entity }
     end
+
+    trait :created do
+      status { 0 }
+    end
+
+    trait :started do
+      status { 1 }
+
+      sequence(:jid) { |n| "bulk_import_entity_#{n}" }
+    end
+
+    trait :finished do
+      status { 2 }
+
+      sequence(:jid) { |n| "bulk_import_entity_#{n}" }
+    end
+
+    trait :failed do
+      status { -1 }
+    end
   end
 end

spec/factories/import_configurations.rb

+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :bulk_import_configuration, class: 'BulkImports::Configuration' do
+    association :bulk_import, factory: :bulk_import
+
+    url { 'https://gitlab.example' }
+    access_token { 'token' }
+  end
+end

spec/lib/gitlab/bulk_import/client_spec.rb → spec/lib/bulk_imports/clients/http_spec.rb

@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe Gitlab::BulkImport::Client do
+RSpec.describe BulkImports::Clients::Http do
   include ImportSpecHelper
 
   let(:uri) { 'http://gitlab.example' }

spec/lib/bulk_imports/common/extractors/graphql_extractor_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Common::Extractors::GraphqlExtractor do
+  let(:graphql_client) { instance_double(BulkImports::Clients::Graphql) }
+  let(:import_entity) { create(:bulk_import_entity) }
+  let(:response) { double(original_hash: { foo: :bar }) }
+  let(:query) { { query: double(to_s: 'test', variables: {}) } }
+  let(:context) do
+    instance_double(
+      BulkImports::Pipeline::Context,
+      entities: [import_entity]
+    )
+  end
+
+  subject { described_class.new(query) }
+
+  before do
+    allow(subject).to receive(:graphql_client).and_return(graphql_client)
+    allow(graphql_client).to receive(:parse)
+  end
+
+  describe '#extract' do
+    before do
+      allow(subject).to receive(:query_variables).and_return({})
+      allow(graphql_client).to receive(:execute).and_return(response)
+    end
+
+    it 'returns an enumerator with fetched results' do
+      response = subject.extract(context)
+
+      expect(response).to be_instance_of(Enumerator)
+      expect(response.first).to eq({ foo: :bar })
+    end
+  end
+
+  describe 'query variables' do
+    before do
+      allow(graphql_client).to receive(:execute).and_return(response)
+    end
+
+    context 'when variables are present' do
+      let(:query) { { query: double(to_s: 'test', variables: { full_path: :source_full_path }) } }
+
+      it 'builds graphql query variables for import entity' do
+        expected_variables = { full_path: import_entity.source_full_path }
+
+        expect(graphql_client).to receive(:execute).with(anything, expected_variables)
+
+        subject.extract(context).first
+      end
+    end
+
+    context 'when no variables are present' do
+      let(:query) { { query: double(to_s: 'test', variables: nil) } }
+
+      it 'returns empty hash' do
+        expect(graphql_client).to receive(:execute).with(anything, nil)
+
+        subject.extract(context).first
+      end
+    end
+
+    context 'when variables are empty hash' do
+      let(:query) { { query: double(to_s: 'test', variables: {}) } }
+
+      it 'makes graphql request with empty hash' do
+        expect(graphql_client).to receive(:execute).with(anything, {})
+
+        subject.extract(context).first
+      end
+    end
+  end
+end

spec/lib/bulk_imports/common/transformers/graphql_cleaner_transformer_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Common::Transformers::GraphqlCleanerTransformer do
+  describe '#transform' do
+    let_it_be(:expected_output) do
+      {
+        'name' => 'test',
+        'fullName' => 'test',
+        'description' => 'test',
+        'labels' => [
+          { 'title' => 'label1' },
+          { 'title' => 'label2' },
+          { 'title' => 'label3' }
+        ]
+      }
+    end
+
+    it 'deep cleans hash from GraphQL keys' do
+      data = {
+        'data' => {
+          'group' => {
+            'name' => 'test',
+            'fullName' => 'test',
+            'description' => 'test',
+            'labels' => {
+              'edges' => [
+                { 'node' => { 'title' => 'label1' } },
+                { 'node' => { 'title' => 'label2' } },
+                { 'node' => { 'title' => 'label3' } }
+              ]
+            }
+          }
+        }
+      }
+
+      transformed_data = described_class.new.transform(nil, data)
+
+      expect(transformed_data).to eq(expected_output)
+    end
+
+    context 'when data does not have data/group nesting' do
+      it 'deep cleans hash from GraphQL keys' do
+        data = {
+          'name' => 'test',
+          'fullName' => 'test',
+          'description' => 'test',
+          'labels' => {
+            'edges' => [
+              { 'node' => { 'title' => 'label1' } },
+              { 'node' => { 'title' => 'label2' } },
+              { 'node' => { 'title' => 'label3' } }
+            ]
+          }
+        }
+
+        transformed_data = described_class.new.transform(nil, data)
+
+        expect(transformed_data).to eq(expected_output)
+      end
+    end
+
+    context 'when data is not a hash' do
+      it 'does not perform transformation' do
+        data = 'test'
+
+        transformed_data = described_class.new.transform(nil, data)
+
+        expect(transformed_data).to eq(data)
+      end
+    end
+
+    context 'when nested data is not an array or hash' do
+      it 'only removes top level data/group keys' do
+        data = {
+          'data' => {
+            'group' => 'test'
+          }
+        }
+
+        transformed_data = described_class.new.transform(nil, data)
+
+        expect(transformed_data).to eq('test')
+      end
+    end
+  end
+end

spec/lib/bulk_imports/common/transformers/underscorify_keys_transformer_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Common::Transformers::UnderscorifyKeysTransformer do
+  describe '#transform' do
+    it 'deep underscorifies hash keys' do
+      data = {
+        'fullPath' => 'Foo',
+        'snakeKeys' => {
+          'snakeCaseKey' => 'Bar',
+          'moreKeys' => {
+            'anotherSnakeCaseKey' => 'Test'
+          }
+        }
+      }
+
+      transformed_data = described_class.new.transform(nil, data)
+
+      expect(transformed_data).to have_key('full_path')
+      expect(transformed_data).to have_key('snake_keys')
+      expect(transformed_data['snake_keys']).to have_key('snake_case_key')
+      expect(transformed_data['snake_keys']).to have_key('more_keys')
+      expect(transformed_data.dig('snake_keys', 'more_keys')).to have_key('another_snake_case_key')
+    end
+  end
+end

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

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Groups::Loaders::GroupLoader do
+  describe '#load' do
+    let(:user) { create(:user) }
+    let(:data) { { foo: :bar } }
+    let(:service_double) { instance_double(::Groups::CreateService) }
+    let(:context) do
+      instance_double(
+        BulkImports::Pipeline::Context,
+        current_user: user
+      )
+    end
+
+    subject { described_class.new }
+
+    context 'when user can create group' do
+      shared_examples 'calls Group Create Service to create a new group' do
+        it 'calls Group Create Service to create a new group' do
+          expect(::Groups::CreateService).to receive(:new).with(context.current_user, data).and_return(service_double)
+          expect(service_double).to receive(:execute)
+
+          subject.load(context, data)
+        end
+      end
+
+      context 'when there is no parent group' do
+        before do
+          allow(Ability).to receive(:allowed?).with(user, :create_group).and_return(true)
+        end
+
+        include_examples 'calls Group Create Service to create a new group'
+      end
+
+      context 'when there is parent group' do
+        let(:parent) { create(:group) }
+        let(:data) { { 'parent_id' => parent.id } }
+
+        before do
+          allow(Ability).to receive(:allowed?).with(user, :create_subgroup, parent).and_return(true)
+        end
+
+        include_examples 'calls Group Create Service to create a new group'
+      end
+    end
+
+    context 'when user cannot create group' do
+      shared_examples 'does not create new group' do
+        it 'does not create new group' do
+          expect(::Groups::CreateService).not_to receive(:new)
+
+          subject.load(context, data)
+        end
+      end
+
+      context 'when there is no parent group' do
+        before do
+          allow(Ability).to receive(:allowed?).with(user, :create_group).and_return(false)
+        end
+
+        include_examples 'does not create new group'
+      end
+
+      context 'when there is parent group' do
+        let(:parent) { create(:group) }
+        let(:data) { { 'parent_id' => parent.id } }
+
+        before do
+          allow(Ability).to receive(:allowed?).with(user, :create_subgroup, parent).and_return(false)
+        end
+
+        include_examples 'does not create new group'
+      end
+    end
+  end
+end

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

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Groups::Pipelines::GroupPipeline do
+  describe '#run' do
+    let(:user) { create(:user) }
+    let(:parent) { create(:group) }
+    let(:entity) do
+      instance_double(
+        BulkImports::Entity,
+        source_full_path: 'source/full/path',
+        destination_name: 'My Destination Group',
+        destination_namespace: parent.full_path
+      )
+    end
+
+    let(:entities) { [entity] }
+    let(:context) do
+      instance_double(
+        BulkImports::Pipeline::Context,
+        current_user: user,
+        entities: entities
+      )
+    end
+
+    let(:group_data) do
+      {
+        'data' => {
+          'group' => {
+            'name' => 'source_name',
+            'fullPath' => 'source/full/path',
+            'visibility' => 'private',
+            'projectCreationLevel' => 'developer',
+            'subgroupCreationLevel' => 'maintainer',
+            'description' => 'Group Description',
+            'emailsDisabled' => true,
+            'lfsEnabled' => false,
+            'mentionsDisabled' => true
+          }
+        }
+      }
+    end
+
+    subject { described_class.new }
+
+    before do
+      allow_next_instance_of(BulkImports::Common::Extractors::GraphqlExtractor) do |extractor|
+        allow(extractor).to receive(:extract).and_return([group_data])
+      end
+
+      parent.add_owner(user)
+    end
+
+    it 'imports new group into destination group' do
+      group_path = 'my-destination-group'
+
+      subject.run(context)
+
+      imported_group = Group.find_by_path(group_path)
+
+      expect(imported_group).not_to be_nil
+      expect(imported_group.parent).to eq(parent)
+      expect(imported_group.path).to eq(group_path)
+      expect(imported_group.description).to eq(group_data.dig('data', 'group', 'description'))
+      expect(imported_group.visibility).to eq(group_data.dig('data', 'group', 'visibility'))
+      expect(imported_group.project_creation_level).to eq(Gitlab::Access.project_creation_string_options[group_data.dig('data', 'group', 'projectCreationLevel')])
+      expect(imported_group.subgroup_creation_level).to eq(Gitlab::Access.subgroup_creation_string_options[group_data.dig('data', 'group', 'subgroupCreationLevel')])
+      expect(imported_group.lfs_enabled?).to eq(group_data.dig('data', 'group', 'lfsEnabled'))
+      expect(imported_group.emails_disabled?).to eq(group_data.dig('data', 'group', 'emailsDisabled'))
+      expect(imported_group.mentions_disabled?).to eq(group_data.dig('data', 'group', 'mentionsDisabled'))
+    end
+  end
+
+  describe 'pipeline parts' do
+    it { expect(described_class).to include_module(BulkImports::Pipeline) }
+    it { expect(described_class).to include_module(BulkImports::Pipeline::Attributes) }
+    it { expect(described_class).to include_module(BulkImports::Pipeline::Runner) }
+
+    it 'has extractors' do
+      expect(described_class.extractors)
+        .to contain_exactly(
+          {
+            klass: BulkImports::Common::Extractors::GraphqlExtractor,
+            options: {
+              query: BulkImports::Groups::Graphql::GetGroupQuery
+            }
+          }
+        )
+    end
+
+    it 'has transformers' do
+      expect(described_class.transformers)
+        .to contain_exactly(
+          { klass: BulkImports::Common::Transformers::GraphqlCleanerTransformer, options: nil },
+          { klass: BulkImports::Common::Transformers::UnderscorifyKeysTransformer, options: nil },
+          { klass: BulkImports::Groups::Transformers::GroupAttributesTransformer, options: nil })
+    end
+
+    it 'has loaders' do
+      expect(described_class.loaders).to contain_exactly({ klass: BulkImports::Groups::Loaders::GroupLoader, options: nil })
+    end
+  end
+end

spec/lib/bulk_imports/groups/transformers/group_attributes_transformer_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Groups::Transformers::GroupAttributesTransformer do
+  describe '#transform' do
+    let(:user) { create(:user) }
+    let(:parent) { create(:group) }
+    let(:group) { create(:group, name: 'My Source Group', parent: parent) }
+    let(:entity) do
+      instance_double(
+        BulkImports::Entity,
+        source_full_path: 'source/full/path',
+        destination_name: group.name,
+        destination_namespace: parent.full_path
+      )
+    end
+
+    let(:entities) { [entity] }
+    let(:context) do
+      instance_double(
+        BulkImports::Pipeline::Context,
+        current_user: user,
+        entities: entities
+      )
+    end
+
+    let(:data) do
+      {
+        'name' => 'source_name',
+        'full_path' => 'source/full/path',
+        'visibility' => 'private',
+        'project_creation_level' => 'developer',
+        'subgroup_creation_level' => 'maintainer'
+      }
+    end
+
+    subject { described_class.new }
+
+    it 'transforms name to destination name' do
+      transformed_data = subject.transform(context, data)
+
+      expect(transformed_data['name']).not_to eq('source_name')
+      expect(transformed_data['name']).to eq(group.name)
+    end
+
+    it 'removes full path' do
+      transformed_data = subject.transform(context, data)
+
+      expect(transformed_data).not_to have_key('full_path')
+    end
+
+    it 'transforms path to parameterized name' do
+      transformed_data = subject.transform(context, data)
+
+      expect(transformed_data['path']).to eq(group.name.parameterize)
+    end
+
+    it 'transforms visibility level' do
+      visibility = data['visibility']
+      transformed_data = subject.transform(context, data)
+
+      expect(transformed_data).not_to have_key('visibility')
+      expect(transformed_data['visibility_level']).to eq(Gitlab::VisibilityLevel.string_options[visibility])
+    end
+
+    it 'transforms project creation level' do
+      level = data['project_creation_level']
+      transformed_data = subject.transform(context, data)
+
+      expect(transformed_data['project_creation_level']).to eq(Gitlab::Access.project_creation_string_options[level])
+    end
+
+    it 'transforms subgroup creation level' do
+      level = data['subgroup_creation_level']
+      transformed_data = subject.transform(context, data)
+
+      expect(transformed_data['subgroup_creation_level']).to eq(Gitlab::Access.subgroup_creation_string_options[level])
+    end
+
+    describe 'parent group transformation' do
+      it 'sets parent id' do
+        transformed_data = subject.transform(context, data)
+
+        expect(transformed_data['parent_id']).to eq(parent.id)
+      end
+
+      context 'when destination namespace is user namespace' do
+        let(:entity) do
+          instance_double(
+            BulkImports::Entity,
+            source_full_path: 'source/full/path',
+            destination_name: group.name,
+            destination_namespace: user.namespace.full_path
+          )
+        end
+
+        it 'does not set parent id' do
+          transformed_data = subject.transform(context, data)
+
+          expect(transformed_data).not_to have_key('parent_id')
+        end
+      end
+    end
+  end
+end

spec/lib/bulk_imports/importers/group_importer_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Importers::GroupImporter do
+  let(:user) { create(:user) }
+  let(:bulk_import) { create(:bulk_import) }
+  let(:bulk_import_entity) { create(:bulk_import_entity, bulk_import: bulk_import) }
+  let(:bulk_import_configuration) { create(:bulk_import_configuration, bulk_import: bulk_import) }
+  let(:context) do
+    instance_double(
+      BulkImports::Pipeline::Context,
+      current_user: user,
+      entities: [bulk_import_entity],
+      configuration: bulk_import_configuration
+    )
+  end
+
+  subject { described_class.new(bulk_import_entity.id) }
+
+  describe '#execute' do
+    before do
+      allow(BulkImports::Pipeline::Context).to receive(:new).and_return(context)
+    end
+
+    context 'when import entity does not have parent' do
+      it 'executes GroupPipeline' do
+        expect_next_instance_of(BulkImports::Groups::Pipelines::GroupPipeline) do |pipeline|
+          expect(pipeline).to receive(:run).with(context)
+        end
+
+        subject.execute
+      end
+    end
+
+    context 'when import entity has parent' do
+      let(:bulk_import_entity_parent) { create(:bulk_import_entity, bulk_import: bulk_import) }
+      let(:bulk_import_entity) { create(:bulk_import_entity, bulk_import: bulk_import, parent: bulk_import_entity_parent) }
+
+      it 'does not execute GroupPipeline' do
+        expect(BulkImports::Groups::Pipelines::GroupPipeline).not_to receive(:new)
+
+        subject.execute
+      end
+    end
+  end
+end

spec/lib/bulk_imports/pipeline/attributes_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Pipeline::Attributes do
+  describe 'pipeline attributes' do
+    before do
+      stub_const('BulkImports::Extractor', Class.new)
+      stub_const('BulkImports::Transformer', Class.new)
+      stub_const('BulkImports::Loader', Class.new)
+
+      klass = Class.new do
+        include BulkImports::Pipeline::Attributes
+
+        extractor BulkImports::Extractor, { foo: :bar }
+        transformer BulkImports::Transformer, { foo: :bar }
+        loader BulkImports::Loader, { foo: :bar }
+      end
+
+      stub_const('BulkImports::MyPipeline', klass)
+    end
+
+    describe 'getters' do
+      it 'retrieves class attributes' do
+        expect(BulkImports::MyPipeline.extractors).to contain_exactly({ klass: BulkImports::Extractor, options: { foo: :bar } })
+        expect(BulkImports::MyPipeline.transformers).to contain_exactly({ klass: BulkImports::Transformer, options: { foo: :bar } })
+        expect(BulkImports::MyPipeline.loaders).to contain_exactly({ klass: BulkImports::Loader, options: { foo: :bar } })
+      end
+    end
+
+    describe 'setters' do
+      it 'sets class attributes' do
+        klass = Class.new
+        options = { test: :test }
+
+        BulkImports::MyPipeline.extractor(klass, options)
+        BulkImports::MyPipeline.transformer(klass, options)
+        BulkImports::MyPipeline.loader(klass, options)
+
+        expect(BulkImports::MyPipeline.extractors)
+          .to contain_exactly(
+            { klass: BulkImports::Extractor, options: { foo: :bar } },
+            { klass: klass, options: options })
+
+        expect(BulkImports::MyPipeline.transformers)
+          .to contain_exactly(
+            { klass: BulkImports::Transformer, options: { foo: :bar } },
+            { klass: klass, options: options })
+
+        expect(BulkImports::MyPipeline.loaders)
+          .to contain_exactly(
+            { klass: BulkImports::Loader, options: { foo: :bar } },
+            { klass: klass, options: options })
+      end
+    end
+  end
+end

spec/lib/bulk_imports/pipeline/context_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Pipeline::Context do
+  describe '#initialize' do
+    it 'initializes with permitted attributes' do
+      args = {
+        current_user: create(:user),
+        entities: [],
+        configuration: create(:bulk_import_configuration)
+      }
+
+      context = described_class.new(args)
+
+      args.each do |k, v|
+        expect(context.public_send(k)).to eq(v)
+      end
+    end
+
+    context 'when invalid argument is passed' do
+      it 'raises NoMethodError' do
+        expect { described_class.new(test: 'test').test }.to raise_exception(NoMethodError)
+      end
+    end
+  end
+end

spec/lib/bulk_imports/pipeline/runner_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Pipeline::Runner do
+  describe 'pipeline runner' do
+    before do
+      extractor = Class.new do
+        def initialize(options = {}); end
+
+        def extract(context); end
+      end
+
+      transformer = Class.new do
+        def initialize(options = {}); end
+
+        def transform(context, entry); end
+      end
+
+      loader = Class.new do
+        def initialize(options = {}); end
+
+        def load(context, entry); end
+      end
+
+      stub_const('BulkImports::Extractor', extractor)
+      stub_const('BulkImports::Transformer', transformer)
+      stub_const('BulkImports::Loader', loader)
+
+      pipeline = Class.new do
+        include BulkImports::Pipeline
+
+        extractor BulkImports::Extractor
+        transformer BulkImports::Transformer
+        loader BulkImports::Loader
+      end
+
+      stub_const('BulkImports::MyPipeline', pipeline)
+    end
+
+    it 'runs pipeline extractor, transformer, loader' do
+      context = instance_double(BulkImports::Pipeline::Context)
+      entries = [{ foo: :bar }]
+
+      expect_next_instance_of(BulkImports::Extractor) do |extractor|
+        expect(extractor).to receive(:extract).with(context).and_return(entries)
+      end
+
+      expect_next_instance_of(BulkImports::Transformer) do |transformer|
+        expect(transformer).to receive(:transform).with(context, entries.first).and_return(entries.first)
+      end
+
+      expect_next_instance_of(BulkImports::Loader) do |loader|
+        expect(loader).to receive(:load).with(context, entries.first)
+      end
+
+      BulkImports::MyPipeline.new.run(context)
+    end
+  end
+end

spec/services/bulk_import_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImportService do
+  let(:user) { create(:user) }
+  let(:credentials) { { url: 'http://gitlab.example', access_token: 'token' } }
+  let(:params) do
+    [
+      {
+        source_type: 'group_entity',
+        source_full_path: 'full/path/to/group1',
+        destination_name: 'destination group 1',
+        destination_namespace: 'full/path/to/destination1'
+      },
+      {
+        source_type: 'group_entity',
+        source_full_path: 'full/path/to/group2',
+        destination_name: 'destination group 2',
+        destination_namespace: 'full/path/to/destination2'
+      },
+      {
+        source_type: 'project_entity',
+        source_full_path: 'full/path/to/project1',
+        destination_name: 'destination project 1',
+        destination_namespace: 'full/path/to/destination1'
+      }
+    ]
+  end
+
+  subject { described_class.new(user, params, credentials) }
+
+  describe '#execute' do
+    it 'creates bulk import' do
+      expect { subject.execute }.to change { BulkImport.count }.by(1)
+    end
+
+    it 'creates bulk import entities' do
+      expect { subject.execute }.to change { BulkImports::Entity.count }.by(3)
+    end
+
+    it 'creates bulk import configuration' do
+      expect { subject.execute }.to change { BulkImports::Configuration.count }.by(1)
+    end
+
+    it 'updates bulk import state' do
+      expect_next_instance_of(BulkImport) do |bulk_import|
+        expect(bulk_import).to receive(:start!)
+      end
+
+      subject.execute
+    end
+
+    it 'enqueues BulkImportWorker' do
+      expect(BulkImportWorker).to receive(:perform_async)
+
+      subject.execute
+    end
+  end
+end

spec/workers/bulk_import_worker_spec.rb

+#  frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImportWorker do
+  let!(:bulk_import) { create(:bulk_import, :started) }
+  let!(:entity) { create(:bulk_import_entity, bulk_import: bulk_import) }
+  let(:importer) { double(execute: nil) }
+
+  subject { described_class.new }
+
+  describe '#perform' do
+    before do
+      allow(BulkImports::Importers::GroupImporter).to receive(:new).and_return(importer)
+    end
+
+    it 'executes Group Importer' do
+      expect(importer).to receive(:execute)
+
+      subject.perform(bulk_import.id)
+    end
+
+    it 'updates bulk import and entity state' do
+      subject.perform(bulk_import.id)
+
+      expect(bulk_import.reload.human_status_name).to eq('finished')
+      expect(entity.reload.human_status_name).to eq('finished')
+    end
+
+    context 'when bulk import could not be found' do
+      it 'does nothing' do
+        expect(bulk_import).not_to receive(:top_level_groups)
+        expect(bulk_import).not_to receive(:finish!)
+
+        subject.perform(non_existing_record_id)
+      end
+    end
+  end
+end