package com.atlassian.bitbucket.mesh.git;

import com.atlassian.bitbucket.mesh.config.*;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.execution.GrpcExecutionManager;
import com.atlassian.bitbucket.mesh.git.builder.GitProcess;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilder;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.GitQuarantineEnvironment;
import com.atlassian.bitbucket.mesh.git.builder.httpbackend.GitHttpBackendBuilder;
import com.atlassian.bitbucket.mesh.git.builder.receivepack.GitReceivePackBuilder;
import com.atlassian.bitbucket.mesh.git.cache.PackCacheService;
import com.atlassian.bitbucket.mesh.git.hook.GitHook;
import com.atlassian.bitbucket.mesh.git.hook.GitHookInputParserBuilder;
import com.atlassian.bitbucket.mesh.git.hook.GitHookOutputWriter;
import com.atlassian.bitbucket.mesh.git.hook.TestableGitHookService;
import com.atlassian.bitbucket.mesh.git.hosting.GitReceivePackFilter;
import com.atlassian.bitbucket.mesh.git.hosting.GitReceivePackListener;
import com.atlassian.bitbucket.mesh.git.ref.RefUpdate;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.management.SidebandManager;
import com.atlassian.bitbucket.mesh.process.OsProcess;
import com.atlassian.bitbucket.mesh.process.ProcessStdioHandler;
import com.atlassian.bitbucket.mesh.repository.RefsUpdatedEvent;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.repository.RepositoryManager;
import com.atlassian.bitbucket.mesh.request.RequestContext;
import com.atlassian.bitbucket.mesh.request.RequestManager;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.test.ExecutionContextRule;
import com.atlassian.bitbucket.mesh.transaction.UpdateRefsContext;
import com.atlassian.bitbucket.mesh.transaction.UpdateRefsContextHolder;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.protobuf.ByteString;
import com.google.protobuf.MessageOrBuilder;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import org.apache.commons.lang3.tuple.Triple;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.io.IOException;
import java.nio.file.Path;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.atlassian.bitbucket.mesh.git.GitConstants.*;
import static com.atlassian.bitbucket.mesh.git.hook.GitHook.POST_RECEIVE;
import static com.atlassian.bitbucket.mesh.git.hook.GitHook.PRE_RECEIVE;
import static com.atlassian.bitbucket.mesh.test.MoreMatchers.*;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.joining;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
@SuppressWarnings("unchecked")
public class GrpcHostingServiceMockTest {

    private static final RpcCgiEnvironment CGI_ENV_RECEIVE_PACK = RpcCgiEnvironment.newBuilder()
            .setContentType("application/x-git-receive-pack-request")
            .setContentLength(445)
            .setPathInfo("/git-receive-pack")
            .setRemoteAddr("0:0:0:0:0:0:0:1")
            .setRemoteUser("admin")
            .setRequestMethod("POST")
            .setServerProtocol("HTTP/1.1")
            .setServerSoftware("Apache Tomcat/9.0.39")
            .addAllHttpAccept(singletonList("application/x-git-receive-pack-result"))
            .addAllHttpAcceptEncoding(singletonList("deflate, gzip, zstd"))
            .build();
    private static final RpcCgiEnvironment CGI_ENV_UPLOAD_PACK = RpcCgiEnvironment.newBuilder()
            .setContentType("application/x-git-upload-pack-request")
            .setContentLength(445)
            .setPathInfo("/git-upload-pack")
            .setRemoteAddr("0:0:0:0:0:0:0:1")
            .setRemoteUser("admin")
            .setRequestMethod("POST")
            .setServerProtocol("HTTP/1.1")
            .setServerSoftware("Apache Tomcat/9.0.39")
            .addAllHttpAccept(singletonList("application/x-git-upload-pack-result"))
            .addAllHttpAcceptEncoding(singletonList("deflate, gzip, zstd"))
            .build();
    private static final RpcGitOptions GIT_OPTIONS = RpcGitOptions.newBuilder()
            .build();
    private static final String HOOK_ADDR = "127.0.0.1";
    private static final int HOOK_PORT = 40547;
    private static final long MAX_FRAGMENT_SIZE = 128L;
    private static final String QUARANTINE_ENV = "00aa";
    private static final String REPO_ID = "Contrary to popular belief, I know exactly what I’m doing.";

    private static final RpcHttpBackendFragment HTTP_FETCH = RpcHttpBackendFragment.newBuilder()
            .setRequest(RpcHttpBackendRequest.newBuilder()
                    .setRepository(REPO_ID)
                    .setOptions(GIT_OPTIONS)
                    .setCgiEnvironment(CGI_ENV_UPLOAD_PACK)
                    .build())
            .build();
    private static final RpcHttpBackendFragment HTTP_PUSH = RpcHttpBackendFragment.newBuilder()
            .setRequest(RpcHttpBackendRequest.newBuilder()
                    .setRepository(REPO_ID)
                    .setOptions(GIT_OPTIONS)
                    .setCgiEnvironment(CGI_ENV_RECEIVE_PACK)
                    .build())
            .build();

