BulkImports: Migrate Group Membership

Group membership is migrated for existing users in the destination
Gitlab instance. The users are mapped if they have a public email on the
source instance that matches either a primary or secondary email on the
destination instance.

Related to: https://gitlab.com/gitlab-org/gitlab/-/issues/299415

changelogs/unreleased/kassio-bulkimports-import-group-membership.yml

+---
+title: 'BulkImports: Migrate Group Membership'
+merge_request: 53083
+author:
+type: added

ee/spec/lib/ee/bulk_imports/importers/group_importer_spec.rb

@@ -18,9 +18,10 @@ RSpec.describe BulkImports::Importers::GroupImporter do
   describe '#execute' do
     it "starts the entity and run its pipelines" do
       expect_to_run_pipeline BulkImports::Groups::Pipelines::GroupPipeline, context: context
+      expect_to_run_pipeline BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline, context: context
+      expect_to_run_pipeline BulkImports::Groups::Pipelines::MembersPipeline, context: context
       expect_to_run_pipeline BulkImports::Groups::Pipelines::LabelsPipeline, context: context
       expect_to_run_pipeline EE::BulkImports::Groups::Pipelines::EpicsPipeline, context: context
-      expect_to_run_pipeline BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline, context: context
 
       subject.execute
 

lib/bulk_imports/groups/graphql/get_members_query.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Groups
+    module Graphql
+      module GetMembersQuery
+        extend self
+        def to_s
+          <<-'GRAPHQL'
+          query($full_path: ID!, $cursor: String) {
+            group(fullPath: $full_path) {
+              group_members: groupMembers(relations: DIRECT, first: 100, after: $cursor) {
+                page_info: pageInfo {
+                  end_cursor: endCursor
+                  has_next_page: hasNextPage
+                }
+                nodes {
+                  created_at: createdAt
+                  updated_at: updatedAt
+                  expires_at: expiresAt
+                  access_level: accessLevel {
+                    integer_value: integerValue
+                  }
+                  user {
+                    public_email: publicEmail
+                  }
+                }
+              }
+            }
+          }
+          GRAPHQL
+        end
+
+        def variables(entity)
+          {
+            full_path: entity.source_full_path,
+            cursor: entity.next_page_for(:group_members)
+          }
+        end
+
+        def base_path
+          %w[data group group_members]
+        end
+
+        def data_path
+          base_path << 'nodes'
+        end
+
+        def page_info_path
+          base_path << 'page_info'
+        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/members_pipeline.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Groups
+    module Pipelines
+      class MembersPipeline
+        include Pipeline
+
+        extractor BulkImports::Common::Extractors::GraphqlExtractor,
+          query: BulkImports::Groups::Graphql::GetMembersQuery
+
+        transformer Common::Transformers::ProhibitedAttributesTransformer
+        transformer BulkImports::Groups::Transformers::MemberAttributesTransformer
+
+        loader BulkImports::Groups::Loaders::MembersLoader
+
+        def after_run(context, extracted_data)
+          context.entity.update_tracker_for(
+            relation: :group_members,
+            has_next_page: extracted_data.has_next_page?,
+            next_page: extracted_data.next_page
+          )
+
+          if extracted_data.has_next_page?
+            run(context)
+          end
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/groups/transformers/member_attributes_transformer.rb

+# frozen_string_literal: true
+
+module BulkImports
+  module Groups
+    module Transformers
+      class MemberAttributesTransformer
+        def initialize(*); end
+
+        def transform(context, data)
+          data
+            .then { |data| add_user(data) }
+            .then { |data| add_access_level(data) }
+            .then { |data| add_author(data, context) }
+        end
+
+        private
+
+        def add_user(data)
+          user = find_user(data&.dig('user', 'public_email'))
+
+          return unless user
+
+          data
+            .except('user')
+            .merge('user_id' => user.id)
+        end
+
+        def find_user(email)
+          return unless email
+
+          User.find_by_any_email(email, confirmed: true)
+        end
+
+        def add_access_level(data)
+          access_level = data&.dig('access_level', 'integer_value')
+
+          return unless valid_access_level?(access_level)
+
+          data.merge('access_level' => access_level)
+        end
+
+        def valid_access_level?(access_level)
+          Gitlab::Access
+            .options_with_owner
+            .value?(access_level)
+        end
+
+        def add_author(data, context)
+          return unless data
+
+          data.merge('created_by_id' => context.current_user.id)
+        end
+      end
+    end
+  end
+end

