package com.atlassian.bitbucket.mesh.git.httpbackend;

import com.atlassian.bitbucket.mesh.AbstractGrpcTest;
import com.atlassian.bitbucket.mesh.config.ApplicationConfig;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.cache.PackCacheService;
import com.atlassian.bitbucket.mesh.git.hosting.GitPacketStream;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.process.handler.StringStdioHandler;
import com.atlassian.bitbucket.mesh.repository.Repository;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.test.annotation.Repo;
import com.atlassian.bitbucket.mesh.util.PackfileListener;
import com.atlassian.bitbucket.mesh.util.Version;
import com.atlassian.util.contentcache.CacheEntry;
import com.atlassian.util.contentcache.CacheStatistics;
import com.google.common.collect.ImmutableList;
import com.google.protobuf.ByteString;
import com.zaxxer.nuprocess.NuProcess;
import io.grpc.ManagedChannel;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

import javax.annotation.Resource;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import static com.atlassian.bitbucket.mesh.git.hosting.GitPackets.FLUSH_PACKET;
import static com.atlassian.bitbucket.mesh.util.MoreCollectors.toImmutableList;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.hamcrest.Matchers.*;
import static org.junit.Assume.assumeThat;

/**
 * Verifies {@link RpcHttpBackendRequest} without using {@code git remote-http} as a client.
 * <p>
 * {@code git http-backend} reads and writes data in Git's {@code pkt-line} format (for format details,
 * <a href="https://github.com/git/git/blob/v2.27.0/Documentation/technical/http-protocol.txt">read this</a>).
 * That makes the tests here a little more complicated, because they need to parse the {@code pkt-line} format
 * so they can process the pack without unexpected {@code pkt-line} cruft.
 */
public class HttpBackendRequestTest extends AbstractGrpcTest {

    private static final Version VERSION_2_18 = new Version(2, 18);

    @Rule
    public final TemporaryDirectory tempDir = new TemporaryDirectory();

    @Resource
    private GitProcessBuilderFactory builderFactory;
    private CacheStatistics cacheStatistics;
    @Resource
    private ApplicationConfig config;
    @Resource
    private ManagedChannel managedChannel;
    @Resource
    private PackCacheService packCacheService;
    @Repo
    private Repository repository;
    private HostingServiceGrpc.HostingServiceStub stub;
    private Version version;

    @Before
    public void setup() {
        cacheStatistics = packCacheService.getStatistics();
        packCacheService.clear();
        stub = HostingServiceGrpc.newStub(managedChannel);
        version = config.getGit().getBinary().getVersion();
    }

