104 lines
4.1 KiB
Go
104 lines
4.1 KiB
Go
package notifications
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"encoding/pem"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
"github.com/stretchr/testify/require"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/metadata"
|
|
)
|
|
|
|
type rpcStub struct {
|
|
pb.NotificationsServiceClient
|
|
request *pb.SendRequest
|
|
auth []string
|
|
result *pb.SendResponse
|
|
err error
|
|
}
|
|
|
|
func (s *rpcStub) Send(ctx context.Context, r *pb.SendRequest, _ ...grpc.CallOption) (*pb.SendResponse, error) {
|
|
s.request = r
|
|
md, _ := metadata.FromOutgoingContext(ctx)
|
|
s.auth = md.Get("authorization")
|
|
return s.result, s.err
|
|
}
|
|
func TestSendMapsOwnedContractAndRejectsFailedReceipt(t *testing.T) {
|
|
rpc := &rpcStub{result: &pb.SendResponse{Notification: &pb.Notification{Id: "job", Deliveries: []*pb.Delivery{{State: pb.DeliveryState_DELIVERY_STATE_QUEUED}}}}}
|
|
client := &Client{rpc: rpc}
|
|
expiry := time.Now().Add(time.Minute)
|
|
m := Message{Key: "key", UserID: "customer", Email: "user@example.com", Locale: "ru", Template: "auth.login", Parameters: map[string]string{"code": "123456"}, ExpiresAt: expiry}
|
|
require.NoError(t, client.Send(context.Background(), m))
|
|
require.Empty(t, rpc.auth)
|
|
require.Equal(t, m.Email, rpc.request.Targets[0].GetEmail().Address)
|
|
require.Equal(t, "1", rpc.request.TemplateVersion)
|
|
require.Equal(t, m.Parameters, rpc.request.Parameters)
|
|
require.WithinDuration(t, expiry, rpc.request.ExpiresAt.AsTime(), time.Nanosecond)
|
|
rpc.result.Notification.Deliveries[0].State = pb.DeliveryState_DELIVERY_STATE_UNKNOWN
|
|
require.ErrorIs(t, client.Send(context.Background(), m), merrors.ErrServiceUnavailable)
|
|
rpc.result = nil
|
|
require.ErrorIs(t, client.Send(context.Background(), m), merrors.ErrServiceUnavailable)
|
|
}
|
|
func TestOpenRequiresAddressAndValidCA(t *testing.T) {
|
|
_, err := Open(Config{})
|
|
require.ErrorIs(t, err, merrors.ErrInvalidArg)
|
|
_, err = Open(Config{Address: "localhost:8443", CAFile: "missing.pem"})
|
|
require.Error(t, err)
|
|
}
|
|
|
|
type tlsRPC struct {
|
|
pb.UnimplementedNotificationsServiceServer
|
|
auth chan []string
|
|
}
|
|
|
|
func (s *tlsRPC) Send(ctx context.Context, _ *pb.SendRequest) (*pb.SendResponse, error) {
|
|
md, _ := metadata.FromIncomingContext(ctx)
|
|
s.auth <- md.Get("authorization")
|
|
return &pb.SendResponse{Notification: &pb.Notification{Id: "tls-job", Deliveries: []*pb.Delivery{{State: pb.DeliveryState_DELIVERY_STATE_QUEUED}}}}, nil
|
|
}
|
|
func TestOpenUsesVerifiedTLSWithoutCallerToken(t *testing.T) {
|
|
fixture := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
|
|
cert := fixture.TLS.Certificates[0]
|
|
fixture.Close()
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
require.NoError(t, err)
|
|
server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12})))
|
|
rpc := &tlsRPC{auth: make(chan []string, 1)}
|
|
pb.RegisterNotificationsServiceServer(server, rpc)
|
|
go func() { _ = server.Serve(listener) }()
|
|
t.Cleanup(server.Stop)
|
|
ca := filepath.Join(t.TempDir(), "ca.pem")
|
|
require.NoError(t, os.WriteFile(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Certificate[0]}), 0600))
|
|
client, err := Open(Config{Address: listener.Addr().String(), CAFile: ca})
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { _ = client.Close() })
|
|
require.NoError(t, client.Send(context.Background(), Message{ExpiresAt: time.Now().Add(time.Minute)}))
|
|
require.Empty(t, <-rpc.auth)
|
|
}
|
|
|
|
func TestOpenUsesPlaintextWithoutCallerToken(t *testing.T) {
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
require.NoError(t, err)
|
|
server := grpc.NewServer()
|
|
rpc := &tlsRPC{auth: make(chan []string, 1)}
|
|
pb.RegisterNotificationsServiceServer(server, rpc)
|
|
go func() { _ = server.Serve(listener) }()
|
|
t.Cleanup(server.Stop)
|
|
client, err := Open(Config{Address: listener.Addr().String()})
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { _ = client.Close() })
|
|
require.NoError(t, client.Send(context.Background(), Message{ExpiresAt: time.Now().Add(time.Minute)}))
|
|
require.Empty(t, <-rpc.auth)
|
|
}
|