Implement ConflictsService.ListConflictFiles RPC

5 files changed, 370 insertions, 1 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 44bdda0a1..fbf458ad9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,8 @@
 
 UNRELEASED
 
+- Implement ConflictsService.ListConflictFiles RPC
+  https://gitlab.com/gitlab-org/gitaly/merge_requests/470
 - Implement RemoteService.RemoveRemote RPC
   https://gitlab.com/gitlab-org/gitaly/merge_requests/490
 - Implement RemoteService.AddRemote RPC
diff --git a/internal/service/conflicts/resolver.go b/internal/service/conflicts/resolver.go
index 926bf865c..08662fc9d 100644
--- a/internal/service/conflicts/resolver.go
+++ b/internal/service/conflicts/resolver.go
@@ -1,12 +1,60 @@
 package conflicts
 
 import (
+	"fmt"
+
 	pb "gitlab.com/gitlab-org/gitaly-proto/go"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
+	"gitlab.com/gitlab-org/gitaly/internal/rubyserver"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 )
 
 func (s *server) ListConflictFiles(in *pb.ListConflictFilesRequest, stream pb.ConflictsService_ListConflictFilesServer) error {
-	return helper.Unimplemented
+	ctx := stream.Context()
+
+	if err := validateListConflictFilesRequest(in); err != nil {
+		return grpc.Errorf(codes.InvalidArgument, "ListConflictFiles: %v", err)
+	}
+
+	client, err := s.ConflictsServiceClient(ctx)
+	if err != nil {
+		return err
+	}
+
+	clientCtx, err := rubyserver.SetHeaders(ctx, in.GetRepository())
+	if err != nil {
+		return err
+	}
+
+	rubyStream, err := client.ListConflictFiles(clientCtx, in)
+	if err != nil {
+		return err
+	}
+
+	return rubyserver.Proxy(func() error {
+		resp, err := rubyStream.Recv()
+		if err != nil {
+			md := rubyStream.Trailer()
+			stream.SetTrailer(md)
+			return err
+		}
+		return stream.Send(resp)
+	})
+}
+
+func validateListConflictFilesRequest(in *pb.ListConflictFilesRequest) error {
+	if in.GetRepository() == nil {
+		return fmt.Errorf("empty Repository")
+	}
+	if in.GetOurCommitOid() == "" {
+		return fmt.Errorf("empty OurCommitOid")
+	}
+	if in.GetTheirCommitOid() == "" {
+		return fmt.Errorf("empty TheirCommitOid")
+	}
+
+	return nil
 }
 
 func (s *server) ResolveConflicts(stream pb.ConflictsService_ResolveConflictsServer) error {
diff --git a/internal/service/conflicts/resolver_test.go b/internal/service/conflicts/resolver_test.go
new file mode 100644
index 000000000..a231f32a9
--- /dev/null
+++ b/internal/service/conflicts/resolver_test.go
@@ -0,0 +1,191 @@
+package conflicts
+
+import (
+	"io"
+	"testing"
+
+	"google.golang.org/grpc/codes"
+
+	"github.com/stretchr/testify/require"
+
+	pb "gitlab.com/gitlab-org/gitaly-proto/go"
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
+)
+
+type conflictFile struct {
+	header  *pb.ConflictFileHeader
+	content []byte
+}
+
+func TestSuccessfulListConflictFilesRequest(t *testing.T) {
+	server, serverSocketPath := runConflictsServer(t)
+	defer server.Stop()
+
+	client, conn := newConflictsClient(t, serverSocketPath)
+	defer conn.Close()
+
+	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
+	defer cleanupFn()
+
+	ourCommitOid := "0b4bc9a49b562e85de7cc9e834518ea6828729b9"
+	theirCommitOid := "bb5206fee213d983da88c47f9cf4cc6caf9c66dc"
+	conflictContent := `<<<<<<< files/ruby/feature.rb
+class Feature
+  def foo
+    puts 'bar'
+  end
+=======
+# This file was changed in feature branch
+# We put different code here to make merge conflict
+class Conflict
+>>>>>>> files/ruby/feature.rb
+end
+`
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	request := &pb.ListConflictFilesRequest{
+		Repository:     testRepo,
+		OurCommitOid:   ourCommitOid,
+		TheirCommitOid: theirCommitOid,
+	}
+
+	c, err := client.ListConflictFiles(ctx, request)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	files := getConflictFiles(t, c)
+	require.Len(t, files, 1)
+
+	file := files[0]
+	require.Equal(t, ourCommitOid, file.header.CommitOid)
+	require.Equal(t, int32(0100644), file.header.OurMode)
+	require.Equal(t, "files/ruby/feature.rb", string(file.header.OurPath))
+	require.Equal(t, "files/ruby/feature.rb", string(file.header.TheirPath))
+	require.Equal(t, conflictContent, string(file.content))
+}
+
+func TestFailedListConflictFilesRequestDueToConflictSideMissing(t *testing.T) {
+	server, serverSocketPath := runConflictsServer(t)
+	defer server.Stop()
+
+	client, conn := newConflictsClient(t, serverSocketPath)
+	defer conn.Close()
+
+	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
+	defer cleanupFn()
+
+	ourCommitOid := "eb227b3e214624708c474bdab7bde7afc17cefcc" // conflict-missing-side
+	theirCommitOid := "824be604a34828eb682305f0d963056cfac87b2d"
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	request := &pb.ListConflictFilesRequest{
+		Repository:     testRepo,
+		OurCommitOid:   ourCommitOid,
+		TheirCommitOid: theirCommitOid,
+	}
+
+	c, _ := client.ListConflictFiles(ctx, request)
+	testhelper.AssertGrpcError(t, drainListConflictFilesResponse(c), codes.FailedPrecondition, "")
+}
+
+func TestFailedListConflictFilesRequestDueToValidation(t *testing.T) {
+	server, serverSocketPath := runConflictsServer(t)
+	defer server.Stop()
+
+	client, conn := newConflictsClient(t, serverSocketPath)
+	defer conn.Close()
+
+	testRepo, _, cleanupFn := testhelper.NewTestRepo(t)
+	defer cleanupFn()
+
+	ourCommitOid := "0b4bc9a49b562e85de7cc9e834518ea6828729b9"
+	theirCommitOid := "bb5206fee213d983da88c47f9cf4cc6caf9c66dc"
+
+	testCases := []struct {
+		desc    string
+		request *pb.ListConflictFilesRequest
+		code    codes.Code
+	}{
+		{
+			desc: "empty repo",
+			request: &pb.ListConflictFilesRequest{
+				Repository:     nil,
+				OurCommitOid:   ourCommitOid,
+				TheirCommitOid: theirCommitOid,
+			},
+			code: codes.InvalidArgument,
+		},
+		{
+			desc: "empty OurCommitId repo",
+			request: &pb.ListConflictFilesRequest{
+				Repository:     testRepo,
+				OurCommitOid:   "",
+				TheirCommitOid: theirCommitOid,
+			},
+			code: codes.InvalidArgument,
+		},
+		{
+			desc: "empty TheirCommitId repo",
+			request: &pb.ListConflictFilesRequest{
+				Repository:     testRepo,
+				OurCommitOid:   ourCommitOid,
+				TheirCommitOid: "",
+			},
+			code: codes.InvalidArgument,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.desc, func(t *testing.T) {
+			ctx, cancel := testhelper.Context()
+			defer cancel()
+
+			c, _ := client.ListConflictFiles(ctx, testCase.request)
+			testhelper.AssertGrpcError(t, drainListConflictFilesResponse(c), testCase.code, "")
+		})
+	}
+}
+
+func getConflictFiles(t *testing.T, c pb.ConflictsService_ListConflictFilesClient) []conflictFile {
+	files := []conflictFile{}
+	currentFile := conflictFile{}
+	for {
+		r, err := c.Recv()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			t.Fatal(err)
+		}
+		for _, file := range r.GetFiles() {
+			// If there's a header this is the beginning of a new file
+			if header := file.GetHeader(); header != nil {
+				// Save previous file, except on the first iteration
+				if len(files) > 0 {
+					files = append(files, currentFile)
+				}
+
+				currentFile = conflictFile{header: header}
+			} else {
+				// Append to current file's content
+				currentFile.content = append(currentFile.content, file.GetContent()...)
+			}
+		}
+	}
+	// Append leftover file
+	files = append(files, currentFile)
+
+	return files
+}
+
+func drainListConflictFilesResponse(c pb.ConflictsService_ListConflictFilesClient) error {
+	var err error
+	for err == nil {
+		_, err = c.Recv()
+	}
+	return err
+}
diff --git a/internal/service/conflicts/testhelper_test.go b/internal/service/conflicts/testhelper_test.go
new file mode 100644
index 000000000..4a63dbd5c
--- /dev/null
+++ b/internal/service/conflicts/testhelper_test.go
@@ -0,0 +1,72 @@
+package conflicts
+
+import (
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+
+	pb "gitlab.com/gitlab-org/gitaly-proto/go"
+
+	"gitlab.com/gitlab-org/gitaly/internal/rubyserver"
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
+)
+
+func TestMain(m *testing.M) {
+	os.Exit(testMain(m))
+}
+
+var rubyServer *rubyserver.Server
+
+func testMain(m *testing.M) int {
+	defer testhelper.MustHaveNoChildProcess()
+
+	var err error
+
+	testhelper.ConfigureRuby()
+	rubyServer, err = rubyserver.Start()
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer rubyServer.Stop()
+
+	return m.Run()
+}
+
+func runConflictsServer(t *testing.T) (*grpc.Server, string) {
+	server := testhelper.NewTestGrpcServer(t, nil, nil)
+
+	serverSocketPath := testhelper.GetTemporaryGitalySocketFileName()
+	listener, err := net.Listen("unix", serverSocketPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pb.RegisterConflictsServiceServer(server, NewServer(rubyServer))
+	reflection.Register(server)
+
+	go server.Serve(listener)
+
+	return server, serverSocketPath
+}
+
+func newConflictsClient(t *testing.T, serverSocketPath string) (pb.ConflictsServiceClient, *grpc.ClientConn) {
+	connOpts := []grpc.DialOption{
+		grpc.WithInsecure(),
+		grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
+			return net.DialTimeout("unix", addr, timeout)
+		}),
+	}
+
+	conn, err := grpc.Dial(serverSocketPath, connOpts...)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return pb.NewConflictsServiceClient(conn), conn
+}
diff --git a/ruby/lib/gitaly_server/conflicts_service.rb b/ruby/lib/gitaly_server/conflicts_service.rb
index dc16e5b3c..50e760fc3 100644
--- a/ruby/lib/gitaly_server/conflicts_service.rb
+++ b/ruby/lib/gitaly_server/conflicts_service.rb
@@ -1,5 +1,61 @@
 module GitalyServer
   class ConflictsService < Gitaly::ConflictsService::Service
     include Utils