    private static final RpcReceivePackFragment SSH_PUSH = RpcReceivePackFragment.newBuilder()
            .setRequest(RpcReceivePackRequest.newBuilder()
                    .setRepository(REPO_ID)
                    .setOptions(GIT_OPTIONS)
                    .build())
            .build();

    @Rule
    public final ExecutionContextRule executionContext = new ExecutionContextRule(new GrpcExecutionManager());
    @Rule
    public final TemporaryDirectory tempDir = new TemporaryDirectory();

    @Mock
    private GitProcessBuilderFactory builderFactory;
    private Set<String> committedRefs;
    @Mock
    private ApplicationConfig config;
    @Mock
    private EventPublisher eventPublisher;
    @Mock
    private GitConfig gitConfig;
    @Mock
    private GitHookOutputWriter hookOutputWriter;
    private TestableGitHookService hookService;
    @Mock
    private HooksConfig hooksConfig;
    @Mock
    private HostingConfig hostingConfig;
    private GrpcHostingService hostingService;
    @Mock(answer = Answers.RETURNS_SELF)
    private GitHttpBackendBuilder httpBackendBuilder;
    @Mock
    private PackCacheService packCacheService;
    @Mock
    private GitProcess<?> process;
    @Mock
    private GitProcessBuilder processBuilder;
    @Mock
    private GrpcPushTracer pushTracer;
    @Mock
    private GitQuarantineRegistry quarantineRegistry;
    @Mock(answer = Answers.RETURNS_SELF)
    private GitReceivePackBuilder receivePackBuilder;
    @Mock
    private GitReceivePackListener receivePackListener;
    @Mock
    private Repository repository;
    @Mock
    private RepositoryManager repositoryManager;
    @Mock
    private RequestManager requestManager;
    @Mock
    private SidebandManager sidebandManager;
    @Mock
    private UpdateRefsContext updateRefsContext;
    @Mock
    private UpdateRefsContextHolder updateRefsContextHolder;

    @Before
    public void setup() throws IOException {
        Path repoDir = tempDir.newFolder("repo-dir");

        when(repositoryManager.getById(REPO_ID)).thenReturn(repository);
        when(repository.getPath()).thenReturn(repoDir);
        when(repository.getHooksDir()).thenReturn(repoDir.resolve("hooks"));
        when(repository.getId()).thenReturn("p/0012/h/abc123abc123abc123ab/r/0001112223");

        when(builderFactory.builder(any(MessageOrBuilder.class))).thenReturn(processBuilder);
        when(gitConfig.getHooks()).thenReturn(hooksConfig);
        when(gitConfig.getHosting()).thenReturn(hostingConfig);
        GrpcConfig grpcConfig = mock(GrpcConfig.class);
        when(grpcConfig.getMaxFragmentSize()).thenReturn(MAX_FRAGMENT_SIZE);
        when(config.getGit()).thenReturn(gitConfig);
        when(config.getGrpc()).thenReturn(grpcConfig);
        when(hooksConfig.getSocketAddress()).thenReturn(HOOK_ADDR);
        when(hooksConfig.getSocketPort()).thenReturn(HOOK_PORT);
        when(hooksConfig.getCallbackTimeout()).thenReturn(Duration.ofSeconds(10));
        when(hostingConfig.isFilterAllowed()).thenReturn(true);
        when(processBuilder.httpBackend()).thenReturn(httpBackendBuilder);
        when(processBuilder.receivePack()).thenReturn(receivePackBuilder);
        when(quarantineRegistry.register(any(GitQuarantineEnvironment.class))).thenAnswer(invocation -> {
            GitQuarantineEnvironment env = invocation.getArgument(0);
            return env.getRepository().getId() + "?q=" + QUARANTINE_ENV;
        });
        ArgumentCaptor<ProcessStdioHandler> stdioHandlerCaptor = ArgumentCaptor.forClass(ProcessStdioHandler.class);
        when(httpBackendBuilder.build(stdioHandlerCaptor.capture())).thenReturn(process);
        when(receivePackBuilder.build(stdioHandlerCaptor.capture())).thenReturn(process);
        when(process.start()).thenAnswer(invocation -> {
            ProcessStdioHandler<?> handler = stdioHandlerCaptor.getValue();
            if (handler != null) {
                OsProcess osProcess = mock(OsProcess.class);
                handler.onPreStart(osProcess);
                handler.onStart(osProcess);
            }
            return new CompletableFuture<>();
        });

        //Setup the UpdateRefsContext correctly, marks every ref update in pre receive as committed
        committedRefs = ImmutableSet.of();
        doAnswer(invocation -> {
            Collection<RefUpdate> refUpdates = invocation.getArgument(1);
            when(updateRefsContextHolder.get()).thenReturn(updateRefsContext);
            Map<String, RefUpdate> returnValue = refUpdates.stream()
                    .collect(Collectors.toMap(RefUpdate::getRefId, identity()));
            when(updateRefsContext.getRefUpdates()).thenReturn(returnValue);
            committedRefs = new CopyOnWriteArraySet<>();
            committedRefs.addAll(refUpdates.stream().map(RefUpdate::getRefId).collect(Collectors.toList()));
            return null;
        }).when(updateRefsContextHolder).set(any(), any());
        doAnswer(invocation -> {
            Collection<String> refs = invocation.getArgument(0);
            committedRefs.removeAll(refs);
            return null;
        }).when(updateRefsContext).clearCommitted(any());
        when(updateRefsContext.getCommittedRefUpdates()).thenAnswer(invocation -> committedRefs);

        hookService = new TestableGitHookService(hooksConfig, executionContext.getExecutionManager(),
                Executors.newSingleThreadExecutor(), emptyList(), repositoryManager, tempDir.newFolder("tmp"));
        hookService.startup();

        hostingService = new GrpcHostingService(builderFactory, config, eventPublisher, hookService,
                packCacheService, pushTracer, quarantineRegistry, ImmutableList.of(receivePackListener),
                repositoryManager, requestManager, sidebandManager, updateRefsContextHolder);
    }

