Add unittests for grpc otlp client/server (#5081)

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>
open-telemetry · Mar 24, 2022 · cb868e2 · cb868e2
1 parent 22d89f0
commit cb868e2
Show file tree

Hide file tree

Showing 3 changed files with 334 additions and 4 deletions.
diff --git a/model/otlpgrpc/logs_test.go b/model/otlpgrpc/logs_test.go
@@ -15,11 +15,23 @@
 package otlpgrpc
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
+	"net"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	"go.opentelemetry.io/collector/model/internal/pdata"
 )
 
 var _ json.Unmarshaler = LogsResponse{}
@@ -60,3 +72,101 @@ func TestLogsRequestJSON(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, strings.Join(strings.Fields(string(logsRequestJSON)), ""), string(got))
 }
+
+func TestLogsRequestJSON_Deprecated(t *testing.T) {
+	lr, err := UnmarshalJSONLogsRequest(logsRequestJSON)
+	assert.NoError(t, err)
+	assert.Equal(t, "test_log_record", lr.Logs().ResourceLogs().At(0).InstrumentationLibraryLogs().At(0).LogRecords().At(0).Body().AsString())
+
+	got, err := lr.MarshalJSON()
+	assert.NoError(t, err)
+	assert.Equal(t, strings.Join(strings.Fields(string(logsRequestJSON)), ""), string(got))
+}
+
+func TestLogsGrpc(t *testing.T) {
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	RegisterLogsServer(s, &fakeLogsServer{t: t})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		assert.NoError(t, s.Serve(lis))
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+		wg.Wait()
+	})
+
+	cc, err := grpc.Dial("bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock())
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, cc.Close())
+	})
+
+	logClient := NewLogsClient(cc)
+
+	resp, err := logClient.Export(context.Background(), generateLogsRequest())
+	assert.NoError(t, err)
+	assert.Equal(t, NewLogsResponse(), resp)
+}
+
+func TestLogsGrpcError(t *testing.T) {
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	RegisterLogsServer(s, &fakeLogsServer{t: t, err: errors.New("my error")})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		assert.NoError(t, s.Serve(lis))
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+		wg.Wait()
+	})
+
+	cc, err := grpc.Dial("bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock())
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, cc.Close())
+	})
+
+	logClient := NewLogsClient(cc)
+	resp, err := logClient.Export(context.Background(), generateLogsRequest())
+	require.Error(t, err)
+	st, okSt := status.FromError(err)
+	require.True(t, okSt)
+	assert.Equal(t, "my error", st.Message())
+	assert.Equal(t, codes.Unknown, st.Code())
+	assert.Equal(t, LogsResponse{}, resp)
+}
+
+type fakeLogsServer struct {
+	t   *testing.T
+	err error
+}
+
+func (f fakeLogsServer) Export(_ context.Context, request LogsRequest) (LogsResponse, error) {
+	assert.Equal(f.t, generateLogsRequest(), request)
+	return NewLogsResponse(), f.err
+}
+
+func generateLogsRequest() LogsRequest {
+	ld := pdata.NewLogs()
+	ld.ResourceLogs().AppendEmpty().InstrumentationLibraryLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStringVal("test_log_record")
+
+	lr := NewLogsRequest()
+	lr.SetLogs(ld)
+	return lr
+}
diff --git a/model/otlpgrpc/metrics_test.go b/model/otlpgrpc/metrics_test.go
@@ -15,11 +15,23 @@
 package otlpgrpc
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
+	"net"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	"go.opentelemetry.io/collector/model/pdata"
 )
 
 var _ json.Unmarshaler = MetricsResponse{}
@@ -56,3 +68,101 @@ func TestMetricsRequestJSON(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, strings.Join(strings.Fields(string(metricsRequestJSON)), ""), string(got))
 }
