test(generate): a Python client streams all four RPC shapes through the companion

cmd/generate/client_streaming_test.go

@@ -0,0 +1,67 @@
+package generate
+
+import (
+	"bytes"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+
+	coreproto "github.com/codefly-dev/core/companions/proto"
+	"github.com/codefly-dev/core/composition"
+	"github.com/codefly-dev/core/resources"
+	"github.com/stretchr/testify/require"
+	googleproto "google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/descriptorpb"
+)
+
+// Exercise the published companion through `codefly generate client`. Checking
+// Core's template source alone misses a companion that still ships the old one.
+func TestGenerateClientPythonStreaming(t *testing.T) {
+	requireQualify(t)
+	ctx := t.Context()
+	root := t.TempDir()
+	protoDir := filepath.Join(root, "proto")
+	writeTestFile(t, filepath.Join(protoDir, "buf.yaml"), "version: v2\nmodules:\n  - path: .\n")
+	fixture, err := os.ReadFile("testdata/python_streaming/shapes.proto")
+	require.NoError(t, err)
+	writeTestFile(t, filepath.Join(protoDir, "shapes.proto"), string(fixture))
+	harness, err := os.ReadFile("testdata/python_streaming/exercise.py")
+	require.NoError(t, err)
+	contract, err := buildDescriptorSet(ctx, protoDir)
+	require.NoError(t, err)
+	var set descriptorpb.FileDescriptorSet
+	require.NoError(t, googleproto.Unmarshal(contract, &set))
+	contracts := writeContractsCatalog(t, root, "codefly/shapes", contract, composition.APIContractEndpoint{
+		Service: "api", Endpoint: "connect", API: "connect",
+		Kind: composition.APIContractKindProtobuf, Package: "shapes.v1",
+		Services: composition.ProtobufServices(&set, "shapes.v1"),
+	})
+	workspace := &resources.Workspace{Name: "streaming-test", Layout: resources.LayoutKindModules}
+	require.NoError(t, workspace.SaveToDirUnsafe(ctx, root))
+	t.Chdir(root)
+	resetClientFlags(t)
+	clientFrom, clientLanguages = "contracts:"+contracts, []string{"python"}
+	clientName, clientModuleName = "shapes-client", "shapes"
+	clientOutput = filepath.Join(root, "library")
+	require.NoError(t, ClientCmd.RunE(ClientCmd, nil))
+
+	pythonDir := filepath.Join(clientOutput, "python")
+	facade := filepath.Join(pythonDir, "shapes_client", "shapes.py")
+	for method, seam := range map[string]string{
+		"Echo": "unary", "Fan": "unary_stream", "Gather": "stream_unary", "Chat": "stream_stream",
+	} {
+		assertFileContains(t, facade, `self._gateway.`+seam+`("/shapes.v1.EchoService/`+method+`",`)
+	}
+	// Use the same generated bindings and runtime as the companion. Requests
+	// travel through the generated facade, never a hand-maintained RPC wrapper.
+	image, err := coreproto.CompanionImage(ctx)
+	require.NoError(t, err)
+	cmd := exec.CommandContext(ctx, "docker", "run", "--rm", "-i",
+		"-v", pythonDir+":/out:ro", "-w", "/out", image.FullName(),
+		"/venv/bin/python", "-")
+	cmd.Stdin = bytes.NewReader(harness)
+	out, err := cmd.CombinedOutput()
+	require.NoError(t, err, "%s", out)
+	t.Logf("%s", out)
+}

cmd/generate/testdata/python_streaming/exercise.py

@@ -0,0 +1,70 @@
+"""Exercise every generated facade method against a real gRPC service.
+
+The facade supplies procedure, message type and RPC shape. Transport selection
+belongs to its supplied Gateway, as it does for every generated Python client.
+"""
+from concurrent import futures
+
+import grpc
+from shapes_client.shapes import shapes
+from shapes_client._gen import shapes_pb2 as pb
+from shapes_client._gen import shapes_pb2_grpc as rpc
+
+
+class Echo(rpc.EchoServiceServicer):
+    def Echo(self, request, context):
+        return request
+
+    def Fan(self, request, context):
+        for suffix in ("1", "2", "3"):
+            yield pb.Message(text=request.text + suffix)
+
+    def Gather(self, requests, context):
+        return pb.Message(text=",".join(request.text for request in requests))
+
+    def Chat(self, requests, context):
+        for request in requests:
+            yield pb.Message(text=request.text.upper())
+
+
+class Gateway:
+    """The existing gateway seam contract over grpcio's four multicallables."""
+    def __init__(self, channel):
+        self.channel = channel
+
+    def call(self, shape, procedure, request, response_type):
+        return getattr(self.channel, shape)(
+            procedure,
+            request_serializer=lambda message: message.SerializeToString(),
+            response_deserializer=response_type.FromString,
+        )(request, timeout=10)
+
+    def unary(self, procedure, request, response_type):
+        return self.call("unary_unary", procedure, request, response_type)
+
+    def unary_stream(self, procedure, request, response_type):
+        return self.call("unary_stream", procedure, request, response_type)
+
+    def stream_unary(self, procedure, requests, response_type):
+        return self.call("stream_unary", procedure, requests, response_type)
+
+    def stream_stream(self, procedure, requests, response_type):
+        return self.call("stream_stream", procedure, requests, response_type)
+
+
+server = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
+rpc.add_EchoServiceServicer_to_server(Echo(), server)
+port = server.add_insecure_port("127.0.0.1:0")
+server.start()
+try:
+    with grpc.insecure_channel(f"127.0.0.1:{port}") as channel:
+        client = shapes(Gateway(channel)).echo()
+        assert client.echo(pb.Message(text="one")).text == "one"
+        assert [r.text for r in client.fan(pb.Message(text="item"))] == ["item1", "item2", "item3"]
+        assert client.gather(iter(pb.Message(text=x) for x in ("a", "b", "c"))).text == "a,b,c"
+        assert client.gather(iter(())).text == ""
+        assert [r.text for r in client.chat(iter(pb.Message(text=x) for x in ("a", "b", "c")))] == ["A", "B", "C"]
+        assert list(client.chat(iter(()))) == []
+finally:
+    server.stop(None).wait()
+print("all four generated RPC shapes passed")

cmd/generate/testdata/python_streaming/shapes.proto

@@ -0,0 +1,10 @@
+syntax = "proto3";
+package shapes.v1;
+
+message Message { string text = 1; }
+service EchoService {
+  rpc Echo(Message) returns (Message);
+  rpc Fan(Message) returns (stream Message);
+  rpc Gather(stream Message) returns (Message);
+  rpc Chat(stream Message) returns (stream Message);
+}