From 70d44a52dec67be342e0a3de7365a96d3d679224 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 17 Aug 2026 21:37:50 +0200 Subject: [PATCH 1/2] chore: overhaul the test suite Replace the per-test endure boilerplate with a Start helper and dial helpers for the three connection kinds the configs need: plaintext, TLS trusting the test CA, and mutual TLS presenting the client certificate. The 22 fixed sleeps are gone. The gzip file repeated every scenario from the main file with one extra dial option, 452 lines for four cases. It now reuses the same configs and helpers, so gzip is a dial option rather than a parallel suite. Add the negative half of mutual TLS: a client that presents no certificate must be refused. The suite only ever asserted the success path, so a server that ignored client_auth_type would have passed. Split the rest by theme: request/response, TLS, gzip, reflection and observability. The status test gains an unregistered plugin name, which must report nothing rather than matching. Fix the test-certs ignore rule. It read ./tests/test-certs/**, and the leading ./ makes the pattern match nothing, so generated certificates and their private keys were never actually ignored. --- .gitignore | 2 +- tests/grpc_plugin_gzip_test.go | 452 ------------ tests/grpc_plugin_test.go | 1181 -------------------------------- tests/grpc_reflection_test.go | 144 ---- tests/grpc_test.go | 96 +++ tests/gzip_test.go | 61 ++ tests/helpers/grpc.go | 98 +++ tests/helpers/rr.go | 207 ++++++ tests/observability_test.go | 163 +++++ tests/reflection_test.go | 71 ++ tests/tls_test.go | 112 +++ 11 files changed, 809 insertions(+), 1778 deletions(-) delete mode 100644 tests/grpc_plugin_gzip_test.go delete mode 100644 tests/grpc_plugin_test.go delete mode 100644 tests/grpc_reflection_test.go create mode 100644 tests/grpc_test.go create mode 100644 tests/gzip_test.go create mode 100644 tests/helpers/grpc.go create mode 100644 tests/helpers/rr.go create mode 100644 tests/observability_test.go create mode 100644 tests/reflection_test.go create mode 100644 tests/tls_test.go diff --git a/.gitignore b/.gitignore index 43f69b8..a894f34 100644 --- a/.gitignore +++ b/.gitignore @@ -12,7 +12,7 @@ *.out # Dependency directories (remove the comment below to include it) -./tests/test-certs/** +tests/test-certs/ vendor/ .idea .vscode diff --git a/tests/grpc_plugin_gzip_test.go b/tests/grpc_plugin_gzip_test.go deleted file mode 100644 index 0a0e510..0000000 --- a/tests/grpc_plugin_gzip_test.go +++ /dev/null @@ -1,452 +0,0 @@ -package grpc_test - -import ( - "context" - "crypto/tls" - "log/slog" - "os" - "os/signal" - "sync" - "syscall" - "testing" - "time" - - "github.com/roadrunner-server/config/v6" - "github.com/roadrunner-server/endure/v2" - "github.com/roadrunner-server/logger/v6" - "github.com/roadrunner-server/resetter/v6" - "github.com/roadrunner-server/server/v6" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/health/grpc_health_v1" - - "tests/proto/service" - - grpcPlugin "github.com/roadrunner-server/grpc/v6" - rpcPlugin "github.com/roadrunner-server/rpc/v6" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials" -) - -func TestGrpcRqRsGzip(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - conn, err := grpc.NewClient("127.0.0.1:9001", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip"))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - wg.Wait() -} - -func TestGrpcRqRsMultipleGzip(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-multiple.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - conn, err := grpc.NewClient("127.0.0.1:9003", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip"))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - - hc := grpc_health_v1.NewHealthClient(conn) - hr, err := hc.Check(context.Background(), &grpc_health_v1.HealthCheckRequest{}) - require.NoError(t, err) - require.Equal(t, "SERVING", hr.Status.String()) - - watch, err := hc.Watch(context.Background(), &grpc_health_v1.HealthCheckRequest{}) - require.NoError(t, err) - - msg := &grpc_health_v1.HealthCheckResponse{} - - err = watch.RecvMsg(msg) - require.NoError(t, err) - require.Equal(t, "SERVING", msg.Status.String()) - - err = watch.CloseSend() - require.NoError(t, err) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRsTLSGzip(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-tls.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - cert, err := tls.LoadX509KeyPair("test-certs/localhost+2-client.pem", "test-certs/localhost+2-client-key.pem") - require.NoError(t, err) - - tlscfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, - } - - conn, err := grpc.NewClient("127.0.0.1:9002", grpc.WithTransportCredentials(credentials.NewTLS(tlscfg)), grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip"))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRsTLSRootCAGzip(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-tls-rootca.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - cert, err := tls.LoadX509KeyPair("test-certs/localhost+2-client.pem", "test-certs/localhost+2-client-key.pem") - require.NoError(t, err) - - tlscfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, - } - - conn, err := grpc.NewClient("127.0.0.1:9003", grpc.WithTransportCredentials(credentials.NewTLS(tlscfg)), grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip"))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRsTLS_WithResetGzip(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-tls.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - &resetter.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - cert, err := tls.LoadX509KeyPair("test-certs/localhost+2-client.pem", "test-certs/localhost+2-client-key.pem") - require.NoError(t, err) - - tlscfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, - } - - conn, err := grpc.NewClient("localhost:9002", grpc.WithTransportCredentials(credentials.NewTLS(tlscfg)), grpc.WithDefaultCallOptions(grpc.UseCompressor("gzip"))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - - // reset - t.Run("SendReset", sendReset("127.0.0.1:6009")) - - resp2, err2 := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err2) - require.Equal(t, "TOST", resp2.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - wg.Wait() -} diff --git a/tests/grpc_plugin_test.go b/tests/grpc_plugin_test.go deleted file mode 100644 index 80a24ca..0000000 --- a/tests/grpc_plugin_test.go +++ /dev/null @@ -1,1181 +0,0 @@ -package grpc_test - -import ( - "context" - "crypto/tls" - "io" - "log/slog" - "net" - "net/http" - "net/rpc" - "os" - "os/signal" - "slices" - "sync" - "syscall" - "testing" - "time" - - mocklogger "tests/mock" - "tests/proto/service" - - resetterV1 "github.com/roadrunner-server/api-go/v6/resetter/v1" - "github.com/roadrunner-server/config/v6" - "github.com/roadrunner-server/endure/v2" - goridgeRpc "github.com/roadrunner-server/goridge/v4/pkg/rpc" - grpcPlugin "github.com/roadrunner-server/grpc/v6" - "github.com/roadrunner-server/logger/v6" - "github.com/roadrunner-server/metrics/v6" - "github.com/roadrunner-server/resetter/v6" - rpcPlugin "github.com/roadrunner-server/rpc/v6" - "github.com/roadrunner-server/server/v6" - "github.com/roadrunner-server/status/v6" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - sdktrace "go.opentelemetry.io/otel/sdk/trace" - "go.opentelemetry.io/otel/sdk/trace/tracetest" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/health/grpc_health_v1" -) - -const getAddr = "http://127.0.0.1:2112/metrics" - -type inMemoryTracer struct { - tp *sdktrace.TracerProvider - exp *tracetest.InMemoryExporter -} - -func newInMemoryTracer(t *testing.T) *inMemoryTracer { - t.Helper() - exp := tracetest.NewInMemoryExporter() - tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) - t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) - return &inMemoryTracer{tp: tp, exp: exp} -} - -func (m *inMemoryTracer) Init() error { return nil } -func (m *inMemoryTracer) Name() string { return "inMemoryTracer" } -func (m *inMemoryTracer) Tracer() *sdktrace.TracerProvider { return m.tp } - -func TestGrpcInit(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-init.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - - conn, err := grpc.NewClient("127.0.0.1:9091", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcOtel(t *testing.T) { - tracer := newInMemoryTracer(t) - - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.2.0", - Path: "configs/.rr-grpc-otel.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - tracer, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - assert.NoError(t, err) - case <-sig: - err = cont.Stop() - assert.NoError(t, err) - return - case <-stopCh: - err = cont.Stop() - assert.NoError(t, err) - return - } - } - }) - - time.Sleep(time.Second) - - conn, err := grpc.NewClient("127.0.0.1:9092", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - wg.Wait() - - spans := tracer.exp.GetSpans() - spanNames := make([]string, len(spans)) - for i, s := range spans { - spanNames[i] = s.Name - } - require.True(t, slices.Contains(spanNames, "service.Echo/Ping"), "expected span 'service.Echo/Ping', got: %v", spanNames) -} - -func TestGrpcCheckStatus(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-status.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &status.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - - client := &http.Client{ - Timeout: time.Second * 10, - } - req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:35544/health?plugin=grpc", nil) - require.NoError(t, err) - - resp, err := client.Do(req) - require.NoError(t, err) - require.NotNil(t, resp) - - body, _ := io.ReadAll(resp.Body) - assert.Equal(t, `[{"plugin_name":"grpc","error_message":"","status_code":200}]`, string(body)) - assert.Equal(t, http.StatusOK, resp.StatusCode) - _ = resp.Body.Close() - - req, err = http.NewRequestWithContext(context.Background(), http.MethodGet, "http://127.0.0.1:35544/ready?plugin=grpc", nil) - require.NoError(t, err) - - resp, err = client.Do(req) - require.NoError(t, err) - require.NotNil(t, resp) - - body, _ = io.ReadAll(resp.Body) - assert.Equal(t, `[{"plugin_name":"grpc","error_message":"","status_code":200}]`, string(body)) - assert.Equal(t, http.StatusOK, resp.StatusCode) - _ = resp.Body.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -// different services, same methods inside -func TestGrpcInitDup2(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-init-duplicate-2.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 2) - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcInitMultiple(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-init-multiple.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 2) - - conn, err := grpc.NewClient("127.0.0.1:9001", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - time.Sleep(time.Second) - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRs(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - conn, err := grpc.NewClient("127.0.0.1:9001", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcFullErrorMessageIssue1193(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2024.1.0", - Path: "configs/.rr-grpc-rq-issue1193.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - _, err = cont.Serve() - assert.Error(t, err) - assert.Contains(t, err.Error(), " If you want to be c001 you just need to contribute to 0pensource.") -} - -func TestGrpcRqRsException(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-exception.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - conn, err := grpc.NewClient("127.0.0.1:9001", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.Error(t, err) - require.Equal(t, "rpc error: code = Internal desc = FOOOOOOOOOOOO", err.Error()) - require.Nil(t, resp) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRsMultiple(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-multiple.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - conn, err := grpc.NewClient("127.0.0.1:9003", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - - hc := grpc_health_v1.NewHealthClient(conn) - hr, err := hc.Check(context.Background(), &grpc_health_v1.HealthCheckRequest{}) - require.NoError(t, err) - require.Equal(t, "SERVING", hr.Status.String()) - - watch, err := hc.Watch(context.Background(), &grpc_health_v1.HealthCheckRequest{}) - require.NoError(t, err) - - msg := &grpc_health_v1.HealthCheckResponse{} - - err = watch.RecvMsg(msg) - require.NoError(t, err) - require.Equal(t, "SERVING", msg.Status.String()) - - err = watch.CloseSend() - require.NoError(t, err) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRsTLS(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-tls.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - cert, err := tls.LoadX509KeyPair("test-certs/localhost+2-client.pem", "test-certs/localhost+2-client-key.pem") - require.NoError(t, err) - - tlscfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, - } - - conn, err := grpc.NewClient("127.0.0.1:9002", grpc.WithTransportCredentials(credentials.NewTLS(tlscfg))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -func TestGrpcRqRsTLSRootCA(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-tls-rootca.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - wg := &sync.WaitGroup{} - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - cert, err := tls.LoadX509KeyPair("test-certs/localhost+2-client.pem", "test-certs/localhost+2-client-key.pem") - require.NoError(t, err) - - tlscfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, - } - - conn, err := grpc.NewClient("127.0.0.1:9003", grpc.WithTransportCredentials(credentials.NewTLS(tlscfg))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - wg.Wait() -} - -func TestGrpcRqRsTLS_WithReset(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-tls.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - &resetter.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - cert, err := tls.LoadX509KeyPair("test-certs/localhost+2-client.pem", "test-certs/localhost+2-client-key.pem") - require.NoError(t, err) - - tlscfg := &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, - } - - conn, err := grpc.NewClient("127.0.0.1:9002", grpc.WithTransportCredentials(credentials.NewTLS(tlscfg))) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - - // reset - t.Run("SendReset", sendReset("127.0.0.1:6009")) - - resp2, err2 := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err2) - require.Equal(t, "TOST", resp2.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - wg.Wait() -} - -func TestGRPCMetrics(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-metrics.yaml", - } - - l, oLogger := mocklogger.SlogTestLogger(slog.LevelDebug) - err := cont.RegisterAll( - cfg, - &server.Plugin{}, - &grpcPlugin.Plugin{}, - &metrics.Plugin{}, - l, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - tt := time.NewTimer(time.Minute * 3) - wg := &sync.WaitGroup{} - - wg.Go(func() { - defer tt.Stop() - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-tt.C: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 2) - - conn, err := grpc.NewClient("127.0.0.1:9005", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - - time.Sleep(time.Millisecond * 500) - genericOut, err := get() - assert.NoError(t, err) - assert.Contains(t, genericOut, `rr_grpc_workers_memory_bytes`) - assert.Contains(t, genericOut, `rr_grpc_worker_state`) - assert.Contains(t, genericOut, `rr_grpc_worker_memory_bytes`) - assert.Contains(t, genericOut, `rr_grpc_request_duration_seconds`) - assert.Contains(t, genericOut, `rr_grpc_request_total`) - assert.Contains(t, genericOut, `rr_grpc_requests_queue`) - - _ = conn.Close() - close(sig) - wg.Wait() - - require.Equal(t, 1, oLogger.FilterMessageSnippet("grpc server was started").Len()) - require.Equal(t, 1, oLogger.FilterMessageSnippet("method was called successfully").Len()) -} - -func sendReset(address string) func(t *testing.T) { - return func(t *testing.T) { - var d net.Dialer - conn, err := d.DialContext(context.Background(), "tcp", address) - require.NoError(t, err) - client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) - defer func() { _ = client.Close() }() - - resp := &resetterV1.Response{} - err = client.Call("resetter.Reset", &resetterV1.ResetRequest{Plugin: "grpc"}, resp) - require.NoError(t, err) - assert.True(t, resp.GetOk()) - - list := &resetterV1.PluginsList{} - err = client.Call("resetter.ListPlugins", &resetterV1.ListPluginsRequest{}, list) - require.NoError(t, err) - require.Equal(t, []string{"grpc"}, list.GetPlugins()) - } -} - -// get request and return body -func get() (string, error) { - r, err := http.Get(getAddr) //nolint:noctx - if err != nil { - return "", err - } - - b, err := io.ReadAll(r.Body) - if err != nil { - return "", err - } - - err = r.Body.Close() - if err != nil { - return "", err - } - // unsafe - return string(b), err -} - -func Test_GrpcRqOtlp(t *testing.T) { - tracer := newInMemoryTracer(t) - - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-rq-otlp.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - tracer, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - // timeout - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second * 1) - - conn, err := grpc.NewClient("127.0.0.1:9001", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - client := service.NewEchoClient(conn) - resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"}) - require.NoError(t, err) - require.Equal(t, "TOST", resp.Msg) - _ = conn.Close() - - stopCh <- struct{}{} - wg.Wait() - - spans := tracer.exp.GetSpans() - spanNames := make([]string, len(spans)) - for i, s := range spans { - spanNames[i] = s.Name - } - require.True(t, slices.Contains(spanNames, "service.Echo/Ping"), "expected span 'service.Echo/Ping', got: %v", spanNames) -} diff --git a/tests/grpc_reflection_test.go b/tests/grpc_reflection_test.go deleted file mode 100644 index 6463244..0000000 --- a/tests/grpc_reflection_test.go +++ /dev/null @@ -1,144 +0,0 @@ -package grpc_test - -import ( - "context" - "log/slog" - "os" - "os/signal" - "sync" - "syscall" - "testing" - "time" - - "github.com/roadrunner-server/config/v6" - "github.com/roadrunner-server/endure/v2" - grpcPlugin "github.com/roadrunner-server/grpc/v6" - "github.com/roadrunner-server/logger/v6" - "github.com/roadrunner-server/protoreg/v6" - rpcPlugin "github.com/roadrunner-server/rpc/v6" - "github.com/roadrunner-server/server/v6" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - grpcreflectv1 "google.golang.org/grpc/reflection/grpc_reflection_v1" -) - -// TestGrpcReflection boots the gRPC server together with the protoreg plugin and -// verifies that server reflection lists the dynamically proxied service and -// serves its full file descriptor sourced from the protoreg registry. -func TestGrpcReflection(t *testing.T) { - cont := endure.New(slog.LevelDebug) - - cfg := &config.Plugin{ - Version: "2023.3.0", - Path: "configs/.rr-grpc-reflection.yaml", - } - - err := cont.RegisterAll( - cfg, - &grpcPlugin.Plugin{}, - &rpcPlugin.Plugin{}, - &logger.Plugin{}, - &server.Plugin{}, - &protoreg.Plugin{}, - ) - assert.NoError(t, err) - - err = cont.Init() - if err != nil { - t.Fatal(err) - } - - ch, err := cont.Serve() - assert.NoError(t, err) - - sig := make(chan os.Signal, 1) - signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - - wg := &sync.WaitGroup{} - - stopCh := make(chan struct{}, 1) - - wg.Go(func() { - for { - select { - case e := <-ch: - assert.Fail(t, "error", e.Error.Error()) - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - case <-sig: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - case <-stopCh: - err = cont.Stop() - if err != nil { - assert.FailNow(t, "error", err.Error()) - } - return - } - } - }) - - time.Sleep(time.Second) - - conn, err := grpc.NewClient("127.0.0.1:9099", grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - require.NotNil(t, conn) - - assertReflection(t, conn) - - _ = conn.Close() - - stopCh <- struct{}{} - - wg.Wait() -} - -// assertReflection drives the v1 server-reflection API: it lists the registered -// services, then resolves the file descriptor that contains the proxied -// service. A non-empty descriptor proves the protoreg registry is actually -// backing reflection (plain reflection over the dynamic proxy could not serve -// the file descriptors). -func assertReflection(t *testing.T, conn *grpc.ClientConn) { - t.Helper() - - ctx, cancel := context.WithTimeout(t.Context(), time.Minute) - defer cancel() - - client := grpcreflectv1.NewServerReflectionClient(conn) - stream, err := client.ServerReflectionInfo(ctx) - require.NoError(t, err) - - err = stream.Send(&grpcreflectv1.ServerReflectionRequest{ - MessageRequest: &grpcreflectv1.ServerReflectionRequest_ListServices{ListServices: "*"}, - }) - require.NoError(t, err) - - resp, err := stream.Recv() - require.NoError(t, err) - - services := resp.GetListServicesResponse().GetService() - names := make([]string, 0, len(services)) - for _, svc := range services { - names = append(names, svc.GetName()) - } - require.Contains(t, names, "service.Echo") - - err = stream.Send(&grpcreflectv1.ServerReflectionRequest{ - MessageRequest: &grpcreflectv1.ServerReflectionRequest_FileContainingSymbol{FileContainingSymbol: "service.Echo"}, - }) - require.NoError(t, err) - - resp, err = stream.Recv() - require.NoError(t, err) - - fdResp := resp.GetFileDescriptorResponse() - require.NotNil(t, fdResp) - require.NotEmpty(t, fdResp.GetFileDescriptorProto()) -} diff --git a/tests/grpc_test.go b/tests/grpc_test.go new file mode 100644 index 0000000..cf471b2 --- /dev/null +++ b/tests/grpc_test.go @@ -0,0 +1,96 @@ +package grpc_test + +import ( + "testing" + + "tests/helpers" + "tests/proto/service" + + grpcPlugin "github.com/roadrunner-server/grpc/v6" + rpcPlugin "github.com/roadrunner-server/rpc/v6" + "github.com/roadrunner-server/server/v6" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + grpchealth "google.golang.org/grpc/health/grpc_health_v1" +) + +const ( + initAddr = "127.0.0.1:9091" + rqAddr = "127.0.0.1:9001" + multipleAddr = "127.0.0.1:9003" + exceptionAddr = "127.0.0.1:9001" + issue1193Addr = "127.0.0.1:9001" +) + +func grpcPlugins() []any { + return []any{&grpcPlugin.Plugin{}, &rpcPlugin.Plugin{}, &server.Plugin{}} +} + +// ping sends one Ping through the connection and returns the echoed message. +func ping(t *testing.T, conn *grpc.ClientConn, msg string) (string, error) { + t.Helper() + + resp, err := service.NewEchoClient(conn).Ping(t.Context(), &service.Message{Msg: msg}) + if err != nil { + return "", err + } + + return resp.GetMsg(), nil +} + +// TestBoots covers the plain init config: the server comes up and serves. +func TestBoots(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-init.yaml", grpcPlugins(), helpers.WithTCPProbe(initAddr)) +} + +// TestPingEchoesMessage is the basic request/response path over plaintext. +func TestPingEchoesMessage(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq.yaml", grpcPlugins(), helpers.WithTCPProbe(rqAddr)) + + got, err := ping(t, helpers.Dial(t, rqAddr), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +// TestMultipleProtoFiles covers a config listing more than one proto file. The +// echo service still answers, and the health service the plugin registers +// alongside it reports serving, for the whole server and for the named service. +func TestMultipleProtoFiles(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-multiple.yaml", grpcPlugins(), helpers.WithTCPProbe(multipleAddr)) + + conn := helpers.Dial(t, multipleAddr) + + got, err := ping(t, conn, "TOST") + require.NoError(t, err) + require.Equal(t, "TOST", got) + + health := grpchealth.NewHealthClient(conn) + + server, err := health.Check(t.Context(), &grpchealth.HealthCheckRequest{}) + require.NoError(t, err) + require.Equal(t, grpchealth.HealthCheckResponse_SERVING, server.GetStatus()) + + svc, err := health.Check(t.Context(), &grpchealth.HealthCheckRequest{Service: "service.Echo"}) + require.NoError(t, err) + require.Equal(t, grpchealth.HealthCheckResponse_SERVING, svc.GetStatus()) +} + +// TestWorkerExceptionIsReported checks a throwing worker surfaces as an rpc +// error rather than a hang or an empty success. +func TestWorkerExceptionIsReported(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-exception.yaml", grpcPlugins(), helpers.WithTCPProbe(exceptionAddr)) + + _, err := ping(t, helpers.Dial(t, exceptionAddr), "TOST") + + require.Error(t, err) +} + +// TestStdoutGarbageIsReportedInFull covers issue 1193. The worker prints to +// stdout, which corrupts the goridge frame, and the resulting error has to +// carry the whole offending message rather than a truncated prefix. +func TestStdoutGarbageIsReportedInFull(t *testing.T) { + err := helpers.StartExpectServeError(t, "configs/.rr-grpc-rq-issue1193.yaml", grpcPlugins()) + + require.ErrorContains(t, err, "If you want to be c001 you just need to contribute to 0pensource.") +} diff --git a/tests/gzip_test.go b/tests/gzip_test.go new file mode 100644 index 0000000..d1ddb8e --- /dev/null +++ b/tests/gzip_test.go @@ -0,0 +1,61 @@ +package grpc_test + +import ( + "testing" + + "tests/helpers" + + "github.com/roadrunner-server/resetter/v6" + "github.com/stretchr/testify/require" +) + +// The gzip cases mirror the plain ones with a compressing connection. They used +// to be a separate 452-line file repeating every scenario; the only difference +// is the dial option, so they reuse the same configs and helpers here. + +func TestPingEchoesMessageGzip(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq.yaml", grpcPlugins(), helpers.WithTCPProbe(rqAddr)) + + got, err := ping(t, helpers.Dial(t, rqAddr, helpers.Gzip()), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +func TestTLSGzip(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-tls.yaml", grpcPlugins(), helpers.WithTCPProbe(tlsAddr)) + + got, err := ping(t, helpers.DialTLS(t, tlsAddr, helpers.Gzip()), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +func TestMutualTLSGzip(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-tls-rootca.yaml", grpcPlugins(), helpers.WithTCPProbe(tlsRootCAAddr)) + + got, err := ping(t, helpers.DialMutualTLS(t, tlsRootCAAddr, helpers.Gzip()), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +func TestTLSGzipSurvivesReset(t *testing.T) { + helpers.Start(t, + "configs/.rr-grpc-rq-tls.yaml", + append(grpcPlugins(), &resetter.Plugin{}), + helpers.WithTCPProbe(tlsAddr), + ) + + conn := helpers.DialTLS(t, tlsAddr, helpers.Gzip()) + + got, err := ping(t, conn, "BEFORE") + require.NoError(t, err) + require.Equal(t, "BEFORE", got) + + resetAll(t, tlsRPCAddr) + + got, err = ping(t, conn, "AFTER") + require.NoError(t, err) + require.Equal(t, "AFTER", got) +} diff --git a/tests/helpers/grpc.go b/tests/helpers/grpc.go new file mode 100644 index 0000000..5c2f1bb --- /dev/null +++ b/tests/helpers/grpc.go @@ -0,0 +1,98 @@ +package helpers + +import ( + "crypto/tls" + "crypto/x509" + "net" + "os" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/encoding/gzip" +) + +const ( + // clientCert and clientKey are presented when the server asks for a client + // certificate (client_auth_type: require_and_verify_client_cert). + clientCert = "test-certs/localhost+2-client.pem" + clientKey = "test-certs/localhost+2-client-key.pem" + // rootCA signed all of the above; CI creates them with mkcert, see linux.yml. + rootCA = "test-certs/rootCA.pem" +) + +// Dial opens a plaintext connection and closes it on cleanup. +func Dial(t *testing.T, addr string, opts ...grpc.DialOption) *grpc.ClientConn { + t.Helper() + + conn, err := grpc.NewClient(addr, append([]grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, opts...)...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +// DialTLS opens a connection that trusts the test CA but sends no client +// certificate. +func DialTLS(t *testing.T, addr string, opts ...grpc.DialOption) *grpc.ClientConn { + t.Helper() + + conn, err := grpc.NewClient(addr, append([]grpc.DialOption{grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{ + RootCAs: testRootCAs(t), + MinVersion: tls.VersionTLS12, + }))}, opts...)...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +// DialMutualTLS opens a connection presenting the client certificate, which the +// server requires when client_auth_type is set. +func DialMutualTLS(t *testing.T, addr string, opts ...grpc.DialOption) *grpc.ClientConn { + t.Helper() + + cert, err := tls.LoadX509KeyPair(clientCert, clientKey) + require.NoError(t, err) + + conn, err := grpc.NewClient(addr, append([]grpc.DialOption{grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{cert}, + RootCAs: testRootCAs(t), + MinVersion: tls.VersionTLS12, + }))}, opts...)...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +// testRootCAs returns a pool holding the CA that signed the test certificates. +func testRootCAs(t *testing.T) *x509.CertPool { + t.Helper() + + pem, err := os.ReadFile(rootCA) + require.NoError(t, err, "generate the certificates with mkcert first, see .github/workflows/linux.yml") + + pool := x509.NewCertPool() + require.True(t, pool.AppendCertsFromPEM(pem), "%s holds no usable certificate", rootCA) + + return pool +} + +// DialRPC opens a plain tcp connection to the goridge rpc listener. +func DialRPC(t *testing.T, addr string) net.Conn { + t.Helper() + + conn, err := new(net.Dialer).DialContext(t.Context(), "tcp", addr) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +// Gzip makes the connection compress request and response bodies. +func Gzip() grpc.DialOption { + return grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name)) +} diff --git a/tests/helpers/rr.go b/tests/helpers/rr.go new file mode 100644 index 0000000..03b48cb --- /dev/null +++ b/tests/helpers/rr.go @@ -0,0 +1,207 @@ +package helpers + +import ( + "context" + "log/slog" + "net" + "net/http" + "sync" + "testing" + "time" + + mocklogger "tests/mock" + + "github.com/roadrunner-server/config/v6" + "github.com/roadrunner-server/endure/v2" + "github.com/roadrunner-server/logger/v6" + "github.com/stretchr/testify/require" +) + +const ( + // defaultConfigVersion is the config schema version used by the test configs. + defaultConfigVersion = "2023.3.0" + // probeTimeout caps how long Start waits for the server to answer the probe. + probeTimeout = time.Second * 15 + probeTick = time.Millisecond * 20 + probeDial = time.Second +) + +// bootCfg holds the options applied to a container before it is started. +type bootCfg struct { + version string + logLevel slog.Level + logger loggerKind + probe func(ctx context.Context) bool +} + +// loggerKind selects which logger plugin Start registers. +type loggerKind int + +const ( + realLogger loggerKind = iota + observedLogger +) + +// Option customizes the container built by Start. +type Option func(*bootCfg) + +// WithConfigVersion overrides the config schema version. +func WithConfigVersion(v string) Option { + return func(b *bootCfg) { b.version = v } +} + +// WithLogLevel sets the endure container log level (debug by default). +func WithLogLevel(l slog.Level) Option { + return func(b *bootCfg) { b.logLevel = l } +} + +// WithObservedLogger registers an in-memory logger instead of the real logger +// plugin and exposes the captured records as RR.Logs. +func WithObservedLogger() Option { + return func(b *bootCfg) { b.logger = observedLogger } +} + +// WithTCPProbe makes Start return only once addr accepts a connection. The +// listener binds after the worker pool is allocated, so this proves readiness +// without sending a request through the pool. +func WithTCPProbe(addr string) Option { + return func(b *bootCfg) { + b.probe = func(ctx context.Context) bool { + d := net.Dialer{Timeout: probeDial} + conn, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return false + } + + _ = conn.Close() + return true + } + } +} + +// WithProbe makes Start return only once a GET to url gets a response. This +// reaches the worker pool, so tests asserting exact log counts want +// WithTCPProbe instead. +func WithProbe(url string) Option { + return func(b *bootCfg) { + b.probe = func(ctx context.Context) bool { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return false + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return false + } + + _ = resp.Body.Close() + return true + } + } +} + +// RR is a running container. +type RR struct { + // Logs holds the captured log records, non-nil only with WithObservedLogger. + Logs *mocklogger.ObservedLogs +} + +// Start registers the plugins, boots the container and waits for the probe, if +// any, to answer. Errors arriving on the container channel are reported through +// t.Errorf and stop the container, but they do not abort the test. +// +// The returned stop is idempotent and also registered with t.Cleanup, so tests +// asserting on logs written during shutdown can stop the container mid-test. +func Start(t *testing.T, cfgPath string, plugins []any, opts ...Option) (*RR, func()) { + t.Helper() + + cont, rr, bc := newContainer(t, cfgPath, plugins, opts) + require.NoError(t, cont.Init()) + + ch, err := cont.Serve() + require.NoError(t, err) + + stopCont := sync.OnceValue(cont.Stop) + done := make(chan struct{}) + wg := &sync.WaitGroup{} + + wg.Go(func() { + for { + select { + case res := <-ch: + if res == nil { + return + } + t.Errorf("plugin %s reported an error: %v", res.VertexID, res.Error) + if errS := stopCont(); errS != nil { + t.Errorf("container stop: %v", errS) + } + case <-done: + if errS := stopCont(); errS != nil { + t.Errorf("container stop: %v", errS) + } + return + } + } + }) + + // The drain goroutine calls t.Errorf, so it has to be joined while the test + // is still running. + stop := sync.OnceFunc(func() { + close(done) + wg.Wait() + }) + t.Cleanup(stop) + + if bc.probe != nil { + require.Eventually(t, func() bool { return bc.probe(t.Context()) }, probeTimeout, probeTick, "server did not become ready") + } + + return rr, stop +} + +// newContainer builds the container and registers the config, a logger and the +// caller's plugins. The container is not initialized yet. +func newContainer(t *testing.T, cfgPath string, plugins []any, opts []Option) (*endure.Endure, *RR, *bootCfg) { + t.Helper() + + bc := &bootCfg{version: defaultConfigVersion, logLevel: slog.LevelDebug} + for _, o := range opts { + o(bc) + } + + cfg := &config.Plugin{Version: bc.version, Path: cfgPath} + + rr := &RR{} + all := []any{cfg} + + switch bc.logger { + case realLogger: + all = append(all, &logger.Plugin{}) + case observedLogger: + l, obs := mocklogger.SlogTestLogger(slog.LevelDebug) + rr.Logs = obs + all = append(all, l) + } + + cont := endure.New(bc.logLevel) + require.NoError(t, cont.RegisterAll(append(all, plugins...)...)) + + return cont, rr, bc +} + +// StartExpectServeError registers the plugins, requires Init to pass and Serve +// to fail, and returns the Serve error. +func StartExpectServeError(t *testing.T, cfgPath string, plugins []any, opts ...Option) error { + t.Helper() + + cont, _, _ := newContainer(t, cfgPath, plugins, opts) + require.NoError(t, cont.Init()) + + _, err := cont.Serve() + require.Error(t, err) + t.Cleanup(func() { _ = cont.Stop() }) + + return err +} diff --git a/tests/observability_test.go b/tests/observability_test.go new file mode 100644 index 0000000..a2c3fbe --- /dev/null +++ b/tests/observability_test.go @@ -0,0 +1,163 @@ +package grpc_test + +import ( + "io" + "net/http" + "slices" + "testing" + + "tests/helpers" + + "github.com/roadrunner-server/metrics/v6" + "github.com/roadrunner-server/status/v6" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +const ( + metricsAddr = "127.0.0.1:2112" + statusAddr = "127.0.0.1:35544" + metricsGRPCAddr = "127.0.0.1:9005" + otelAddr = "127.0.0.1:9092" + otlpAddr = "127.0.0.1:9001" +) + +// inMemoryTracer stands in for the otel plugin, collecting spans in memory so a +// test can assert on them without a collector. +type inMemoryTracer struct { + tp *sdktrace.TracerProvider + exp *tracetest.InMemoryExporter +} + +func newInMemoryTracer(t *testing.T) *inMemoryTracer { + t.Helper() + + exp := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) + t.Cleanup(func() { _ = tp.Shutdown(t.Context()) }) + + return &inMemoryTracer{tp: tp, exp: exp} +} + +func (m *inMemoryTracer) Init() error { return nil } +func (m *inMemoryTracer) Name() string { return "inMemoryTracer" } +func (m *inMemoryTracer) Tracer() *sdktrace.TracerProvider { return m.tp } + +// spanNames returns the names of every span collected so far. +func (m *inMemoryTracer) spanNames() []string { + spans := m.exp.GetSpans() + names := make([]string, len(spans)) + for i := range spans { + names[i] = spans[i].Name + } + return names +} + +// get fetches a plain http endpoint and returns its status and body. +func get(t *testing.T, url string) (int, string) { + t.Helper() + + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, url, nil) + require.NoError(t, err) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + + defer func() { require.NoError(t, resp.Body.Close()) }() + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + + return resp.StatusCode, string(body) +} + +// TestWorkerMetricsAreExported drives one call and then checks the pool gauges +// the plugin registers reach the exporter. +func TestWorkerMetricsAreExported(t *testing.T) { + helpers.Start(t, + "configs/.rr-grpc-metrics.yaml", + append(grpcPlugins(), &metrics.Plugin{}), + helpers.WithTCPProbe(metricsGRPCAddr), + ) + + got, err := ping(t, helpers.Dial(t, metricsGRPCAddr), "TOST") + require.NoError(t, err) + require.Equal(t, "TOST", got) + + code, body := get(t, "http://"+metricsAddr+"/metrics") + + require.Equal(t, http.StatusOK, code) + for _, want := range []string{ + "rr_grpc_workers_memory_bytes", + "rr_grpc_worker_state", + "rr_grpc_worker_memory_bytes", + } { + require.Contains(t, body, want) + } +} + +// TestStatusEndpoints covers health and ready for the grpc plugin, plus a name +// that is not registered. +func TestStatusEndpoints(t *testing.T) { + helpers.Start(t, + "configs/.rr-grpc-status.yaml", + append(grpcPlugins(), &status.Plugin{}), + helpers.WithTCPProbe(statusAddr), + ) + + const healthy = `[{"plugin_name":"grpc","error_message":"","status_code":200}]` + + for _, path := range []string{"/health?plugin=grpc", "/ready?plugin=grpc"} { + t.Run(path, func(t *testing.T) { + code, body := get(t, "http://"+statusAddr+path) + + require.Equal(t, http.StatusOK, code) + require.JSONEq(t, healthy, body) + }) + } + + t.Run("unknown plugin reports nothing", func(t *testing.T) { + code, body := get(t, "http://"+statusAddr+"/health?plugin=not-registered") + + require.Equal(t, http.StatusOK, code) + require.JSONEq(t, `[]`, body) + }) +} + +// TestOtelSpanIsRecorded checks the plugin opens a span named after the called +// method. +func TestOtelSpanIsRecorded(t *testing.T) { + tracer := newInMemoryTracer(t) + + helpers.Start(t, + "configs/.rr-grpc-otel.yaml", + append(grpcPlugins(), tracer), + helpers.WithTCPProbe(otelAddr), + ) + + got, err := ping(t, helpers.Dial(t, otelAddr), "TOST") + require.NoError(t, err) + require.Equal(t, "TOST", got) + + require.True(t, slices.Contains(tracer.spanNames(), "service.Echo/Ping"), + "expected a span for the called method, got: %v", tracer.spanNames()) +} + +// TestOtlpSpanIsRecorded is the same over the otlp-configured server. +func TestOtlpSpanIsRecorded(t *testing.T) { + tracer := newInMemoryTracer(t) + + helpers.Start(t, + "configs/.rr-grpc-rq-otlp.yaml", + append(grpcPlugins(), tracer), + helpers.WithTCPProbe(otlpAddr), + ) + + got, err := ping(t, helpers.Dial(t, otlpAddr), "TOST") + require.NoError(t, err) + require.Equal(t, "TOST", got) + + require.True(t, slices.Contains(tracer.spanNames(), "service.Echo/Ping"), + "expected a span for the called method, got: %v", tracer.spanNames()) +} diff --git a/tests/reflection_test.go b/tests/reflection_test.go new file mode 100644 index 0000000..f67f296 --- /dev/null +++ b/tests/reflection_test.go @@ -0,0 +1,71 @@ +package grpc_test + +import ( + "context" + "testing" + "time" + + "tests/helpers" + + "github.com/roadrunner-server/protoreg/v6" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + grpcreflectv1 "google.golang.org/grpc/reflection/grpc_reflection_v1" +) + +const reflectionAddr = "127.0.0.1:9099" + +// TestReflectionListsService drives the server-reflection API against a running +// server. +func TestReflectionListsService(t *testing.T) { + helpers.Start(t, + "configs/.rr-grpc-reflection.yaml", + append(grpcPlugins(), &protoreg.Plugin{}), + helpers.WithTCPProbe(reflectionAddr), + ) + + assertReflection(t, helpers.Dial(t, reflectionAddr)) +} + +// assertReflection drives the v1 server-reflection API: it lists the registered +// services, then resolves the file descriptor that contains the proxied +// service. A non-empty descriptor proves the protoreg registry is actually +// backing reflection (plain reflection over the dynamic proxy could not serve +// the file descriptors). +func assertReflection(t *testing.T, conn *grpc.ClientConn) { + t.Helper() + + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + + client := grpcreflectv1.NewServerReflectionClient(conn) + stream, err := client.ServerReflectionInfo(ctx) + require.NoError(t, err) + + err = stream.Send(&grpcreflectv1.ServerReflectionRequest{ + MessageRequest: &grpcreflectv1.ServerReflectionRequest_ListServices{ListServices: "*"}, + }) + require.NoError(t, err) + + resp, err := stream.Recv() + require.NoError(t, err) + + services := resp.GetListServicesResponse().GetService() + names := make([]string, 0, len(services)) + for _, svc := range services { + names = append(names, svc.GetName()) + } + require.Contains(t, names, "service.Echo") + + err = stream.Send(&grpcreflectv1.ServerReflectionRequest{ + MessageRequest: &grpcreflectv1.ServerReflectionRequest_FileContainingSymbol{FileContainingSymbol: "service.Echo"}, + }) + require.NoError(t, err) + + resp, err = stream.Recv() + require.NoError(t, err) + + fdResp := resp.GetFileDescriptorResponse() + require.NotNil(t, fdResp) + require.NotEmpty(t, fdResp.GetFileDescriptorProto()) +} diff --git a/tests/tls_test.go b/tests/tls_test.go new file mode 100644 index 0000000..8e3c35b --- /dev/null +++ b/tests/tls_test.go @@ -0,0 +1,112 @@ +package grpc_test + +import ( + "net/rpc" + "testing" + + "tests/helpers" + + resetterV1 "github.com/roadrunner-server/api-go/v6/resetter/v1" + goridgeRpc "github.com/roadrunner-server/goridge/v4/pkg/rpc" + "github.com/roadrunner-server/resetter/v6" + "github.com/stretchr/testify/require" +) + +const ( + tlsAddr = "127.0.0.1:9002" + tlsRootCAAddr = "127.0.0.1:9003" + tlsRPCAddr = "127.0.0.1:6009" + rootCARPCAddr = "127.0.0.1:6001" +) + +// TestTLS serves with a certificate and no client auth, so trusting the CA is +// enough to call. +func TestTLS(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-tls.yaml", grpcPlugins(), helpers.WithTCPProbe(tlsAddr)) + + got, err := ping(t, helpers.DialTLS(t, tlsAddr), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +// TestMutualTLS uses client_auth_type require_and_verify_client_cert, so the +// call only succeeds when the client presents its certificate. +func TestMutualTLS(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-tls-rootca.yaml", grpcPlugins(), helpers.WithTCPProbe(tlsRootCAAddr)) + + got, err := ping(t, helpers.DialMutualTLS(t, tlsRootCAAddr), "TOST") + + require.NoError(t, err) + require.Equal(t, "TOST", got) +} + +// TestMutualTLSRejectsClientWithoutCertificate is the negative half the suite +// was missing: without a client certificate the server must refuse the call. +func TestMutualTLSRejectsClientWithoutCertificate(t *testing.T) { + helpers.Start(t, "configs/.rr-grpc-rq-tls-rootca.yaml", grpcPlugins(), helpers.WithTCPProbe(tlsRootCAAddr)) + + _, err := ping(t, helpers.DialTLS(t, tlsRootCAAddr), "TOST") + + require.Error(t, err, "the server accepted a client that presented no certificate") +} + +// TestTLSSurvivesReset resets the worker pool over rpc and checks the server is +// still usable afterwards, which is where a botched pool swap would show up. +func TestTLSSurvivesReset(t *testing.T) { + helpers.Start(t, + "configs/.rr-grpc-rq-tls.yaml", + append(grpcPlugins(), &resetter.Plugin{}), + helpers.WithTCPProbe(tlsAddr), + ) + + conn := helpers.DialTLS(t, tlsAddr) + + got, err := ping(t, conn, "BEFORE") + require.NoError(t, err) + require.Equal(t, "BEFORE", got) + + resetAll(t, tlsRPCAddr) + + got, err = ping(t, conn, "AFTER") + require.NoError(t, err) + require.Equal(t, "AFTER", got) +} + +// TestMutualTLSSurvivesReset is the same over the mutually authenticated port. +func TestMutualTLSSurvivesReset(t *testing.T) { + helpers.Start(t, + "configs/.rr-grpc-rq-tls-rootca.yaml", + append(grpcPlugins(), &resetter.Plugin{}), + helpers.WithTCPProbe(tlsRootCAAddr), + ) + + conn := helpers.DialMutualTLS(t, tlsRootCAAddr) + + got, err := ping(t, conn, "BEFORE") + require.NoError(t, err) + require.Equal(t, "BEFORE", got) + + resetAll(t, rootCARPCAddr) + + got, err = ping(t, conn, "AFTER") + require.NoError(t, err) + require.Equal(t, "AFTER", got) +} + +// resetAll asks the resetter plugin to rebuild every pool. +func resetAll(t *testing.T, rpcAddr string) { + t.Helper() + + conn := helpers.DialRPC(t, rpcAddr) + client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) + t.Cleanup(func() { _ = client.Close() }) + + var plugins resetterV1.PluginsList + require.NoError(t, client.Call("resetter.ListPlugins", &resetterV1.ListPluginsRequest{}, &plugins)) + require.Contains(t, plugins.GetPlugins(), "grpc") + + var out resetterV1.Response + require.NoError(t, client.Call("resetter.Reset", &resetterV1.ResetRequest{Plugin: "grpc"}, &out)) + require.True(t, out.GetOk()) +} From 5b091094552a59da2f7e9e568b3fc3c17ee4f9e7 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 17 Aug 2026 21:37:51 +0200 Subject: [PATCH 2/2] chore: measure the whole module in both tiers coverpkg named the root package only, so codec, parser and proxy went unmeasured, and the e2e step listed test files by name so a new file would not run. Drop -failfast, unfold the ee step from a single escaped line, and fail the codecov job when the merged summary maps to no plugin source. --- .github/workflows/linux.yml | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index 3b92e77..702bebd 100644 --- a/.github/workflows/linux.yml +++ b/.github/workflows/linux.yml @@ -60,9 +60,24 @@ jobs: repo-token: ${{ secrets.GITHUB_TOKEN }} - name: Run golang root module tests run: | - go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/roadrunner-server/grpc/v6 -coverprofile=./tests/coverage-ci/grpc_root.out -covermode=atomic . ./codec ./parser ./protoc_plugins ./proxy + go test -timeout 20m -v -race -cover -tags=debug -coverpkg=./... -coverprofile=./tests/coverage-ci/grpc_root.out -covermode=atomic ./... - name: Run ee tests with coverage - run: "cd tests\n\nsudo apt update\nsudo apt install -y libnss3-tools\ncurl -JLO \"https://dl.filippo.io/mkcert/latest?for=linux/amd64\"\nchmod +x mkcert-v*-linux-amd64\nsudo cp mkcert-v*-linux-amd64 /usr/local/bin/mkcert\nmkcert -install\nmkcert localhost 127.0.0.1 ::1\nmkcert -client localhost 127.0.0.1 ::1 \nmkdir test-certs\ncp -r localhost+2-client-key.pem localhost+2-client.pem localhost+2-key.pem localhost+2.pem test-certs/\ncp -r $(mkcert -CAROOT)/rootCA.pem test-certs/\n\ngo test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/roadrunner-server/grpc/v6 -coverprofile=./coverage-ci/grpc.out -covermode=atomic grpc_plugin_gzip_test.go grpc_plugin_test.go grpc_reflection_test.go\n" + run: | + cd tests + + sudo apt update + sudo apt install -y libnss3-tools + curl -JLO "https://dl.filippo.io/mkcert/latest?for=linux/amd64" + chmod +x mkcert-v*-linux-amd64 + sudo cp mkcert-v*-linux-amd64 /usr/local/bin/mkcert + mkcert -install + mkcert localhost 127.0.0.1 ::1 + mkcert -client localhost 127.0.0.1 ::1 + mkdir test-certs + cp -r localhost+2-client-key.pem localhost+2-client.pem localhost+2-key.pem localhost+2.pem test-certs/ + cp -r "$(mkcert -CAROOT)/rootCA.pem" test-certs/ + + go test -timeout 20m -v -race -cover -tags=debug -coverpkg=github.com/roadrunner-server/grpc/v6/... -coverprofile=./coverage-ci/grpc.out -covermode=atomic ./... - name: Archive code coverage results uses: actions/upload-artifact@v7 with: @@ -98,6 +113,12 @@ jobs: } ' summary.txt > summary.filtered.txt mv summary.filtered.txt summary.txt + # a profile that maps to no plugin source uploads fine and reports 0% + blocks=$(($(wc -l < summary.txt) - 1)) + if [ "$blocks" -lt 10 ]; then + echo "::error::coverage summary holds $blocks blocks, the profile does not map to plugin sources" + exit 1 + fi - name: upload to codecov uses: codecov/codecov-action@v7 # Docs: with: