go-transaction-manager

1
//go:build go1.14
2
// +build go1.14
3

4
package mongo
5

6
import (
7
	"context"
8
	"errors"
9
	"sync"
10
	"testing"
11
	"time"
12

13
	"github.com/stretchr/testify/assert"
14
	"github.com/stretchr/testify/require"
15
	"go.mongodb.org/mongo-driver/bson/primitive"
16
	"go.mongodb.org/mongo-driver/mongo"
17
	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
18
	"go.mongodb.org/mongo-driver/mongo/options"
19
	"go.mongodb.org/mongo-driver/mongo/writeconcern"
20

21
	trmcontext "github.com/avito-tech/go-transaction-manager/trm/v2/context"
22
	"github.com/avito-tech/go-transaction-manager/trm/v2/drivers/mock"
23

24
	"github.com/avito-tech/go-transaction-manager/trm/v2"
25
	"github.com/avito-tech/go-transaction-manager/trm/v2/manager"
26
	"github.com/avito-tech/go-transaction-manager/trm/v2/settings"
27
)
28

29
type user struct {
30
	ID primitive.ObjectID `bson:"_id,omitempty"`
31
}
32

33
func TestTransaction(t *testing.T) {
34
	t.Parallel()
35

36
	type args struct {
37
		ctx context.Context
38
	}
39

40
	type fields struct {
41
		settings trm.Settings
42
	}
43

44
	testErr := errors.New("error test")
45
	doNil := func(_ *mtest.T, _ context.Context) error {
46
		return nil
47
	}
48
	defaultFields := func(_ *mtest.T) fields {
49
		return fields{
50
			settings: MustSettings(settings.Must(
51
				settings.WithPropagation(trm.PropagationRequiresNew),
52
			), WithSessionOpts(&options.SessionOptions{})),
53
		}
54
	}
55

56
	mt := mtest.New(
57
		t,
58
		mtest.NewOptions().ClientType(mtest.Mock),
59
	)
60

61
	tests := map[string]struct {
62
		fields  func(mt *mtest.T) fields
63
		args    args
64
		do      func(mt *mtest.T, ctx context.Context) error
65
		wantErr assert.ErrorAssertionFunc
66
	}{
67
		"success": {
68
			fields: defaultFields,
69
			args: args{
70
				ctx: context.Background(),
71
			},
72
			do:      doNil,
73
			wantErr: assert.NoError,
74
		},
75
		"begin_session_error": {
76
			fields: func(_ *mtest.T) fields {
77
				return fields{
78
					settings: MustSettings(settings.Must(
79
						settings.WithPropagation(trm.PropagationNested),
80
					), WithSessionOpts((&options.SessionOptions{}).
81
						SetSnapshot(true).
82
						SetCausalConsistency(true))),
83
				}
84
			},
85
			args: args{
86
				ctx: context.Background(),
87
			},
88
			do: func(mt *mtest.T, _ context.Context) error {
89
				require.NotNil(mt, 1, "should not be here")
90

91
				return nil
92
			},
93
			wantErr: func(t assert.TestingT, err error, _ ...interface{}) bool {
94
				return assert.ErrorIs(t, err, trm.ErrBegin)
95
			},
96
		},
97
		"begin_transaction_error": {
98
			fields: func(_ *mtest.T) fields {
99
				return fields{
100
					settings: MustSettings(settings.Must(
101
						settings.WithPropagation(trm.PropagationNested),
102
					), WithTransactionOpts((&options.TransactionOptions{}).
103
						SetWriteConcern(&writeconcern.WriteConcern{W: 0}))),
104
				}
105
			},
106
			args: args{
107
				ctx: context.Background(),
108
			},
109
			do: func(mt *mtest.T, _ context.Context) error {
110
				require.NotNil(mt, 1, "should not be here")
111

112
				return nil
113
			},
114
			wantErr: func(t assert.TestingT, err error, _ ...interface{}) bool {
115
				return assert.ErrorIs(t, err, trm.ErrBegin)
116
			},
117
		},
118
		"commit_error": {
119
			fields: defaultFields,
120
			args: args{
121
				ctx: context.Background(),
122
			},
123
			do: func(mt *mtest.T, ctx context.Context) error {
124
				_, _ = mt.Coll.InsertOne(ctx, user{
125
					ID: primitive.NewObjectID(),
126
				})
127

128
				return nil
129
			},
130
			wantErr: func(t assert.TestingT, err error, _ ...interface{}) bool {
131
				var divErr mongo.CommandError
132

133
				return assert.ErrorAs(t, err, &divErr) &&
134
					assert.ErrorIs(t, err, trm.ErrCommit)
135
			},
136
		},
137
		"rollback_after_error": {
138
			fields: defaultFields,
139
			args: args{
140
				ctx: context.Background(),
141
			},
142
			do: func(mt *mtest.T, ctx context.Context) error {
143
				s := mongo.SessionFromContext(ctx)
144

145
				require.NoError(mt, s.AbortTransaction(ctx))
146

147
				return testErr
148
			},
149
			wantErr: func(t assert.TestingT, err error, _ ...interface{}) bool {
150
				return assert.ErrorIs(t, err, testErr) &&
151
					assert.ErrorIs(t, err, trm.ErrRollback)
152
			},
153
		},
154
	}
155
	for name, tt := range tests {
156
		tt := tt
157
		mt.Run(name, func(mt *mtest.T) {
158
			mt.Parallel()
159

160
			log := mock.NewLog()
161

162
			f := tt.fields(mt)
163

164
			m := manager.Must(
165
				NewDefaultFactory(mt.Client),
166
				manager.WithLog(log),
167
				manager.WithSettings(f.settings),
168
			)
169

170
			var tr trm.Transaction
171
			err := m.Do(tt.args.ctx, func(ctx context.Context) error {
172
				tr = trmcontext.DefaultManager.Default(ctx)
173

174
				var trNested trm.Transaction
175
				err := m.Do(ctx, func(ctx context.Context) error {
176
					trNested = trmcontext.DefaultManager.Default(ctx)
177

178
					require.NotNil(t, trNested)
179

180
					return tt.do(mt, ctx)
181
				})
182

183
				if trNested != nil {
184
					require.False(t, trNested.IsActive())
185
				}
186

187
				return err
188
			})
189

190
			if tr != nil {
191
				require.False(t, tr.IsActive())
192
			}
193

194
			if !tt.wantErr(t, err) {
195
				return
196
			}
197
		})
198
	}
199
}
200

201
func TestTransaction_awaitDone_byContext(t *testing.T) {
202
	t.Parallel()
203

204
	mt := mtest.New(
205
		t,
206
		mtest.NewOptions().
207
			ClientType(mtest.Mock).
208
			ShareClient(true),
209
	)
210

211
	wg := sync.WaitGroup{}
212
	wg.Add(1)
213

214
	f := NewDefaultFactory(mt.Client)
215
	ctx, cancel := context.WithCancel(context.Background())
216

217
	go func() {
218
		defer wg.Done()
219

220
		_, tr, err := f(ctx, settings.Must())
221

222
		cancel()
223
		<-time.After(time.Second)
224

225
		<-ctx.Done()
226

227
		require.NoError(mt, err)
228
		require.False(mt, tr.IsActive())
229
	}()
230

231
	wg.Wait()
232
}
233