214 lines
6.7 KiB
Go
214 lines
6.7 KiB
Go
package identity
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
customer "git.sendico.io/tech/sxchnge/product/customers/model"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
"github.com/stretchr/testify/require"
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
"go.uber.org/zap"
|
|
"go.uber.org/zap/zaptest/observer"
|
|
)
|
|
|
|
type memoryStore struct {
|
|
mu sync.Mutex
|
|
rows map[bson.ObjectID]Binding
|
|
bindErr, acceptErr error
|
|
}
|
|
|
|
func (s *memoryStore) Get(_ context.Context, id bson.ObjectID) (*Binding, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
b, ok := s.rows[id]
|
|
if !ok {
|
|
return nil, merrors.NoData("missing")
|
|
}
|
|
return &b, nil
|
|
}
|
|
func (s *memoryStore) Create(_ context.Context, b *Binding) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if _, ok := s.rows[b.ID]; ok {
|
|
return merrors.DataConflict("duplicate")
|
|
}
|
|
s.rows[b.ID] = *b
|
|
return nil
|
|
}
|
|
func (s *memoryStore) Claim(_ context.Context, id bson.ObjectID, namespace string) (bool, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
b := s.rows[id]
|
|
if b.Namespace != namespace || b.Registering || b.SubjectRef != "" {
|
|
return false, nil
|
|
}
|
|
b.Registering = true
|
|
b.RegisteringAt = time.Now()
|
|
s.rows[id] = b
|
|
return true, nil
|
|
}
|
|
func (s *memoryStore) Bind(_ context.Context, id bson.ObjectID, namespace, subject string) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.bindErr != nil {
|
|
return s.bindErr
|
|
}
|
|
b := s.rows[id]
|
|
b.SubjectRef = subject
|
|
b.Registering = false
|
|
s.rows[id] = b
|
|
return nil
|
|
}
|
|
func (s *memoryStore) Accept(_ context.Context, id bson.ObjectID, namespace string, r Receipt) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.acceptErr != nil {
|
|
return s.acceptErr
|
|
}
|
|
b := s.rows[id]
|
|
b.Receipt = &r
|
|
s.rows[id] = b
|
|
return nil
|
|
}
|
|
|
|
type providerStub struct {
|
|
registrations, submissions int
|
|
registerErr, submitErr error
|
|
snapshots []Facts
|
|
sources []string
|
|
entered, release chan struct{}
|
|
}
|
|
|
|
func (*providerStub) Namespace() string { return "provider-1" }
|
|
func (p *providerStub) Register(context.Context, string) (string, error) {
|
|
p.registrations++
|
|
if p.entered != nil {
|
|
close(p.entered)
|
|
<-p.release
|
|
}
|
|
return "subject-1", p.registerErr
|
|
}
|
|
func (p *providerStub) Submit(_ context.Context, subject, source string, f Facts) (Receipt, error) {
|
|
p.submissions++
|
|
p.snapshots = append(p.snapshots, f)
|
|
p.sources = append(p.sources, source)
|
|
return Receipt{SubjectRef: subject, SubmissionRef: "submission-1", AcceptedAt: time.Now()}, p.submitErr
|
|
}
|
|
func fixture() (*Service, *memoryStore, *providerStub, *customer.Account) {
|
|
store := &memoryStore{rows: map[bson.ObjectID]Binding{}}
|
|
provider := &providerStub{}
|
|
a := &customer.Account{}
|
|
a.ID = bson.NewObjectID()
|
|
a.Name = "Ada Lovelace"
|
|
a.Login = "ada@example.com"
|
|
a.Password = "never transmit"
|
|
return &Service{Store: store, Provider: provider, Logger: zap.NewNop()}, store, provider, a
|
|
}
|
|
|
|
func TestIntakeRetryUsesPersistedSubjectSnapshotAndStableSource(t *testing.T) {
|
|
s, store, p, a := fixture()
|
|
p.submitErr = merrors.ServiceUnavailable("upstream")
|
|
require.Error(t, s.Ensure(context.Background(), a))
|
|
require.Error(t, s.Ready(context.Background(), a.ID))
|
|
require.Equal(t, 1, p.registrations)
|
|
b, err := store.Get(context.Background(), a.ID)
|
|
require.NoError(t, err)
|
|
require.Equal(t, "subject-1", b.SubjectRef)
|
|
require.Nil(t, b.Receipt)
|
|
a.Name = "changed retry"
|
|
p.submitErr = nil
|
|
require.NoError(t, s.Ensure(context.Background(), a))
|
|
require.NoError(t, s.Ready(context.Background(), a.ID))
|
|
require.Equal(t, 1, p.registrations)
|
|
require.Equal(t, 2, p.submissions)
|
|
require.Equal(t, p.snapshots[0], p.snapshots[1])
|
|
require.Equal(t, "Ada Lovelace", p.snapshots[1].FullName)
|
|
require.Equal(t, p.sources[0], p.sources[1])
|
|
require.NoError(t, s.Ensure(context.Background(), a))
|
|
require.Equal(t, 2, p.submissions)
|
|
other := a.Copy()
|
|
other.ID = bson.NewObjectID()
|
|
require.Error(t, s.Ready(context.Background(), other.ID))
|
|
}
|
|
func TestLostAcceptanceReceiptIsSafelyReplayed(t *testing.T) {
|
|
s, store, p, a := fixture()
|
|
store.acceptErr = merrors.ServiceUnavailable("storage")
|
|
require.Error(t, s.Ensure(context.Background(), a))
|
|
store.acceptErr = nil
|
|
require.NoError(t, s.Ensure(context.Background(), a))
|
|
require.Equal(t, 1, p.registrations)
|
|
require.Equal(t, 2, p.submissions)
|
|
require.Equal(t, p.sources[0], p.sources[1])
|
|
}
|
|
func TestAmbiguousRegistrationNeverAutomaticallyRegistersAgain(t *testing.T) {
|
|
for _, bindFailure := range []bool{false, true} {
|
|
t.Run(map[bool]string{false: "request", true: "binding"}[bindFailure], func(t *testing.T) {
|
|
s, store, p, a := fixture()
|
|
if bindFailure {
|
|
store.bindErr = merrors.ServiceUnavailable("storage")
|
|
} else {
|
|
p.registerErr = merrors.ServiceUnavailable("lost response")
|
|
}
|
|
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrFailedPrecondition)
|
|
b := store.rows[a.ID]
|
|
b.RegisteringAt = time.Now().Add(-time.Minute)
|
|
store.rows[a.ID] = b
|
|
store.bindErr = nil
|
|
p.registerErr = nil
|
|
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrFailedPrecondition)
|
|
require.Equal(t, 1, p.registrations)
|
|
require.Zero(t, p.submissions)
|
|
})
|
|
}
|
|
}
|
|
func TestConcurrentRequestCannotRegisterSameCustomerTwice(t *testing.T) {
|
|
s, _, p, a := fixture()
|
|
p.entered = make(chan struct{})
|
|
p.release = make(chan struct{})
|
|
done := make(chan error, 1)
|
|
go func() { done <- s.Ensure(context.Background(), a) }()
|
|
<-p.entered
|
|
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrServiceUnavailable)
|
|
close(p.release)
|
|
require.NoError(t, <-done)
|
|
require.Equal(t, 1, p.registrations)
|
|
require.NoError(t, s.Ensure(context.Background(), a))
|
|
require.Equal(t, 1, p.submissions)
|
|
}
|
|
func TestNamespaceChangeFailsClosed(t *testing.T) {
|
|
s, store, p, a := fixture()
|
|
require.NoError(t, s.Ensure(context.Background(), a))
|
|
b := store.rows[a.ID]
|
|
b.Namespace = "other-account-or-organization"
|
|
store.rows[a.ID] = b
|
|
require.ErrorIs(t, s.Ensure(context.Background(), a), merrors.ErrFailedPrecondition)
|
|
require.Error(t, s.Ready(context.Background(), a.ID))
|
|
require.Equal(t, 1, p.submissions)
|
|
}
|
|
|
|
func TestHandoffLogsExcludeAssertionsAndCredentials(t *testing.T) {
|
|
s, _, p, a := fixture()
|
|
core, logs := observer.New(zap.InfoLevel)
|
|
s.Logger = zap.New(core)
|
|
p.submitErr = merrors.ServiceUnavailable(a.Login + " " + a.Password)
|
|
require.Error(t, s.Ensure(context.Background(), a))
|
|
p.submitErr = nil
|
|
require.NoError(t, s.Ensure(context.Background(), a))
|
|
require.Len(t, logs.All(), 2)
|
|
for _, entry := range logs.All() {
|
|
require.NotContains(t, entry.Message, a.Login)
|
|
require.NotContains(t, entry.Message, a.Name)
|
|
fields := entry.ContextMap()
|
|
require.Equal(t, a.ID.Hex(), fields["account_ref"])
|
|
require.NotEmpty(t, fields["request_ref"])
|
|
for _, value := range fields {
|
|
require.NotEqual(t, a.Login, value)
|
|
require.NotEqual(t, a.Name, value)
|
|
require.NotEqual(t, a.Password, value)
|
|
}
|
|
}
|
|
}
|