+
+    def list_conflict_files(request, call)
+      bridge_exceptions do
+        begin
+          resolver = conflict_resolver(request, call)
+          conflicts = resolver.conflicts
+          files = []
+          msg_size = 0
+
+          Enumerator.new do |y|
+            conflicts.each do |file|
+              files << Gitaly::ConflictFile.new(header: conflict_file_header(file))
+
+              strio = StringIO.new(file.content)
+              while chunk = strio.read(Gitlab.config.git.write_buffer_size - msg_size)
+                files << Gitaly::ConflictFile.new(content: chunk)
+                msg_size += chunk.bytesize
+
+                # We don't send a message for each chunk because the content of
+                # a file may be smaller than the size limit, which means we can
+                # keep adding data to the message
+                next if msg_size < Gitlab.config.git.write_buffer_size
+
+                y.yield(Gitaly::ListConflictFilesResponse.new(files: files))
+
+                files = []
+                msg_size = 0
+              end
+            end
+
+            # Send leftover data, if any
+            y.yield(Gitaly::ListConflictFilesResponse.new(files: files)) if files.any?
+          end
+        rescue Gitlab::Git::Conflict::Resolver::ConflictSideMissing => e
+          raise GRPC::FailedPrecondition.new(e.message)
+        end
+      end
+    end
+
+    private
+
+    def conflict_resolver(request, call)
+      repo = Gitlab::Git::Repository.from_gitaly(request.repository, call)
+
+      Gitlab::Git::Conflict::Resolver.new(repo, request.our_commit_oid, request.their_commit_oid)
+    end
+
+    def conflict_file_header(file)
+      Gitaly::ConflictFileHeader.new(
+        repository: file.repository.gitaly_repository,
+        commit_oid: file.commit_oid,
+        their_path: file.their_path,
+        our_path: file.our_path,
+        our_mode: file.our_mode
+      )
+    end
   end
 end