+
+func TestMetricsRequestJSON_Deprecated(t *testing.T) {
+	mr, err := UnmarshalJSONMetricsRequest(metricsRequestJSON)
+	assert.NoError(t, err)
+	assert.Equal(t, "test_metric", mr.Metrics().ResourceMetrics().At(0).InstrumentationLibraryMetrics().At(0).Metrics().At(0).Name())
+
+	got, err := mr.MarshalJSON()
+	assert.NoError(t, err)
+	assert.Equal(t, strings.Join(strings.Fields(string(metricsRequestJSON)), ""), string(got))
+}
+
+func TestMetricsGrpc(t *testing.T) {
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	RegisterMetricsServer(s, &fakeMetricsServer{t: t})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		assert.NoError(t, s.Serve(lis))
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+		wg.Wait()
+	})
+
+	cc, err := grpc.Dial("bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock())
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, cc.Close())
+	})
+
+	logClient := NewMetricsClient(cc)
+
+	resp, err := logClient.Export(context.Background(), generateMetricsRequest())
+	assert.NoError(t, err)
+	assert.Equal(t, NewMetricsResponse(), resp)
+}
+
+func TestMetricsGrpcError(t *testing.T) {
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	RegisterMetricsServer(s, &fakeMetricsServer{t: t, err: errors.New("my error")})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		assert.NoError(t, s.Serve(lis))
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+		wg.Wait()
+	})
+
+	cc, err := grpc.Dial("bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock())
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, cc.Close())
+	})
+
+	logClient := NewMetricsClient(cc)
+	resp, err := logClient.Export(context.Background(), generateMetricsRequest())
+	require.Error(t, err)
+	st, okSt := status.FromError(err)
+	require.True(t, okSt)
+	assert.Equal(t, "my error", st.Message())
+	assert.Equal(t, codes.Unknown, st.Code())
+	assert.Equal(t, MetricsResponse{}, resp)
+}
+
+type fakeMetricsServer struct {
+	t   *testing.T
+	err error
+}
+
+func (f fakeMetricsServer) Export(_ context.Context, request MetricsRequest) (MetricsResponse, error) {
+	assert.Equal(f.t, generateMetricsRequest(), request)
+	return NewMetricsResponse(), f.err
+}
+
+func generateMetricsRequest() MetricsRequest {
+	md := pdata.NewMetrics()
+	md.ResourceMetrics().AppendEmpty().InstrumentationLibraryMetrics().AppendEmpty().Metrics().AppendEmpty().SetName("test_metric")
+
+	mr := NewMetricsRequest()
+	mr.SetMetrics(md)
+	return mr
+}
diff --git a/model/otlpgrpc/traces_test.go b/model/otlpgrpc/traces_test.go
@@ -15,11 +15,23 @@
 package otlpgrpc
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
+	"net"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	"go.opentelemetry.io/collector/model/pdata"
 )
 
 var _ json.Unmarshaler = TracesResponse{}
@@ -52,11 +64,109 @@ var tracesRequestJSON = []byte(`
 	}`)
 
 func TestTracesRequestJSON(t *testing.T) {
-	mr := NewTracesRequest()
-	assert.NoError(t, mr.UnmarshalJSON(tracesRequestJSON))
-	assert.Equal(t, "test_span", mr.Traces().ResourceSpans().At(0).InstrumentationLibrarySpans().At(0).Spans().At(0).Name())
+	tr := NewTracesRequest()
+	assert.NoError(t, tr.UnmarshalJSON(tracesRequestJSON))
+	assert.Equal(t, "test_span", tr.Traces().ResourceSpans().At(0).InstrumentationLibrarySpans().At(0).Spans().At(0).Name())
+
+	got, err := tr.MarshalJSON()
+	assert.NoError(t, err)
+	assert.Equal(t, strings.Join(strings.Fields(string(tracesRequestJSON)), ""), string(got))
+}
+
+func TestTracesRequestJSON_Deprecated(t *testing.T) {
+	tr, err := UnmarshalJSONTracesRequest(tracesRequestJSON)
+	assert.NoError(t, err)
+	assert.Equal(t, "test_span", tr.Traces().ResourceSpans().At(0).InstrumentationLibrarySpans().At(0).Spans().At(0).Name())
 
-	got, err := mr.MarshalJSON()
+	got, err := tr.MarshalJSON()
 	assert.NoError(t, err)
 	assert.Equal(t, strings.Join(strings.Fields(string(tracesRequestJSON)), ""), string(got))
 }
+
+func TestTracesGrpc(t *testing.T) {
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	RegisterTracesServer(s, &fakeTracesServer{t: t})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		assert.NoError(t, s.Serve(lis))
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+		wg.Wait()
+	})
+
+	cc, err := grpc.Dial("bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock())
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, cc.Close())
+	})
+
+	logClient := NewTracesClient(cc)
+
+	resp, err := logClient.Export(context.Background(), generateTracesRequest())
+	assert.NoError(t, err)
+	assert.Equal(t, NewTracesResponse(), resp)
+}
+
+func TestTracesGrpcError(t *testing.T) {
+	lis := bufconn.Listen(1024 * 1024)
+	s := grpc.NewServer()
+	RegisterTracesServer(s, &fakeTracesServer{t: t, err: errors.New("my error")})
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		assert.NoError(t, s.Serve(lis))
+	}()
+	t.Cleanup(func() {
+		s.Stop()
+		wg.Wait()
+	})
+
+	cc, err := grpc.Dial("bufnet",
+		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return lis.Dial()
+		}),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithBlock())
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, cc.Close())
+	})
+
+	logClient := NewTracesClient(cc)
+	resp, err := logClient.Export(context.Background(), generateTracesRequest())
+	require.Error(t, err)
+	st, okSt := status.FromError(err)
+	require.True(t, okSt)
+	assert.Equal(t, "my error", st.Message())
+	assert.Equal(t, codes.Unknown, st.Code())
+	assert.Equal(t, TracesResponse{}, resp)
+}
+
+type fakeTracesServer struct {
+	t   *testing.T
+	err error
+}
+
+func (f fakeTracesServer) Export(_ context.Context, request TracesRequest) (TracesResponse, error) {
+	assert.Equal(f.t, generateTracesRequest(), request)
+	return NewTracesResponse(), f.err
+}
+
+func generateTracesRequest() TracesRequest {
+	td := pdata.NewTraces()
+	td.ResourceSpans().AppendEmpty().InstrumentationLibrarySpans().AppendEmpty().Spans().AppendEmpty().SetName("test_span")
+
+	tr := NewTracesRequest()
+	tr.SetTraces(td)
+	return tr
+}