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: 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()) +}