Function TestNewRPCServer

pkg/rpcserver/rpcserver_test.go:64–142

(t *testing.T)

)

func TestNewRPCServer(t *testing.T) {
	a := assertions.New(t)
	ctx, cancel := context.WithCancel(test.Context())
	defer cancel()

	logHandler := memory.New()
	logger := log.NewLogger(logHandler)
	ctx = log.NewContext(ctx, logger)

	server := rpcserver.New(ctx,
		rpcserver.WithContextFiller(
			func(ctx context.Context) context.Context {
				return context.WithValue(ctx, &mockKey{}, "foo")
			},
		),
		rpcserver.WithUnaryInterceptors(UnaryServerInterceptor),
		rpcserver.WithStreamInterceptors(StreamServerInterceptor),
	)
	a.So(server, should.NotBeNil)
	mock := &mockServer{}
	ttnpb.RegisterAppAsServer(server.Server, mock)

	loopbackConn, err := rpcserver.StartLoopback(ctx, server.Server, rpcclient.DefaultDialOptions(ctx)...)
	a.So(loopbackConn, should.NotBeNil)
	a.So(err, should.BeNil)

	cli := ttnpb.NewAppAsClient(loopbackConn)

	t.Run("Unary", func(t *testing.T) {
		a := assertions.New(t)

		_, err = cli.DownlinkQueuePush(ctx, downlinkQueueReq)
		a.So(err, should.BeNil)

		a.So(mock.pushReq, should.NotBeNil)
		a.So(mock.pushReq, should.Resemble, downlinkQueueReq)

		a.So(mock.pushCtx, should.NotBeNil)
		a.So(mock.pushCtx.Value(&mockKey{}), should.Resemble, "foo")
		a.So(grpc_ctxtags.Extract(mock.pushCtx).Values(), should.Resemble, map[string]any{
			"peer.address":                "pipe",
			"grpc.request.device_id":      "foo",
			"grpc.request.application_id": "bar",
		})
		a.So(mock.pushCtx.Value(&mockKey2{}), should.Resemble, "bar")

		runtime.Gosched()
		time.Sleep(test.Delay)

		a.So(logHandler.Entries, should.HaveLength, 1)
	})

	t.Run("Stream", func(t *testing.T) {
		a := assertions.New(t)

		sub, err := cli.Subscribe(ctx, &ttnpb.ApplicationIdentifiers{
			ApplicationId: "bar",
		})