    @Test
    public void testCapabilitiesProtocolV2() throws Exception {
        assumeThat(version, greaterThanOrEqualTo(VERSION_2_18));

        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-upload-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-upload-pack-request")
                        .setPathInfo("/pkey/slug.git/info/refs")
                        .setQueryString("service=git-upload-pack")
                        .setRequestMethod("GET")
                        .build())
                .setProtocolHeader(protocolVersion(2))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, "", tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_CAPABILITIES);
        assertEquals(2, result.getProtocolVersion());
    }

    @Test
    public void testCapabilitiesProtocolV3() throws Exception {
        //Note that protocol "v3" does not exist (yet) - but if it is requested, the handling for it
        //should be the same as "v2".
        assumeThat(version, greaterThanOrEqualTo(VERSION_2_18));

        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-upload-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-upload-pack-request")
                        .setPathInfo("/pkey/slug.git/info/refs")
                        .setQueryString("service=git-upload-pack")
                        .setRequestMethod("GET")
                        .build())
                .setProtocolHeader(protocolVersion(3))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, "", tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_CAPABILITIES);
        assertEquals(3, result.getProtocolVersion());
    }

    @Test
    public void testReceivePackRefAdvertisement() throws Exception {
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("*/*")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setPathInfo("/info/refs")
                        .setQueryString("service=git-receive-pack")
                        .setRequestMethod("GET")
                        .build())
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, "", tempDir.newFile("never-written.pack"), 2);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertEquals(8, responseObserver.getRefs().size());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_REF_ADVERTISEMENT);
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testRefAdvertisementWithoutProtocolVersion() throws Exception {
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("*/*")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setPathInfo("/info/refs")
                        .setQueryString("service=git-upload-pack")
                        .setRequestMethod("GET")
                        .build())
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, "", tempDir.newFile("never-written.pack"), 2);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertEquals(12, responseObserver.getRefs().size());
        //Skip the first 2 lines (service and capability declarations)
        assertV1Refs(responseObserver.getRefs().stream().skip(2).collect(toImmutableList()));
        //The request was not cacheable, so the cache should be empty
        assertEquals(0L, cacheStatistics.getSize());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_REF_ADVERTISEMENT);
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testRefAdvertisementProtocolV0() throws Exception {
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("*/*")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setPathInfo("/info/refs")
                        .setQueryString("service=git-upload-pack")
                        .setRequestMethod("GET")
                        .build())
                .setProtocolHeader(protocolVersion(0))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, "", tempDir.newFile("never-written.pack"), 2);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertEquals(12, responseObserver.getRefs().size());
        //Skip the first 2 lines (service and capability declarations)
        assertV1Refs(responseObserver.getRefs().stream().skip(2).collect(toImmutableList()));
        //The request was not cacheable, so the cache should be empty
        assertEquals(0L, cacheStatistics.getSize());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_REF_ADVERTISEMENT);
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testRefAdvertisementProtocolV1() throws Exception {
        //Note that "v1" is the same as "v0" but with a "version 1" line included in the output.
        //(The only way to get a "v1" response is by specifically requesting it, as it was a transition/detection
        //protocol and not universally supported.)
        assumeThat(version, greaterThanOrEqualTo(VERSION_2_18));

        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("*/*")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setPathInfo("/info/refs")
                        .setQueryString("service=git-upload-pack")
                        .setRequestMethod("GET")
                        .build())
                .setProtocolHeader(protocolVersion(1))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, "", tempDir.newFile("never-written.pack"), 2);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertEquals(13, responseObserver.getRefs().size());
        //Skip the first 3 lines (service, version and capability declarations)
        assertV1Refs(responseObserver.getRefs().stream().skip(3).collect(toImmutableList()));
        //The request was not cacheable, so the cache should be empty
        assertEquals(0L, cacheStatistics.getSize());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_REF_ADVERTISEMENT);
        assertEquals(1, result.getProtocolVersion());
    }

    @Test
    public void testRefAdvertisementProtocolV2() throws Exception {
        assumeThat(version, greaterThanOrEqualTo(VERSION_2_18));

        String stdin = "0014command=ls-refs\n" +
                "0014agent=git/2.28.000010009peel\n" +
                "000csymrefs\n" +
                "0014ref-prefix HEAD\n" +
                "001bref-prefix refs/heads/\n" +
                FLUSH_PACKET;

        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-upload-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-upload-pack-request")
                        .setPathInfo("/git-upload-pack")
                        .setRequestMethod("POST")
                        .build())
                .setProtocolHeader(protocolVersion(2))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertRefs(responseObserver.getRefs());
        //The request was not cacheable, so the cache should be empty
        assertEquals(0L, cacheStatistics.getSize());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_REF_ADVERTISEMENT);
        assertEquals(2, result.getProtocolVersion());
    }

    @Test
    public void testNegotiationProtocolV0() throws Exception {
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-upload-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-upload-pack-request")
                        .setPathInfo("/git-upload-pack")
                        .setRequestMethod("POST")
                        .build())
                .setProtocolHeader(protocolVersion(0))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, FLUSH_PACKET, tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_NEGOTIATION);
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testUploadPack() throws Exception {
        Path pack = tempDir.newFile("test.pack");
        String stdin = getUploadPackArgs("0a943a29376f2336b78312d99e65da17048951db");
        RpcHttpBackendRequest request = createBuilder()
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, pack, 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertPackFile(pack);
        //Verify that the pack file has been cached
        Map<String, CacheEntry> entries = packCacheService.getCache(repository).getEntries();
        assertThat(entries.size(), equalTo(1));
        assertThat(cacheStatistics.getHits(), equalTo(0L));
        assertThat(cacheStatistics.getMisses(), equalTo(1L));
        assertHostingResult(result, RpcPackCacheResult.PACK_CACHE_MISS, RpcHostingOperationType.HOSTING_OP_CLONE);
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testUploadPackErrorsOnUnknownWant() throws Exception {
        String stdin = getUploadPackArgs("58b8128a3a3aed8add0eb86246590783ec9ed785");
        RpcHttpBackendRequest request = createBuilder()
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertNotEquals(0, result.getExitCode());
        assertThat(responseObserver.getStderr(), containsString("fatal: git upload-pack: not our ref"));
    }

    @Test
    public void testUploadPackFromCache() throws Exception {
        Path pack = tempDir.newFile("test.pack");
        RpcHttpBackendRequest request = createBuilder()
                .setRepository(repository.getId())
                .build();
        String stdin = getUploadPackArgs("0a943a29376f2336b78312d99e65da17048951db");

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, pack, 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        //Verify that the pack file has been cached
        Map<String, CacheEntry> entries = packCacheService.getCache(repository).getEntries();
        assertThat(entries.size(), equalTo(1));
        assertThat(cacheStatistics.getHits(), equalTo(0L));
        assertThat(cacheStatistics.getMisses(), equalTo(1L));
        assertHostingResult(result, RpcPackCacheResult.PACK_CACHE_MISS, RpcHostingOperationType.HOSTING_OP_CLONE);

        //Make the same request again which should result in a cache hit (the pack file is served from the cache)
        Path packFromCache = tempDir.newFile("fromcache.pack");
        responseObserver = new HttpBackendResponseObserver(
                request, stdin, packFromCache, 1);

        stub.httpBackend(responseObserver);
        result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertPackFile(packFromCache);

        //Verify that the request resulted in a cache hit
        assertThat(cacheStatistics.getHits(), equalTo(1L));
        assertThat(cacheStatistics.getMisses(), equalTo(1L));
        assertHostingResult(result, RpcPackCacheResult.PACK_CACHE_HIT, RpcHostingOperationType.HOSTING_OP_CLONE);
        assertEquals(0, result.getProtocolVersion());
    }

    /**
     * When a git push is done over HTTP with a large pack file, the client sends two HTTP requests.
     *
     * The first one looks like this:
     *
     *   POST http://localhost:7990/plugins/servlet/scm/TEST/test2.git/git-receive-pack
     *   authorization: Basic YWRtaW46YWRtaW4=
     *   user-agent: git/2.20.1
     *   host: localhost:7990
     *   content-type: application/x-git-receive-pack-request
     *   accept: application/x-git-receive-pack-result
     *   content-length: 4
     *   0000
     *
     * Following this request, the client will send a second HTTP request with the actual pack file using 'chunked'
     * transfer-encoding.
     *
     * This test is for the scenario of the first request.
     */
    @Test
    public void testReceivePackForEmptyPayload() throws Exception {
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-receive-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-receive-pack-request")
                        .setPathInfo("/git-receive-pack")
                        .setRequestMethod("POST")
                        .build())
                .setProtocolHeader(protocolVersion(2))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, FLUSH_PACKET, tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_UNSPECIFIED);
        assertEquals(2, result.getProtocolVersion());
    }

    @Test
    public void testReceivePackForPush() throws Exception {
        //Length of input is 98 bytes = 0x62
        String stdin = "006258b8128a3a3aed8add0eb86246590783ec9ed785 58b8128a3a3aed8add0eb86246590783ec9ed785 My todo list" +
                FLUSH_PACKET;
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-receive-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-receive-pack-request")
                        .setPathInfo("/git-receive-pack")
                        .setRequestMethod("POST")
                        .build())
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, tempDir.newFile("never-written.pack"), 1);

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_PUSH);
        //If no version is given in the request then protocol version 0 will be used
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testPushWithRejectedPreReceiveHook() throws Exception {
        //Length of input is 106 bytes = 0x6a
        String stdin = "006a12ebe2a58367347cd39f19f5a72f3cbec7b8f9a9 0000000000000000000000000000000000000000 refs/tags/signed_tag" +
                FLUSH_PACKET;
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-receive-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-receive-pack-request")
                        .setPathInfo("/git-receive-pack")
                        .setRequestMethod("POST")
                        .build())
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, tempDir.newFile("never-written.pack"), 1) {

                @Override
                protected void onPreReceive(RpcHostingPreReceiveRequest preReceive) {
                    requestStream.onNext(RpcHttpBackendFragment.newBuilder()
                            .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                                    .addOutput(RpcHookChunk.newBuilder()
                                            .setChannel(RpcHookChunk.Channel.CHANNEL_STDERR)
                                            .setData(ByteString.copyFrom("Do not delete", UTF_8)))
                                    .setRejected(true))
                            .build());
                }
            };

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_PUSH);
        assertEquals(RpcHookResult.HOOK_RESULT_REJECTED, result.getHookResult());
        assertThat(responseObserver.getStderr(), containsString("Do not delete"));

        //If no version is given in the request then protocol version 0 will be used
        assertEquals(0, result.getProtocolVersion());
    }

    @Test
    public void testPushWithRejectedPreReceiveHookV2() throws Exception {
        //Length of input is 106 bytes = 0x6a
        String stdin = "006a12ebe2a58367347cd39f19f5a72f3cbec7b8f9a9 0000000000000000000000000000000000000000 refs/tags/signed_tag" +
                FLUSH_PACKET;
        RpcHttpBackendRequest request = RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-receive-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate, gzip"))
                        .setContentType("application/x-git-receive-pack-request")
                        .setPathInfo("/git-receive-pack")
                        .setRequestMethod("POST")
                        .build())
                .setProtocolHeader(protocolVersion(2))
                .setRepository(repository.getId())
                .build();

        HttpBackendResponseObserver responseObserver = new HttpBackendResponseObserver(
                request, stdin, tempDir.newFile("never-written.pack"), 1) {

            @Override
            protected void onPreReceive(RpcHostingPreReceiveRequest preReceive) {
                requestStream.onNext(RpcHttpBackendFragment.newBuilder()
                        .setPreReceive(RpcHostingPreReceiveResponse.newBuilder()
                                .addOutput(RpcHookChunk.newBuilder()
                                        .setChannel(RpcHookChunk.Channel.CHANNEL_STDERR)
                                        .setData(ByteString.copyFrom("Do not delete", UTF_8)))
                                .setRejected(true))
                        .build());
            }
        };

        stub.httpBackend(responseObserver);
        RpcHostingStreamResult result = responseObserver.asResult();

        assertEquals(0, result.getExitCode());
        assertHostingResult(result, RpcHostingOperationType.HOSTING_OP_PUSH);
        assertEquals(RpcHookResult.HOOK_RESULT_REJECTED, result.getHookResult());
        assertThat(responseObserver.getStderr(), containsString("Do not delete"));

        //Verify that protocol 2 was detected and used
        assertEquals(2, result.getProtocolVersion());
    }

    private static void assertHostingResult(RpcHostingStreamResult actualResult,
                                            RpcHostingOperationType expectedOperationType) {
        assertNotNull(actualResult);
        assertThat(actualResult.getOperationType(), equalTo(expectedOperationType));
    }

    private static void assertHostingResult(RpcHostingStreamResult actualResult, RpcPackCacheResult expectedCacheResult,
                                            RpcHostingOperationType expectedOperationType) {
        assertHostingResult(actualResult, expectedOperationType);
        assertThat(actualResult.getCacheResult(), equalTo(expectedCacheResult));
    }

    @SuppressWarnings("unchecked")
    private static void assertRefs(List<String> refs) {
        assertThat(refs, contains(
                containsString("00520a943a29376f2336b78312d99e65da17048951db HEAD symref-target:refs/heads/master"),
                containsString("0048d6edcbf924697ab811a867421dab60d954ccad99 refs/heads/basic_branching"),
                containsString("00496053a1eaa1c009dd11092d09a72f3c41af1b59ad refs/heads/branch_mod_merge"),
                containsString("003f0a943a29376f2336b78312d99e65da17048951db refs/heads/master"),
                containsString("004c2e10dd2d1d5eea9291b296e78312e8a703964a95 refs/heads/out_of_order_branch")));
    }

    @SuppressWarnings("unchecked")
    private static void assertV1Refs(List<String> refs) {
        assertThat(refs, contains(
                containsString("0048d6edcbf924697ab811a867421dab60d954ccad99 refs/heads/basic_branching"),
                containsString("00496053a1eaa1c009dd11092d09a72f3c41af1b59ad refs/heads/branch_mod_merge"),
                containsString("003f0a943a29376f2336b78312d99e65da17048951db refs/heads/master"),
                containsString("004c2e10dd2d1d5eea9291b296e78312e8a703964a95 refs/heads/out_of_order_branch"),
                containsString("004f80aa5be5ecf39660f798858482254f7a2ae9110e refs/tags/backdated_annotated_tag"),
                containsString("005257150c54c38d6570b2fd5e6d6cfc19476de44e84 refs/tags/backdated_annotated_tag^{}"),
                containsString("004b9b4e781dea0769888fe270e06ad76675f73851b1 refs/tags/retagged_signed_tag"),
                containsString("004e0a943a29376f2336b78312d99e65da17048951db refs/tags/retagged_signed_tag^{}"),
                containsString("004212ebe2a58367347cd39f19f5a72f3cbec7b8f9a9 refs/tags/signed_tag"),
                containsString("00450a943a29376f2336b78312d99e65da17048951db refs/tags/signed_tag^{}")));
    }

    private static RpcHttpBackendRequest.Builder createBuilder() {
        return RpcHttpBackendRequest.newBuilder()
                .setCgiEnvironment(RpcCgiEnvironment.newBuilder()
                        .addHttpAccept("application/x-git-upload-pack-result")
                        .addAllHttpAcceptEncoding(ImmutableList.of("deflate", "gzip"))
                        .setContentType("application/x-git-upload-pack-request")
                        .setPathInfo("/git-upload-pack")
                        .setRequestMethod("POST")
                        .build());
    }

    private static String getUploadPackArgs(String ref) {
        return "0090want " + ref +
                " multi_ack_detailed side-band-64k thin-pack ofs-delta deepen-since deepen-not agent=git/2.28.0\n" +
                "00000009done\n";
    }

    private static String protocolVersion(int version) {
        return "version=" + version;
    }

    private void assertPackFile(Path pack) {
        String hash = builderFactory.builder().command("index-pack")
                .argument(pack.toAbsolutePath().toString())
                .workDir(pack.getParent())
                .build(new StringStdioHandler())
                .run();
        assertEquals("539cf1050a0da7a7976dadd88a2671d15ac4e3a3", hash);
    }

    private static class HttpBackendResponseObserver
            implements ClientResponseObserver<RpcHttpBackendFragment, RpcHostingWriteStreamFragment> {

        protected volatile ClientCallStreamObserver<RpcHttpBackendFragment> requestStream;

        private final int expectedFlushPackets;
        private final Path pack;
        private final List<String> refs;
        private final RpcHttpBackendRequest request;
        private final CompletableFuture<RpcHostingStreamResult> resultFuture;
        private final String stdin;
        private final ByteBuffer stdout;

        private int flushPacketCount;
        private boolean inHeaders;
        private boolean inRefsSection;
        private GitPacketStream packetStream;
        private RpcHostingStreamResult result;
        private ByteArrayOutputStream stderr;
        private OutputStream writer;

        HttpBackendResponseObserver(RpcHttpBackendRequest request, String stdin, Path pack,
                                           int expectedFlushPackets) {
            this.expectedFlushPackets = expectedFlushPackets;
            this.request = request;
            this.stdin = stdin;
            this.pack = pack;

            inHeaders = true;
            refs = new ArrayList<>();
            resultFuture = new CompletableFuture<>();
            stdout = ByteBuffer.allocate(NuProcess.BUFFER_CAPACITY);
        }

        @Override
        public void beforeStart(ClientCallStreamObserver<RpcHttpBackendFragment> requestStream) {
            this.requestStream = requestStream;
        }

        @Override
        public void onCompleted() {
            resultFuture.complete(result);
        }

        @Override
        public void onError(Throwable t) {
            resultFuture.completeExceptionally(t);
        }

        @Override
        public void onNext(RpcHostingWriteStreamFragment value) {
            switch (value.getFragmentOneofCase()) {
                case RESULT:
                    onResult(value.getResult());
                    break;
                case START:
                    break;
                case STDERR:
                    onStderr(value.getStderr());
                    break;
                case STDOUT:
                    onStdout(value.getStdout());
                    break;
                case PRE_RECEIVE:
                    onPreReceive(value.getPreReceive());
                    break;
                case POST_RECEIVE:
                    onPostReceive(value.getPostReceive());
                    break;
            }
        }

        protected void onPreReceive(RpcHostingPreReceiveRequest preReceive) {
            requestStream.onNext(RpcHttpBackendFragment.newBuilder()
                    .setPreReceive(RpcHostingPreReceiveResponse.getDefaultInstance())
                    .build());
        }

        RpcHostingStreamResult asResult() throws ExecutionException, InterruptedException {
            //send the request
            requestStream.onNext(RpcHttpBackendFragment.newBuilder()
                    .setRequest(request)
                    .build());
            //send stdin
            requestStream.onNext(RpcHttpBackendFragment.newBuilder()
                    .setStdin(RpcStreamChunk.newBuilder()
                            .setClosed(true)
                            .setData(ByteString.copyFrom(stdin, UTF_8)))
                    .build());
            //wait for the result
            return resultFuture.get();
        }

        List<String> getRefs() {
            return refs;
        }

        String getStderr() {
            return stderr.toString();
        }

        private void maybeAddRef(String line) {
            //Add any line that comes prior to the flush packet (which indicates the end of the refs)
            if (line.equals(FLUSH_PACKET)) {
                flushPacketCount++;
                inRefsSection = inRefsSection && flushPacketCount <= expectedFlushPackets;
            } else if (inRefsSection) {
                refs.add(line);
            }
        }

        private void onPostReceive(RpcHostingPostReceiveRequest postReceive) {
            requestStream.onNext(RpcHttpBackendFragment.newBuilder()
                    .setPostReceive(RpcHostingPostReceiveResponse.getDefaultInstance())
                    .build());
        }

        private void onResult(RpcHostingStreamResult result) {
            this.result = result;
        }

        private void onStderr(RpcStreamChunk chunk) {
            try {
                if (stderr == null) {
                    stderr = new ByteArrayOutputStream();
                }
                chunk.getData().writeTo(stderr);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }

        private void onStdout(RpcStreamChunk chunk) {
            try {
                if (writer == null) {
                    writer = Files.newOutputStream(pack);

                    //Rather than duplicate logic for parsing Git's pkt-line format, we use the
                    //GitPacketStream to do the parsing for us to assemble full packets
                    packetStream = new GitPacketStream(false, 1024);
                    packetStream.addListener(new PackfileListener(writer, this::maybeAddRef, (buf, closed) -> { }));

                    inRefsSection = true;
                }
                for (ByteBuffer buffer : chunk.getData().asReadOnlyByteBufferList()) {
                    if (buffer.hasRemaining()) {
                        if (inHeaders) {
                            byte[] data = new byte[buffer.remaining()];
                            buffer.get(data);
                            stdout.put(data);

                            //Look for the double blank lines that separate the header from the payload
                            int index = new String(stdout.array(), 0, stdout.position(), UTF_8)
                                    .indexOf("\r\n\r\n");
                            if (index >= 0) {
                                inHeaders = false;

                                //Skip past the double blank lines, and append everything after that
                                index += 4;
                                packetStream.append(stdout.array(), index, stdout.position() - index);
                                stdout.clear();
                            }
                        } else {
                            packetStream.append(buffer);
                        }
                    }
                }

                if (chunk.getClosed()) {
                    writer.close();
                }
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }
    }
}