    @After
    public void tearDown() throws Exception {
        //Make sure we don't accidentally send sideband messages
        try {
            verifyNoMoreInteractions(sidebandManager);
            verify(updateRefsContextHolder).clear();
        } finally {
            hookService.shutdown();
            Thread.sleep(100);
        }
    }

    @Test
    public void testHttpBackendDoesNotRegisterGitReceivePackListenerOnFetch() {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);

        httpBackend.onNext(HTTP_FETCH);

        httpBackend.onCompleted();

        verify(receivePackListener, never()).createFilter(repository);
        verify(updateRefsContextHolder).get();
    }

    @Test
    public void testHttpBackendHandlesLargePostReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);

        httpBackend.onNext(HTTP_PUSH);

        Triple<String, String, String> refUpdate1 = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/master");
        Triple<String, String, String> refUpdate2 = Triple.of(
                "90eb07473b7194fd2957eaf2d8ce7e07b2cabebe",
                "9835e3b059b4f5b89b743bbaabbecf6c8cb29371",
                "refs/heads/another-branch");
        Thread hook = invokeHook(POST_RECEIVE, ImmutableList.of(refUpdate1, refUpdate2));

        httpBackend.onNext(RpcHttpBackendFragment.newBuilder()
                .setPostReceive(RpcHostingPostReceiveResponse.newBuilder())
                .build());

        hook.join();

        httpBackend.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(3)).onNext(fragment.capture());

        RpcHostingPostReceiveRequest fragment1 = fragment.getAllValues().get(1).getPostReceive();
        assertThat(fragment1, notNullValue());
        assertThat(fragment1.getRefUpdatesCount(), is(1));
        assertThat(fragment1.getMoreRefUpdates(), is(true));
        assertThat(fragment1.getRefUpdates(0), allOf(
                hasOldHash(refUpdate1.getLeft()),
                hasNewHash(refUpdate1.getMiddle()),
                hasRefId(refUpdate1.getRight())));

        RpcHostingPostReceiveRequest fragment2 = fragment.getAllValues().get(2).getPostReceive();
        assertThat(fragment2, notNullValue());
        assertThat(fragment2.getRefUpdatesCount(), is(1));
        assertThat(fragment2.getMoreRefUpdates(), is(false));
        assertThat(fragment2.getRefUpdates(0), allOf(
                hasOldHash(refUpdate2.getLeft()),
                hasNewHash(refUpdate2.getMiddle()),
                hasRefId(refUpdate2.getRight())));
        verify(eventPublisher).publish(any(RefsUpdatedEvent.class));
        verify(updateRefsContextHolder, times(3)).get();
    }

    @Test
    public void testHttpBackendHandlesLargePreReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);

        httpBackend.onNext(HTTP_PUSH);

        Triple<String, String, String> refUpdate1 = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/branch");
        Triple<String, String, String> refUpdate2 = Triple.of(
                "e90eb07473b7194fd2957eaf2d8ce7e07b2cabeb",
                "e3b059b4f5b89b743bbaabbecf6c8cb293719835",
                "refs/heads/another-branch");
        Thread hook = invokeHook(PRE_RECEIVE, ImmutableList.of(refUpdate1, refUpdate2));

        httpBackend.onNext(RpcHttpBackendFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(false)
                        .build())
                .build());

        hook.join();

        httpBackend.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(3)).onNext(fragment.capture());

        RpcHostingPreReceiveRequest request1 = fragment.getAllValues().get(1).getPreReceive();
        assertThat(request1, notNullValue());
        assertThat(request1.getRepository(), is(repository.getId() + "?q=" + QUARANTINE_ENV));
        assertThat(request1.getRefUpdatesCount(), is(1));
        assertThat(request1.getMoreRefUpdates(), is(true));

        RpcHostingPreReceiveRequest request2 = fragment.getAllValues().get(2).getPreReceive();
        assertThat(request2, notNullValue());
        assertThat(request2.getRepository(), is(repository.getId() + "?q=" + QUARANTINE_ENV));
        assertThat(request2.getRefUpdatesCount(), is(1));
        assertThat(request2.getMoreRefUpdates(), is(false));

        RefUpdate expectedRefUpdate1 =
                RefUpdate.of(refUpdate1.getRight(), refUpdate1.getLeft(), refUpdate1.getMiddle());
        assertEquals(expectedRefUpdate1.toRefUpdate(), request1.getRefUpdates(0));
        RefUpdate expectedRefUpdate2 =
                RefUpdate.of(refUpdate2.getRight(), refUpdate2.getLeft(), refUpdate2.getMiddle());
        assertEquals(expectedRefUpdate2.toRefUpdate(), request2.getRefUpdates(0));

        ArgumentCaptor<Collection<RefUpdate>> refUpdatesCaptor = ArgumentCaptor.forClass(Collection.class);
        verify(updateRefsContextHolder).set(eq(repository), refUpdatesCaptor.capture());
        assertThat(refUpdatesCaptor.getValue(), contains(expectedRefUpdate1, expectedRefUpdate2));
        verify(hookOutputWriter).writeExitCode(0);
    }

    @Test
    public void testHttpBackendHandlesPostReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);

        httpBackend.onNext(HTTP_PUSH);

        Triple<String, String, String> refUpdate = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/master");
        Thread hook = invokeHook(POST_RECEIVE, singletonList(refUpdate));

        httpBackend.onNext(RpcHttpBackendFragment.newBuilder()
                .setPostReceive(RpcHostingPostReceiveResponse.newBuilder())
                .build());

        hook.join();

        httpBackend.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(2)).onNext(fragment.capture());

        RpcHostingPostReceiveRequest request = fragment.getValue().getPostReceive();
        assertThat(request, notNullValue());
        assertThat(request.getRefUpdatesCount(), is(1));

        assertThat(request.getRefUpdates(0), allOf(
                hasOldHash(refUpdate.getLeft()),
                hasNewHash(refUpdate.getMiddle()),
                hasRefId(refUpdate.getRight())
        ));
        verify(eventPublisher).publish(any(RefsUpdatedEvent.class));
        verify(updateRefsContextHolder, times(2)).get();
    }

    @Test
    public void testHttpBackendHandlesPreReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);

        httpBackend.onNext(HTTP_PUSH);

        Triple<String, String, String> refUpdate = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/branch");
        Thread hook = invokeHook(PRE_RECEIVE, singletonList(refUpdate));

        httpBackend.onNext(RpcHttpBackendFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(false)
                        .build())
                .build());

        hook.join();

        httpBackend.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(2)).onNext(fragment.capture());

        RpcHostingPreReceiveRequest request = fragment.getValue().getPreReceive();
        assertThat(request, notNullValue());
        assertThat(request.getRepository(), is(repository.getId() + "?q=" + QUARANTINE_ENV));
        assertThat(request.getRefUpdatesCount(), is(1));

        RefUpdate expectedRefUpdate = RefUpdate.of(refUpdate.getRight(), refUpdate.getLeft(), refUpdate.getMiddle());
        assertEquals(expectedRefUpdate.toRefUpdate(), request.getRefUpdates(0));

        ArgumentCaptor<Collection<RefUpdate>> refUpdatesCaptor = ArgumentCaptor.forClass(Collection.class);
        verify(updateRefsContextHolder).set(eq(repository), refUpdatesCaptor.capture());
        assertThat(refUpdatesCaptor.getValue(), contains(expectedRefUpdate));
        verify(hookOutputWriter).writeExitCode(0);
    }

    @Test
    public void testHttpBackendPrintsPostReceiveHookResponse() throws Exception {
        ByteString err = ByteString.copyFromUtf8(
                "In all my years of conquest, violence, slaughter, it was never personal. " +
                        "But I’ll tell you now, what I’m about to do to your stubborn, annoying little planet… " +
                        "I’m gonna enjoy it. Very, very much.");
        ByteString out = ByteString.copyFromUtf8(
                "You’re strong. But I could snap my fingers, and you’d all cease to exist.");

        testHookWritesHookOutput(POST_RECEIVE, err, out, hostingService::httpBackend, HTTP_PUSH,
                RpcHttpBackendFragment.newBuilder()
                        .setPostReceive(RpcHostingPostReceiveResponse.newBuilder()
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDERR)
                                        .setData(err))
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDOUT)
                                        .setData(out)
                                        .build())
                                .build())
                        .build());
    }

    @Test
    public void testHttpBackendPrintsPreReceiveHookResponse() throws Exception {
        ByteString err = ByteString.copyFromUtf8(
                "Fun isn't something one considers when balancing the universe. But this... does put a smile on my " +
                        "face.");
        ByteString out = ByteString.copyFromUtf8("Reality is often disappointing.");

        testHookWritesHookOutput(PRE_RECEIVE, err, out, hostingService::httpBackend, HTTP_PUSH,
                RpcHttpBackendFragment.newBuilder()
                        .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDERR)
                                        .setData(err))
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDOUT)
                                        .setData(out)
                                        .build())
                                .setRejected(false)
                                .build())
                        .build());
    }

    @Test
    public void testHttpBackendRegistersGitReceivePackListenerOnPush() {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);
        GitReceivePackFilter receivePackFilter = mock(GitReceivePackFilter.class);
        when(receivePackListener.createFilter(repository)).thenReturn(receivePackFilter);
        when(receivePackFilter.onPack(any())).thenAnswer(invocation -> invocation.getArgument(0));
        when(receivePackFilter.onPacketLine(any())).thenAnswer(invocation -> invocation.getArgument(0));
        when(receivePackFilter.wantsPack()).thenReturn(true);
        when(receivePackFilter.wantsPacketLines()).thenReturn(true);

        httpBackend.onNext(HTTP_PUSH);

        httpBackend.onCompleted();

        verify(receivePackListener).createFilter(repository);

        //Send a few chunks of stdin. The filter should get invoked with the stdin chunks
        httpBackend.onNext(RpcHttpBackendFragment.newBuilder().setStdin(chunk("0008abcd")).build());
        httpBackend.onNext(RpcHttpBackendFragment.newBuilder().setStdin(chunk("0000")).build());
        httpBackend.onNext(RpcHttpBackendFragment.newBuilder().setStdin(chunk("PACK\0start")).build());
        httpBackend.onNext(RpcHttpBackendFragment.newBuilder().setStdin(chunk("more pack data")).build());

        verify(receivePackFilter, times(2)).onPacketLine(any());
        verify(receivePackFilter, times(2)).onPack(any());
    }

    @Test
    public void testPreReceiveRejectsExecutionOfHttpBackend() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcHttpBackendFragment> httpBackend = hostingService.httpBackend(responseObserver);

        httpBackend.onNext(HTTP_PUSH);

        Thread hook = invokeHook(PRE_RECEIVE, emptyList());

        httpBackend.onNext(RpcHttpBackendFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(true)
                        .build())
                .build());

        hook.join();

        httpBackend.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(2)).onNext(fragment.capture());
        verify(hookOutputWriter).writeExitCode(1);
        verify(updateRefsContextHolder).get();
    }

    @Test
    public void testPreReceiveRejectsExecutionOfReceivePack() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        receivePack.onNext(SSH_PUSH);

        Thread hook = invokeHook(PRE_RECEIVE, emptyList());

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(true)
                        .build())
                .build());

        hook.join();

        receivePack.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(2)).onNext(fragment.capture());
        verify(hookOutputWriter).writeExitCode(1);
        verify(updateRefsContextHolder).get();
    }

    @Test
    public void testReceivePackHandlesCancelWithOutOfBandUpdates() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        receivePack.onNext(SSH_PUSH);

        Triple<String, String, String> refUpdate1 = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/branch");
        Triple<String, String, String> refUpdate2 = Triple.of(
                "e90eb07473b7194fd2957eaf2d8ce7e07b2cabeb",
                "e3b059b4f5b89b743bbaabbecf6c8cb293719835",
                "refs/heads/another-branch");
        Thread hook = invokeHook(PRE_RECEIVE, ImmutableList.of(refUpdate1, refUpdate2));

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(false)
                        .build())
                .build());

        hook.join();

        RequestContext requestContext = mock(RequestContext.class);
        String user = "user1";
        when(requestContext.getUser()).thenReturn(user);
        String correlationId = "abcdefghijkl-0123456789";
        when(requestContext.getCorrelationId()).thenReturn(correlationId);
        when(requestManager.getContext()).thenReturn(Optional.of(requestContext));

        //Send error/cancel message with committed ref updates
        receivePack.onError(mock(Throwable.class));

        ArgumentCaptor<RpcSidebandFollowerFragment> captor =
                ArgumentCaptor.forClass(RpcSidebandFollowerFragment.class);
        verify(sidebandManager, times(2)).send(captor.capture());

        RpcSidebandFollowerFragment fragment = captor.getAllValues().get(0);
        List<RpcRefUpdate> refUpdatesList = fragment.getOutOfBandRefUpdates().getRefUpdatesList();
        assertThat(refUpdatesList, hasSize(1));
        assertThat(refUpdatesList.get(0).getRefId().toString(UTF_8), is(refUpdate1.getRight()));
        assertThat(fragment.getOutOfBandRefUpdates().getCorrelationId(), is(correlationId));
        assertThat(fragment.getOutOfBandRefUpdates().getUser(), is(user));

        fragment = captor.getAllValues().get(1);
        refUpdatesList = fragment.getOutOfBandRefUpdates().getRefUpdatesList();
        assertThat(refUpdatesList, hasSize(1));
        assertThat(refUpdatesList.get(0).getRefId().toString(UTF_8), is(refUpdate2.getRight()));
        assertThat(fragment.getOutOfBandRefUpdates().getCorrelationId(), is(correlationId));
        assertThat(fragment.getOutOfBandRefUpdates().getUser(), is(user));
    }

    @Test
    public void testReceivePackHandlesCancelWithoutContextGracefully() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);
        receivePack.onNext(SSH_PUSH);

        //No update refs context was set up, this should not do anything (but log a warning)
        receivePack.onError(mock(Throwable.class));
    }

    @Test
    public void testReceivePackHandlesCancelWithoutRequestContextGracefully() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);
        receivePack.onNext(SSH_PUSH);
        Triple<String, String, String> refUpdate = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/branch");
        Thread hook = invokeHook(PRE_RECEIVE, ImmutableList.of(refUpdate));
        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(false)
                        .build())
                .build());

        hook.join();

        //No request context was set up, this should not do anything (but log a warning)
        receivePack.onError(mock(Throwable.class));
    }

    @Test
    public void testReceivePackHandlesLargePostReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        receivePack.onNext(SSH_PUSH);

        Triple<String, String, String> refUpdate1 = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/master");
        Triple<String, String, String> refUpdate2 = Triple.of(
                "90eb07473b7194fd2957eaf2d8ce7e07b2cabebe",
                "9835e3b059b4f5b89b743bbaabbecf6c8cb29371",
                "refs/heads/another-branch");
        Thread hook = invokeHook(POST_RECEIVE, ImmutableList.of(refUpdate1, refUpdate2));

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPostReceive(RpcHostingPostReceiveResponse.newBuilder())
                .build());

        hook.join();

        receivePack.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(3)).onNext(fragment.capture());

        RpcHostingPostReceiveRequest fragment1 = fragment.getAllValues().get(1).getPostReceive();
        assertThat(fragment1, notNullValue());
        assertThat(fragment1.getRefUpdatesCount(), is(1));
        assertThat(fragment1.getMoreRefUpdates(), is(true));
        assertThat(fragment1.getRefUpdates(0), allOf(
                hasOldHash(refUpdate1.getLeft()),
                hasNewHash(refUpdate1.getMiddle()),
                hasRefId(refUpdate1.getRight())));

        RpcHostingPostReceiveRequest fragment2 = fragment.getAllValues().get(2).getPostReceive();
        assertThat(fragment2, notNullValue());
        assertThat(fragment2.getRefUpdatesCount(), is(1));
        assertThat(fragment2.getMoreRefUpdates(), is(false));
        assertThat(fragment2.getRefUpdates(0), allOf(
                hasOldHash(refUpdate2.getLeft()),
                hasNewHash(refUpdate2.getMiddle()),
                hasRefId(refUpdate2.getRight())));
        verify(eventPublisher).publish(any(RefsUpdatedEvent.class));
        verify(updateRefsContextHolder, times(3)).get();
    }

    @Test
    public void testReceivePackHandlesLargePreReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        receivePack.onNext(SSH_PUSH);

        Triple<String, String, String> refUpdate1 = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/branch");
        Triple<String, String, String> refUpdate2 = Triple.of(
                "e90eb07473b7194fd2957eaf2d8ce7e07b2cabeb",
                "e3b059b4f5b89b743bbaabbecf6c8cb293719835",
                "refs/heads/another-branch");
        Thread hook = invokeHook(PRE_RECEIVE, ImmutableList.of(refUpdate1, refUpdate2));

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(false)
                        .build())
                .build());

        hook.join();

        receivePack.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(3)).onNext(fragment.capture());

        RpcHostingPreReceiveRequest request1 = fragment.getAllValues().get(1).getPreReceive();
        assertThat(request1, notNullValue());
        assertThat(request1.getRepository(), is(repository.getId() + "?q=" + QUARANTINE_ENV));
        assertThat(request1.getRefUpdatesCount(), is(1));
        assertThat(request1.getMoreRefUpdates(), is(true));

        RpcHostingPreReceiveRequest request2 = fragment.getAllValues().get(2).getPreReceive();
        assertThat(request2, notNullValue());
        assertThat(request2.getRepository(), is(repository.getId() + "?q=" + QUARANTINE_ENV));
        assertThat(request2.getRefUpdatesCount(), is(1));
        assertThat(request2.getMoreRefUpdates(), is(false));

        RefUpdate expectedRefUpdate1 =
                RefUpdate.of(refUpdate1.getRight(), refUpdate1.getLeft(), refUpdate1.getMiddle());
        assertEquals(expectedRefUpdate1.toRefUpdate(), request1.getRefUpdates(0));
        RefUpdate expectedRefUpdate2 =
                RefUpdate.of(refUpdate2.getRight(), refUpdate2.getLeft(), refUpdate2.getMiddle());
        assertEquals(expectedRefUpdate2.toRefUpdate(), request2.getRefUpdates(0));

        ArgumentCaptor<Collection<RefUpdate>> refUpdatesCaptor = ArgumentCaptor.forClass(Collection.class);
        verify(updateRefsContextHolder).set(eq(repository), refUpdatesCaptor.capture());
        assertThat(refUpdatesCaptor.getValue(), contains(expectedRefUpdate1, expectedRefUpdate2));
        verify(hookOutputWriter).writeExitCode(0);
    }

    @Test
    public void testReceivePackHandlesPostReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        receivePack.onNext(SSH_PUSH);

        Triple<String, String, String> refUpdate = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/master");
        Thread hook = invokeHook(PRE_RECEIVE, singletonList(refUpdate));

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder().build())
                .build());

        hook.join();

        hook = invokeHook(POST_RECEIVE, singletonList(refUpdate));

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPostReceive(RpcHostingPostReceiveResponse.newBuilder())
                .build());

        hook.join();

        receivePack.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(3)).onNext(fragment.capture());

        RpcHostingPostReceiveRequest request = fragment.getValue().getPostReceive();
        assertThat(request, notNullValue());
        assertThat(request.getRefUpdatesCount(), is(1));

        assertThat(request.getRefUpdates(0), allOf(
                hasOldHash(refUpdate.getLeft()),
                hasNewHash(refUpdate.getMiddle()),
                hasRefId(refUpdate.getRight())
        ));
        verify(updateRefsContext).clearCommitted(ImmutableList.of(refUpdate.getRight()));
        verify(eventPublisher).publish(any(RefsUpdatedEvent.class));
        verify(updateRefsContextHolder, times(2)).get();
        verifyNoInteractions(sidebandManager);
    }

    @Test
    public void testReceivePackHandlesPreReceiveHook() throws Exception {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        receivePack.onNext(SSH_PUSH);

        Triple<String, String, String> refUpdate = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/branch");
        Thread hook = invokeHook(PRE_RECEIVE, singletonList(refUpdate));

        receivePack.onNext(RpcReceivePackFragment.newBuilder()
                .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                        .setRejected(false)
                        .build())
                .build());

        hook.join();

        receivePack.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(2)).onNext(fragment.capture());

        RpcHostingPreReceiveRequest request = fragment.getValue().getPreReceive();
        assertThat(request, notNullValue());
        assertThat(request.getRepository(), is(repository.getId() + "?q=" + QUARANTINE_ENV));
        assertThat(request.getRefUpdatesCount(), is(1));

        RefUpdate expectedRefUpdate = RefUpdate.of(refUpdate.getRight(), refUpdate.getLeft(), refUpdate.getMiddle());
        assertEquals(expectedRefUpdate.toRefUpdate(), request.getRefUpdates(0));

        ArgumentCaptor<Collection<RefUpdate>> refUpdatesCaptor = ArgumentCaptor.forClass(Collection.class);
        verify(updateRefsContextHolder).set(eq(repository), refUpdatesCaptor.capture());
        assertThat(refUpdatesCaptor.getValue(), contains(expectedRefUpdate));

        verify(hookOutputWriter).writeExitCode(0);
    }

    @Test
    public void testReceivePackPrintsPostReceiveHookResponse() throws Exception {
        ByteString err = ByteString.copyFromUtf8(
                "In all my years of conquest, violence, slaughter, it was never personal. " +
                        "But I’ll tell you now, what I’m about to do to your stubborn, annoying little planet… " +
                        "I’m gonna enjoy it. Very, very much.");
        ByteString out = ByteString.copyFromUtf8(
                "You’re strong. But I could snap my fingers, and you’d all cease to exist.");

        testHookWritesHookOutput(POST_RECEIVE, err, out, hostingService::receivePack, SSH_PUSH,
                RpcReceivePackFragment.newBuilder()
                        .setPostReceive(RpcHostingPostReceiveResponse.newBuilder()
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDERR)
                                        .setData(err))
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDOUT)
                                        .setData(out)
                                        .build())
                                .build())
                        .build());
        verify(updateRefsContextHolder, times(2)).get();
    }

    @Test
    public void testReceivePackPrintsPreReceiveHookResponse() throws Exception {
        ByteString err = ByteString.copyFromUtf8(
                "Fun isn't something one considers when balancing the universe. But this... does put a smile on my " +
                        "face.");
        ByteString out = ByteString.copyFromUtf8("Reality is often disappointing.");

        testHookWritesHookOutput(PRE_RECEIVE, err, out, hostingService::receivePack, SSH_PUSH,
                RpcReceivePackFragment.newBuilder()
                        .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDERR)
                                        .setData(err))
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDOUT)
                                        .setData(out)
                                        .build())
                                .setRejected(false)
                                .build())
                        .build());
    }

    @Test
    public void testReceivePackRegistersGitReceivePackListenerOnPush() {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<RpcReceivePackFragment> receivePack = hostingService.receivePack(responseObserver);

        GitReceivePackFilter receivePackFilter = mock(GitReceivePackFilter.class);
        when(receivePackListener.createFilter(repository)).thenReturn(receivePackFilter);
        when(receivePackFilter.onPack(any())).thenAnswer(invocation -> invocation.getArgument(0));
        when(receivePackFilter.onPacketLine(any())).thenAnswer(invocation -> invocation.getArgument(0));
        when(receivePackFilter.wantsPack()).thenReturn(true);
        when(receivePackFilter.wantsPacketLines()).thenReturn(true);

        receivePack.onNext(SSH_PUSH);

        receivePack.onCompleted();

        verify(receivePackListener).createFilter(repository);

        //Send a few chunks of stdin. The filter should get invoked with the stdin chunks
        receivePack.onNext(RpcReceivePackFragment.newBuilder().setStdin(chunk("0008abcd")).build());
        receivePack.onNext(RpcReceivePackFragment.newBuilder().setStdin(chunk("0000")).build());
        receivePack.onNext(RpcReceivePackFragment.newBuilder().setStdin(chunk("PACK\0start")).build());
        receivePack.onNext(RpcReceivePackFragment.newBuilder().setStdin(chunk("more pack data")).build());

        verify(receivePackFilter, times(2)).onPacketLine(any());
        verify(receivePackFilter, times(2)).onPack(any());
    }

    private static RpcStreamChunk chunk(String data) {
        return RpcStreamChunk.newBuilder()
                .setData(ByteString.copyFromUtf8(data))
                .build();
    }

    private static ServerCallStreamObserver<RpcHostingWriteStreamFragment> makeResponseObserver() {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = mock(ServerCallStreamObserver.class);
        when(responseObserver.isReady()).thenReturn(true);

        return responseObserver;
    }

    private Thread invokeHook(GitHook hook, Collection<Triple<String, String, String>> refUpdates) {
        Thread thread = (new Thread(() -> {
            try {
                GitHookInputParserBuilder builder = new GitHookInputParserBuilder()
                        .hookRequestId("1")
                        .gitDir(repository.getPath().toString())
                        .hookType(hook.getName())
                        .stdin(refUpdates.stream()
                                .map(ru -> String.format("%s %s %s", ru.getLeft(), ru.getMiddle(), ru.getRight()))
                                .collect(joining("\n")));
                if (hook == PRE_RECEIVE) {
                    builder.environment(ENV_ALTERNATE_OBJECT_DIRS, "alternate object dirs")
                            .environment(ENV_OBJECT_DIR, "object dir")
                            .environment(ENV_QUARANTINE_PATH, "quarantine path");
                }
                hookService.handleRawRequest(builder.build(), hookOutputWriter);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }));
        thread.start();
        return thread;
    }

    private <T> void testHookWritesHookOutput(GitHook gitHook, ByteString err, ByteString out,
                                              Function<ServerCallStreamObserver<RpcHostingWriteStreamFragment>,
                                                      StreamObserver<T>> observerFactory, T request, T hookResponse)
            throws InterruptedException, IOException {
        ServerCallStreamObserver<RpcHostingWriteStreamFragment> responseObserver = makeResponseObserver();
        StreamObserver<T> observer = observerFactory.apply(responseObserver);

        observer.onNext(request);

        Triple<String, String, String> refUpdate = Triple.of(
                "e7e07b2cabebe90eb07473b7194fd2957eaf2d8c",
                "8cb293719835e3b059b4f5b89b743bbaabbecf6c",
                "refs/heads/master");
        Thread hook = invokeHook(gitHook, singletonList(refUpdate));

        observer.onNext(hookResponse);

        hook.join();

        observer.onCompleted();

        ArgumentCaptor<RpcHostingWriteStreamFragment> fragment = ArgumentCaptor.forClass(
                RpcHostingWriteStreamFragment.class);
        verify(responseObserver, times(2)).onNext(fragment.capture());

        verify(hookOutputWriter).writeStderr(err.asReadOnlyByteBuffer());
        verify(hookOutputWriter).writeStdout(out.asReadOnlyByteBuffer());
    }
}