lib/bulk_imports/importers/group_importer.rb

@@ -23,6 +23,7 @@ module BulkImports
         [
           BulkImports::Groups::Pipelines::GroupPipeline,
           BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline,
+          BulkImports::Groups::Pipelines::MembersPipeline,
           BulkImports::Groups::Pipelines::LabelsPipeline
         ]
       end

spec/lib/bulk_imports/groups/graphql/get_members_query_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Groups::Graphql::GetMembersQuery do
+  it 'has a valid query' do
+    entity = create(:bulk_import_entity)
+
+    query = GraphQL::Query.new(
+      GitlabSchema,
+      described_class.to_s,
+      variables: described_class.variables(entity)
+    )
+    result = GitlabSchema.static_validator.validate(query)
+
+    expect(result[:errors]).to be_empty
+  end
+
+  describe '#data_path' do
+    it 'returns data path' do
+      expected = %w[data group group_members nodes]
+
+      expect(described_class.data_path).to eq(expected)
+    end
+  end
+
+  describe '#page_info_path' do
+    it 'returns pagination information path' do
+      expected = %w[data group group_members page_info]
+
+      expect(described_class.page_info_path).to eq(expected)
+    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/members_pipeline_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Groups::Pipelines::MembersPipeline do
+  let_it_be(:member_user1) { create(:user, email: 'email1@email.com') }
+  let_it_be(:member_user2) { create(:user, email: 'email2@email.com') }
+
+  let_it_be(:user) { create(:user) }
+  let_it_be(:group) { create(:group) }
+  let_it_be(:cursor) { 'cursor' }
+  let_it_be(:bulk_import) { create(:bulk_import, user: user) }
+  let_it_be(:entity) { create(:bulk_import_entity, bulk_import: bulk_import, group: group) }
+  let_it_be(:context) { BulkImports::Pipeline::Context.new(entity) }
+
+  describe '#run' do
+    it 'maps existing users to the imported group' do
+      first_page = member_data(email: member_user1.email, has_next_page: true, cursor: cursor)
+      last_page = member_data(email: member_user2.email, has_next_page: false)
+
+      allow_next_instance_of(BulkImports::Common::Extractors::GraphqlExtractor) do |extractor|
+        allow(extractor)
+          .to receive(:extract)
+          .and_return(first_page, last_page)
+      end
+
+      expect { subject.run(context) }.to change(GroupMember, :count).by(2)
+
+      members = group.members.map { |m| m.slice(:user_id, :access_level) }
+
+      expect(members).to contain_exactly(
+        { user_id: member_user1.id, access_level: 30 },
+        { user_id: member_user2.id, access_level: 30 }
+      )
+    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) }
+
+    it 'has extractors' do
+      expect(described_class.get_extractor)
+        .to eq(
+          klass: BulkImports::Common::Extractors::GraphqlExtractor,
+          options: {
+            query: BulkImports::Groups::Graphql::GetMembersQuery
+          }
+        )
+    end
+
+    it 'has transformers' do
+      expect(described_class.transformers)
+        .to contain_exactly(
+          { klass: BulkImports::Common::Transformers::ProhibitedAttributesTransformer, options: nil },
+          { 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)
+    data = {
+      'created_at' => '2020-01-01T00:00:00Z',
+      'updated_at' => '2020-01-01T00:00:00Z',
+      'expires_at' => nil,
+      'access_level' => {
+        'integer_value' => 30
+      },
+      'user' => {
+        'public_email' => email
+      }
+    }
+
+    page_info = {
+      'end_cursor' => cursor,
+      'has_next_page' => has_next_page
+    }
+
+    BulkImports::Pipeline::ExtractedData.new(data: data, page_info: page_info)
+  end
+end

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

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::Groups::Transformers::MemberAttributesTransformer do
+  let_it_be(:user) { create(:user) }
+  let_it_be(:secondary_email) { 'secondary@email.com' }
+  let_it_be(:group) { create(:group) }
+  let_it_be(:bulk_import) { create(:bulk_import, user: user) }
+  let_it_be(:entity) { create(:bulk_import_entity, bulk_import: bulk_import, group: group) }
+  let_it_be(:context) { BulkImports::Pipeline::Context.new(entity) }
+
+  it 'returns nil when receives no data' do
+    expect(subject.transform(context, nil)).to eq(nil)
+  end
+
+  it 'returns nil when no user is found' do
+    expect(subject.transform(context, member_data)).to eq(nil)
+    expect(subject.transform(context, member_data(email: 'inexistent@email.com'))).to eq(nil)
+  end
+
+  context 'when the user is not confirmed' do
+    before do
+      user.update!(confirmed_at: nil)
+    end
+
+    it 'returns nil even when the primary email match' do
+      data = member_data(email: user.email)
+
+      expect(subject.transform(context, data)).to eq(nil)
+    end
+
+    it 'returns nil even when a secondary email match' do
+      user.emails << Email.new(email: secondary_email)
+      data = member_data(email: secondary_email)
+
+      expect(subject.transform(context, data)).to eq(nil)
+    end
+  end
+
+  context 'when the user is confirmed' do
+    before do
+      user.update!(confirmed_at: Time.now.utc)
+    end
+
+    it 'finds the user by the primary email' do
+      data = member_data(email: user.email)
+
+      expect(subject.transform(context, data)).to eq(
+        'access_level' => 30,
+        'user_id' => user.id,
+        'created_by_id' => user.id,
+        'created_at' => '2020-01-01T00:00:00Z',
+        'updated_at' => '2020-01-01T00:00:00Z',
+        'expires_at' => nil
+      )
+    end
+
+    it 'finds the user by the secondary email' do
+      user.emails << Email.new(email: secondary_email, confirmed_at: Time.now.utc)
+      data = member_data(email: secondary_email)
+
+      expect(subject.transform(context, data)).to eq(
+        'access_level' => 30,
+        'user_id' => user.id,
+        'created_by_id' => user.id,
+        'created_at' => '2020-01-01T00:00:00Z',
+        'updated_at' => '2020-01-01T00:00:00Z',
+        'expires_at' => nil
+      )
+    end
+
+    context 'format access level' do
+      it 'ignores record if no access level is given' do
+        data = member_data(email: user.email, access_level: nil)
+
+        expect(subject.transform(context, data)).to be_nil
+      end
+
+      it 'ignores record if is not a valid access level' do
+        data = member_data(email: user.email, access_level: 999)
+
+        expect(subject.transform(context, data)).to be_nil
+      end
+    end
+  end
+
+  def member_data(email: '', access_level: 30)
+    {
+      'created_at' => '2020-01-01T00:00:00Z',
+      'updated_at' => '2020-01-01T00:00:00Z',
+      'expires_at' => nil,
+      'access_level' => {
+        'integer_value' => access_level
+      },
+      'user' => {
+        'public_email' => email
+      }
+    }
+  end
+end

spec/lib/bulk_imports/importers/group_importer_spec.rb

@@ -18,9 +18,10 @@ RSpec.describe BulkImports::Importers::GroupImporter do
   describe '#execute' do
     it 'starts the entity and run its pipelines' do
       expect_to_run_pipeline BulkImports::Groups::Pipelines::GroupPipeline, context: context
+      expect_to_run_pipeline BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline, context: context
+      expect_to_run_pipeline BulkImports::Groups::Pipelines::MembersPipeline, context: context
       expect_to_run_pipeline BulkImports::Groups::Pipelines::LabelsPipeline, context: context
       expect_to_run_pipeline('EE::BulkImports::Groups::Pipelines::EpicsPipeline'.constantize, context: context) if Gitlab.ee?
-      expect_to_run_pipeline BulkImports::Groups::Pipelines::SubgroupEntitiesPipeline, context: context
 
       subject.execute