Implement unary operations in conformance client (#195)

This also adds the new conformance tests to CI (but only the unary test cases for now).

.github/workflows/ci.yml

@@ -69,7 +69,7 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-go-
       - name: Run conformance tests
-        run: make conformancerun
+        run: make runconformance
       - name: Upload test reports
         uses: actions/upload-artifact@v4
         if: always()

Makefile

@@ -13,6 +13,8 @@ LICENSE_HEADER_VERSION := v1.28.1
 CONFORMANCE_VERSION := v1.0.0-rc2
 PROTOC_VERSION ?= 25.1
 GRADLE_ARGS ?=
+PROTOC := $(BIN)/protoc
+CONNECT_CONFORMANCE := $(BIN)/connectconformance
 
 UNAME_OS := $(shell uname -s)
 UNAME_ARCH := $(shell uname -m)
@@ -36,8 +38,41 @@ buildplugin: ## Build the connect-kotlin protoc plugin.
 clean: ## Cleans the underlying build.
 	./gradlew $(GRADLE_ARGS) clean
 
-.PHONY: conformancerun
-conformancerun: generate ## Run the conformance tests.
+# TODO: remove the cross-tests and rely solely on new conformance suite
+.PHONY: runconformance
+runconformance: runcrosstests runconformancenew
+
+.PHONY: runconformancenew
+runconformancenew: generate $(CONNECT_CONFORMANCE) ## Run the new conformance test suite.
+	./gradlew $(GRADLE_ARGS) conformance:client:google-java:installDist conformance:client:google-javalite:installDist
+	$(CONNECT_CONFORMANCE) -v --mode client --conf conformance/client/lite-unary-config.yaml \
+		--known-failing conformance/client/known-failing-cases.txt -- \
+		conformance/client/google-javalite/build/install/google-javalite/bin/google-javalite \
+		--style suspend
+	$(CONNECT_CONFORMANCE) -v --mode client --conf conformance/client/lite-unary-config.yaml \
+		--known-failing conformance/client/known-failing-cases.txt -- \
+		conformance/client/google-javalite/build/install/google-javalite/bin/google-javalite \
+		--style callback
+	$(CONNECT_CONFORMANCE) -v --mode client --conf conformance/client/lite-unary-config.yaml \
+		--known-failing conformance/client/known-failing-cases.txt -- \
+		conformance/client/google-javalite/build/install/google-javalite/bin/google-javalite \
+		--style blocking
+	$(CONNECT_CONFORMANCE) -v --mode client --conf conformance/client/standard-unary-config.yaml \
+		--known-failing conformance/client/known-failing-cases.txt -- \
+		conformance/client/google-java/build/install/google-java/bin/google-java \
+		--style suspend
+	$(CONNECT_CONFORMANCE) -v --mode client --conf conformance/client/standard-unary-config.yaml \
+		--known-failing conformance/client/known-failing-cases.txt -- \
+		conformance/client/google-java/build/install/google-java/bin/google-java \
+		--style callback
+	$(CONNECT_CONFORMANCE) -v --mode client --conf conformance/client/standard-unary-config.yaml \
+		--known-failing conformance/client/known-failing-cases.txt -- \
+		conformance/client/google-java/build/install/google-java/bin/google-java \
+		--style blocking
+# TODO: streaming conformance test cases
+
+.PHONY: runcrosstests
+runcrosstests: generate ## Run the old cross-test suite.
 	./gradlew $(GRADLE_ARGS) conformance:google-java:test conformance:google-javalite:test
 
 ifeq ($(UNAME_OS),Darwin)
@@ -53,18 +88,35 @@ PROTOC_OS = linux
 PROTOC_ARCH := $(UNAME_ARCH)
 endif
 
-PROTOC := $(CACHE)/protoc-$(PROTOC_VERSION).zip
-$(PROTOC):
+PROTOC_DOWNLOAD := $(CACHE)/protoc-$(PROTOC_VERSION).zip
+$(PROTOC_DOWNLOAD):
 	@if ! command -v curl >/dev/null 2>/dev/null; then echo "error: curl must be installed" >&2; exit 1; fi
 	@if ! command -v unzip >/dev/null 2>/dev/null; then echo "error: unzip must be installed" >&2; exit 1; fi
-	@rm -f $(BIN)/protoc
 	$(eval PROTOC_TMP := $(shell mktemp -d))
 	curl -sSL https://github.com/protocolbuffers/protobuf/releases/download/v$(PROTOC_VERSION)/protoc-$(PROTOC_VERSION)-$(PROTOC_OS)-$(PROTOC_ARCH).zip -o $(PROTOC_TMP)/protoc.zip
-	@mkdir -p $(BIN)
-	unzip -q $(PROTOC_TMP)/protoc.zip -d $(dir $(BIN)) bin/protoc
+	@mkdir -p $(dir $@)
+	@mv $(PROTOC_TMP)/protoc.zip $@
 	@rm -rf $(PROTOC_TMP)
+
+$(PROTOC): $(PROTOC_DOWNLOAD)
+	@mkdir -p $(BIN)
+	unzip -DD -q $(PROTOC_DOWNLOAD) -d $(dir $(BIN)) bin/protoc
+	chmod u+w $@
+
+CONNECT_CONFORMANCE_DOWNLOAD := $(CACHE)/connect-conformance-$(CONFORMANCE_VERSION).tgz
+$(CONNECT_CONFORMANCE_DOWNLOAD):
+	@if ! command -v curl >/dev/null 2>/dev/null; then echo "error: curl must be installed" >&2; exit 1; fi
+	@if ! command -v tar >/dev/null 2>/dev/null; then echo "error: tar must be installed" >&2; exit 1; fi
+	$(eval CONFORMANCE_TMP := $(shell mktemp -d))
+	curl -sSL https://github.com/connectrpc/conformance/releases/download/$(CONFORMANCE_VERSION)/connectconformance-$(CONFORMANCE_VERSION)-$(UNAME_OS)-$(UNAME_ARCH).tar.gz -o $(CONFORMANCE_TMP)/conformance.tgz
 	@mkdir -p $(dir $@)
-	@touch $@
+	@mv $(CONFORMANCE_TMP)/conformance.tgz $@
+	@rm -rf $(CONFORMANCE_TMP)
+
+$(CONNECT_CONFORMANCE): $(CONNECT_CONFORMANCE_DOWNLOAD)
+	@mkdir -p $(BIN)
+	tar -x -z -f $(CONNECT_CONFORMANCE_DOWNLOAD) -O connectconformance > $@
+	@chmod +x $@
 
 .PHONY: generate
 generate: $(PROTOC) buildplugin generateconformance generateexamples ## Generate proto files for the entire project.

conformance/client/google-java/src/main/kotlin/com/connectrpc/conformance/client/java/JavaHelpers.kt

@@ -23,20 +23,25 @@ import com.connectrpc.conformance.client.adapt.ClientCompatRequest.HttpVersion
 import com.connectrpc.conformance.client.adapt.ClientCompatRequest.TlsCreds
 import com.connectrpc.conformance.client.adapt.ClientCompatResponse
 import com.connectrpc.conformance.v1.BidiStreamResponse
+import com.connectrpc.conformance.v1.ClientCompatProto
 import com.connectrpc.conformance.v1.ClientCompatRequest.Cancel.CancelTimingCase
 import com.connectrpc.conformance.v1.ClientErrorResult
 import com.connectrpc.conformance.v1.ClientResponseResult
 import com.connectrpc.conformance.v1.ClientStreamResponse
 import com.connectrpc.conformance.v1.Codec
 import com.connectrpc.conformance.v1.Compression
+import com.connectrpc.conformance.v1.ConfigProto
 import com.connectrpc.conformance.v1.ConformancePayload
 import com.connectrpc.conformance.v1.Error
 import com.connectrpc.conformance.v1.HTTPVersion
 import com.connectrpc.conformance.v1.Header
 import com.connectrpc.conformance.v1.IdempotentUnaryResponse
 import com.connectrpc.conformance.v1.Protocol
+import com.connectrpc.conformance.v1.ServerCompatProto
 import com.connectrpc.conformance.v1.ServerStreamResponse
+import com.connectrpc.conformance.v1.ServiceProto
 import com.connectrpc.conformance.v1.StreamType
+import com.connectrpc.conformance.v1.SuiteProto
 import com.connectrpc.conformance.v1.UnaryResponse
 import com.connectrpc.conformance.v1.UnimplementedResponse
 import com.connectrpc.extensions.GoogleJavaJSONStrategy
@@ -45,6 +50,7 @@ import com.connectrpc.protocols.NetworkProtocol
 import com.google.protobuf.Any
 import com.google.protobuf.ByteString
 import com.google.protobuf.MessageLite
+import com.google.protobuf.TypeRegistry
 
 class JavaHelpers {
     companion object {
@@ -53,7 +59,7 @@ class JavaHelpers {
         fun serializationStrategy(codec: ClientCompatRequest.Codec): SerializationStrategy {
             return when (codec) {
                 ClientCompatRequest.Codec.PROTO -> GoogleJavaProtobufStrategy()
-                ClientCompatRequest.Codec.JSON -> GoogleJavaJSONStrategy()
+                ClientCompatRequest.Codec.JSON -> GoogleJavaJSONStrategy(getTypes())
                 else -> throw RuntimeException("unsupported codec $codec")
             }
         }
@@ -145,6 +151,16 @@ class JavaHelpers {
         private fun toTypeUrl(typeName: String): String {
             return if (typeName.contains('/')) typeName else TYPE_URL_PREFIX + typeName
         }
+
+        private fun getTypes(): TypeRegistry {
+            return TypeRegistry.newBuilder()
+                .add(ClientCompatProto.getDescriptor().messageTypes)
+                .add(ConfigProto.getDescriptor().messageTypes)
+                .add(ServerCompatProto.getDescriptor().messageTypes)
+                .add(ServiceProto.getDescriptor().messageTypes)
+                .add(SuiteProto.getDescriptor().messageTypes)
+                .build()
+        }
     }
 
     private class ClientCompatRequestImpl(

conformance/client/google-java/src/main/kotlin/com/connectrpc/conformance/client/java/Main.kt

@@ -15,18 +15,20 @@
 package com.connectrpc.conformance.client.java
 
 import com.connectrpc.conformance.client.Client
+import com.connectrpc.conformance.client.ClientArgs
 import com.connectrpc.conformance.client.ConformanceClientLoop
 
 fun main(args: Array<String>) {
-    val invokeStyle = ConformanceClientLoop.parseArgs(args)
+    val clientArgs = ClientArgs.parseArgs(args)
     val loop = ConformanceClientLoop(
         JavaHelpers::unmarshalRequest,
         JavaHelpers::marshalResponse,
+        clientArgs.verbosity,
     )
     val client = Client(
+        args = clientArgs,
         invokerFactory = ::JavaInvoker,
         serializationFactory = JavaHelpers::serializationStrategy,
-        invokeStyle = invokeStyle,
         payloadExtractor = JavaHelpers::extractPayload,
     )
     loop.run(System.`in`, System.out, client)

conformance/client/google-javalite/src/main/kotlin/com/connectrpc/conformance/client/javalite/Main.kt

@@ -15,18 +15,20 @@
 package com.connectrpc.conformance.client.javalite
 
 import com.connectrpc.conformance.client.Client
+import com.connectrpc.conformance.client.ClientArgs
 import com.connectrpc.conformance.client.ConformanceClientLoop
 
 fun main(args: Array<String>) {
-    val invokeStyle = ConformanceClientLoop.parseArgs(args)
+    val clientArgs = ClientArgs.parseArgs(args)
     val loop = ConformanceClientLoop(
         JavaLiteHelpers::unmarshalRequest,
         JavaLiteHelpers::marshalResponse,
+        clientArgs.verbosity,
     )
     val client = Client(
+        args = clientArgs,
         invokerFactory = ::JavaLiteInvoker,
         serializationFactory = JavaLiteHelpers::serializationStrategy,
-        invokeStyle = invokeStyle,
         payloadExtractor = JavaLiteHelpers::extractPayload,
     )
     loop.run(System.`in`, System.out, client)

conformance/client/known-failing-cases.txt

@@ -0,0 +1,3 @@
+# Test runner is overly strict in asserting the contents
+# of the message query param, so these tests currently fail.
+Idempotency/**

conformance/client/lite-unary-config.yaml

@@ -0,0 +1,24 @@
+# This configures the features that this client
+# supports and that will be verified by the
+# conformance test suite.
+features:
+  versions:
+    - HTTP_VERSION_1
+    - HTTP_VERSION_2
+  protocols:
+    - PROTOCOL_CONNECT
+    - PROTOCOL_GRPC
+    - PROTOCOL_GRPC_WEB
+  codecs:
+    - CODEC_PROTO
+    # Lite does not support JSON
+  compressions:
+    - COMPRESSION_IDENTITY
+    - COMPRESSION_GZIP
+  streamTypes:
+    # This config file only runs unary RPC test cases,
+    # so that we can run them all three ways: suspend,
+    # callback, and blocking.
+    - STREAM_TYPE_UNARY
+  supportsTlsClientCerts: true
+  supportsHalfDuplexBidiOverHttp1: true

conformance/client/src/main/kotlin/com/connectrpc/conformance/client/Client.kt

@@ -14,13 +14,16 @@
 
 package com.connectrpc.conformance.client
 
+import com.connectrpc.Code
 import com.connectrpc.ProtocolClientConfig
 import com.connectrpc.RequestCompression
+import com.connectrpc.ResponseMessage
 import com.connectrpc.SerializationStrategy
 import com.connectrpc.compression.GzipCompressionPool
 import com.connectrpc.conformance.client.adapt.AnyMessage
 import com.connectrpc.conformance.client.adapt.BidiStreamClient
 import com.connectrpc.conformance.client.adapt.ClientCompatRequest
+import com.connectrpc.conformance.client.adapt.ClientCompatRequest.Cancel
 import com.connectrpc.conformance.client.adapt.ClientCompatRequest.Codec
 import com.connectrpc.conformance.client.adapt.ClientCompatRequest.Compression
 import com.connectrpc.conformance.client.adapt.ClientCompatRequest.HttpVersion
@@ -34,6 +37,8 @@ import com.connectrpc.impl.ProtocolClient
 import com.connectrpc.okhttp.ConnectOkHttpClient
 import com.connectrpc.protocols.GETConfiguration
 import com.google.protobuf.MessageLite
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.delay
 import okhttp3.OkHttpClient
 import okhttp3.tls.HandshakeCertificates
 import okhttp3.tls.HeldCertificate
@@ -53,9 +58,9 @@ import kotlin.reflect.cast
  * RPC and returning a representation of its result.
  */
 class Client(
+    private val args: ClientArgs,
     private val invokerFactory: (ProtocolClient) -> Invoker,
     private val serializationFactory: (Codec) -> SerializationStrategy,
-    private val invokeStyle: UnaryClient.InvokeStyle,
     private val payloadExtractor: (MessageLite) -> MessageLite,
 ) {
     companion object {
@@ -108,7 +113,57 @@ class Client(
         if (req.streamType != StreamType.UNARY) {
             throw RuntimeException("specified method ${req.method} is unary but stream type indicates ${req.streamType}")
         }
-        TODO("implement me")
+        if (req.requestMessages.size != 1) {
+            throw RuntimeException("unary calls should indicate exactly one request message, got ${req.requestMessages.size}")
+        }
+        if (req.cancel != null && req.cancel !is Cancel.AfterCloseSendMs) {
+            throw RuntimeException("unary calls can only support 'AfterCloseSendMs' cancellation field, instead got ${req.cancel!!::class.simpleName}")
+        }
+        val msg = fromAny(req.requestMessages[0], client.reqTemplate, requestType)
+        val resp = CompletableDeferred<ResponseMessage<Resp>>()
+        val canceler = client.execute(
+            args.invokeStyle,
+            msg,
+            req.requestHeaders,
+            resp::complete,
+        )
+        when (val cancel = req.cancel) {
+            is Cancel.AfterCloseSendMs -> {
+                delay(cancel.millis.toLong())
+                canceler()
+            }
+            else -> {
+                // We already validated the case above.
+                // So this case means no cancellation.
+            }
+        }
+        return when (val result = resp.await()) {
+            is ResponseMessage.Success -> {
+                if (result.code != Code.OK) {
+                    throw RuntimeException("RPC was successful but ended with non-OK code ${result.code}")
+                }
+
+                ClientResponseResult(
+                    headers = result.headers,
+                    payloads = listOf(payloadExtractor(result.message)),
+                    trailers = result.trailers,
+                )
+            }
+            is ResponseMessage.Failure -> {
+                if (result.code != result.cause.code) {
+                    throw RuntimeException("RPC result has mismatching codes: ${result.code} != ${result.cause.code}")
+                }
+                if (args.verbosity > 2) {
+                    System.err.println("* client: RPC failed with code ${result.code}")
+                    result.cause.printStackTrace()
+                }
+                ClientResponseResult(
+                    headers = result.headers,
+                    error = result.cause,
+                    trailers = result.trailers,
+                )
+            }
+        }
     }
 
     private suspend fun <Req : MessageLite, Resp : MessageLite> handleClient(