Dragonfly2

1
/*
2
 *     Copyright 2023 The Dragonfly Authors
3
 *
4
 * Licensed under the Apache License, Version 2.0 (the "License");
5
 * you may not use this file except in compliance with the License.
6
 * You may obtain a copy of the License at
7
 *
8
 *      http://www.apache.org/licenses/LICENSE-2.0
9
 *
10
 * Unless required by applicable law or agreed to in writing, software
11
 * distributed under the License is distributed on an "AS IS" BASIS,
12
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
 * See the License for the specific language governing permissions and
14
 * limitations under the License.
15
 */
16

17
package service
18

19
import (
20
	"context"
21
	"errors"
22
	"fmt"
23
	"io"
24
	"os"
25
	"path/filepath"
26
	"reflect"
27
	"sync"
28
	"testing"
29

30
	"github.com/stretchr/testify/assert"
31
	"go.uber.org/mock/gomock"
32
	"google.golang.org/protobuf/types/known/emptypb"
33

34
	trainerv1 "d7y.io/api/v2/pkg/apis/trainer/v1"
35
	trainerv1mocks "d7y.io/api/v2/pkg/apis/trainer/v1/mocks"
36

37
	"d7y.io/dragonfly/v2/pkg/idgen"
38
	"d7y.io/dragonfly/v2/trainer/config"
39
	storagemocks "d7y.io/dragonfly/v2/trainer/storage/mocks"
40
	trainingmocks "d7y.io/dragonfly/v2/trainer/training/mocks"
41
)
42

43
var (
44
	mockHostName = "localhost"
45
	mockIP       = "127.0.0.1"
46
	mockHostID   = idgen.HostIDV2(mockIP, mockHostName)
47
	mockDataset  = []byte("foo")
48
)
49

50
func TestService_NewV1(t *testing.T) {
51
	tests := []struct {
52
		name string
53
		run  func(t *testing.T, s any)
54
	}{
55
		{
56
			name: "new service",
57
			run: func(t *testing.T, s any) {
58
				assert := assert.New(t)
59
				assert.Equal(reflect.TypeOf(s).Elem().Name(), "V1")
60
			},
61
		},
62
	}
63
	for _, tc := range tests {
64
		t.Run(tc.name, func(t *testing.T) {
65
			ctl := gomock.NewController(t)
66
			defer ctl.Finish()
67
			storage := storagemocks.NewMockStorage(ctl)
68
			training := trainingmocks.NewMockTraining(ctl)
69
			tc.run(t, NewV1(config.New(), storage, training))
70
		})
71
	}
72
}
73

74
func TestV1_Train(t *testing.T) {
75
	tests := []struct {
76
		name string
77
		run  func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
78
			ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder)
79
	}{
80
		{
81
			name: "receive GNN and MLP train requests success",
82
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
83
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
84
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(),
85
					fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
86
				if err != nil {
87
					t.Fatal(err)
88
				}
89

90
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(),
91
					fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
92
				if err != nil {
93
					t.Fatal(err)
94
				}
95

96
				var wg sync.WaitGroup
97
				wg.Add(1)
98
				defer wg.Wait()
99
				gomock.InOrder(
100
					mtts.Recv().Return(&trainerv1.TrainRequest{
101
						Hostname: mockHostName,
102
						Ip:       mockIP,
103
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
104
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
105
								Dataset: mockDataset,
106
							},
107
						},
108
					}, nil).Times(1),
109

110
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
111
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
112
					mtts.Recv().Return(&trainerv1.TrainRequest{
113
						Hostname: mockHostName,
114
						Ip:       mockIP,
115
						Request: &trainerv1.TrainRequest_TrainMlpRequest{
116
							TrainMlpRequest: &trainerv1.TrainMLPRequest{
117
								Dataset: mockDataset,
118
							},
119
						},
120
					}, nil).Times(1),
121
					mtts.Recv().Return(nil, io.EOF).Times(1),
122
					mtts.SendAndClose(new(emptypb.Empty)).Return(nil).Times(1),
123
					mt.Train(context.Background(), mockIP, mockHostName).DoAndReturn(func(ctx context.Context, ip, hostName string) error {
124
						wg.Done()
125
						return nil
126
					}).Times(1),
127
				)
128

129
				assert := assert.New(t)
130
				assert.NoError(svc.Train(stream))
131
			},
132
		},
133
		{
134
			name: "receive error",
135
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
136
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
137
				gomock.InOrder(
138
					mtts.Recv().Return(nil, errors.New("receive error")).Times(1),
139
				)
140

141
				assert := assert.New(t)
142
				assert.EqualError(svc.Train(stream), "receive error")
143
			},
144
		},
145
		{
146
			name: "open network topology file error",
147
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
148
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
149
				gomock.InOrder(
150
					mtts.Recv().Return(&trainerv1.TrainRequest{
151
						Hostname: mockHostName,
152
						Ip:       mockIP,
153
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
154
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
155
								Dataset: mockDataset,
156
							},
157
						},
158
					}, nil).Times(1),
159

160
					ms.OpenNetworkTopology(mockHostID).Return(nil, errors.New("open network topology file error")).Times(1),
161
				)
162

163
				assert := assert.New(t)
164
				assert.EqualError(svc.Train(stream),
165
					"rpc error: code = Internal desc = open network topology failed: open network topology file error")
166
			},
167
		},
168
		{
169
			name: "open download file error",
170
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
171
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
172
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
173
				if err != nil {
174
					t.Fatal(err)
175
				}
176

177
				gomock.InOrder(
178
					mtts.Recv().Return(&trainerv1.TrainRequest{
179
						Hostname: mockHostName,
180
						Ip:       mockIP,
181
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
182
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
183
								Dataset: mockDataset,
184
							},
185
						},
186
					}, nil).Times(1),
187

188
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
189
					ms.OpenDownload(mockHostID).Return(nil, errors.New("open download file error")).Times(1),
190
					ms.ClearNetworkTopology(mockHostID).Do(func(id string) {
191
						networktopologyFile.Close()
192
						if err := os.Remove(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", id, "csv"))); err != nil {
193
							t.Fatal(err)
194
						}
195
					}).Return(nil).Times(1),
196
				)
197

198
				assert := assert.New(t)
199
				assert.EqualError(svc.Train(stream),
200
					"rpc error: code = Internal desc = open download failed: open download file error")
201
			},
202
		},
203
		{
204
			name: "clear network topology file error",
205
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
206
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
207
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
208
				if err != nil {
209
					t.Fatal(err)
210
				}
211

212
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
213
				if err != nil {
214
					t.Fatal(err)
215
				}
216

217
				gomock.InOrder(
218
					mtts.Recv().Return(&trainerv1.TrainRequest{
219
						Hostname: mockHostName,
220
						Ip:       mockIP,
221
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
222
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
223
								Dataset: mockDataset,
224
							},
225
						},
226
					}, nil).Times(1),
227

228
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
229
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
230
					mtts.Recv().Return(nil, errors.New("receive error")).Times(1),
231
					ms.ClearDownload(mockHostID).Do(func(id string) {
232
						downloadFile.Close()
233
						if err := os.Remove(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", id, "csv"))); err != nil {
234
							t.Fatal(err)
235
						}
236
					}).Return(nil).Times(1),
237

238
					ms.ClearNetworkTopology(mockHostID).Do(func(id string) {
239
						networktopologyFile.Close()
240
						if err := os.Remove(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", id, "csv"))); err != nil {
241
							t.Fatal(err)
242
						}
243
					}).Return(errors.New("clear network topology file error")).Times(1),
244
				)
245

246
				assert := assert.New(t)
247
				assert.EqualError(svc.Train(stream), "receive error")
248
			},
249
		},
250
		{
251
			name: "clear download file error",
252
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
253
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
254
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
255
				if err != nil {
256
					t.Fatal(err)
257
				}
258

259
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
260
				if err != nil {
261
					t.Fatal(err)
262
				}
263

264
				gomock.InOrder(
265
					mtts.Recv().Return(&trainerv1.TrainRequest{
266
						Hostname: mockHostName,
267
						Ip:       mockIP,
268
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
269
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
270
								Dataset: mockDataset,
271
							},
272
						},
273
					}, nil).Times(1),
274

275
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
276
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
277
					mtts.Recv().Return(nil, errors.New("receive error")).Times(1),
278
					ms.ClearDownload(mockHostID).Do(func(id string) {
279
						downloadFile.Close()
280
						if err := os.Remove(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", id, "csv"))); err != nil {
281
							t.Fatal(err)
282
						}
283
					}).Return(errors.New("clear download file error")).Times(1),
284

285
					ms.ClearNetworkTopology(mockHostID).Do(func(id string) {
286
						networktopologyFile.Close()
287
						if err := os.Remove(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", id, "csv"))); err != nil {
288
							t.Fatal(err)
289
						}
290
					}).Return(nil).Times(1),
291
				)
292

293
				assert := assert.New(t)
294
				assert.EqualError(svc.Train(stream), "receive error")
295
			},
296
		},
297
		{
298
			name: "store network topology error",
299
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
300
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
301
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
302
				if err != nil {
303
					t.Fatal(err)
304
				}
305

306
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
307
				if err != nil {
308
					t.Fatal(err)
309
				}
310

311
				gomock.InOrder(
312
					mtts.Recv().Return(&trainerv1.TrainRequest{
313
						Hostname: mockHostName,
314
						Ip:       mockIP,
315
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
316
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
317
								Dataset: mockDataset,
318
							},
319
						},
320
					}, nil).Times(1),
321

322
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
323
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
324
				)
325

326
				networktopologyFile.Close()
327
				assert := assert.New(t)
328
				assert.EqualError(svc.Train(stream),
329
					"rpc error: code = Internal desc = write network topology failed: write /tmp/networktopology-52fa2eb710c71cc3e6ba7be6ca82453fcfe59e1c5da358ab3df8b72fd4d2a2cf.csv: file already closed")
330
			},
331
		},
332
		{
333
			name: "store download error",
334
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
335
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
336
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
337
				if err != nil {
338
					t.Fatal(err)
339
				}
340

341
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
342
				if err != nil {
343
					t.Fatal(err)
344
				}
345

346
				gomock.InOrder(
347
					mtts.Recv().Return(&trainerv1.TrainRequest{
348
						Hostname: mockHostName,
349
						Ip:       mockIP,
350
						Request: &trainerv1.TrainRequest_TrainMlpRequest{
351
							TrainMlpRequest: &trainerv1.TrainMLPRequest{
352
								Dataset: mockDataset,
353
							},
354
						},
355
					}, nil).Times(1),
356

357
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
358
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
359
				)
360

361
				downloadFile.Close()
362
				assert := assert.New(t)
363
				assert.EqualError(svc.Train(stream),
364
					"rpc error: code = Internal desc = write download failed: write /tmp/download-52fa2eb710c71cc3e6ba7be6ca82453fcfe59e1c5da358ab3df8b72fd4d2a2cf.csv: file already closed")
365
			},
366
		},
367
		{
368
			name: "receive unknown request",
369
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
370
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
371
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
372
				if err != nil {
373
					t.Fatal(err)
374
				}
375

376
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
377
				if err != nil {
378
					t.Fatal(err)
379
				}
380

381
				gomock.InOrder(
382
					mtts.Recv().Return(&trainerv1.TrainRequest{
383
						Hostname: mockHostName,
384
						Ip:       mockIP,
385
						Request:  nil,
386
					}, nil).Times(1),
387

388
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
389
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
390
				)
391

392
				assert := assert.New(t)
393
				assert.EqualError(svc.Train(stream), "rpc error: code = FailedPrecondition desc = receive unknown request: <nil>")
394
			},
395
		},
396
		{
397
			name: "send and close error",
398
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
399
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
400
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
401
				if err != nil {
402
					t.Fatal(err)
403
				}
404

405
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
406
				if err != nil {
407
					t.Fatal(err)
408
				}
409

410
				gomock.InOrder(
411
					mtts.Recv().Return(&trainerv1.TrainRequest{
412
						Hostname: mockHostName,
413
						Ip:       mockIP,
414
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
415
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
416
								Dataset: mockDataset,
417
							},
418
						},
419
					}, nil).Times(1),
420

421
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
422
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
423
					mtts.Recv().Return(&trainerv1.TrainRequest{
424
						Hostname: mockHostName,
425
						Ip:       mockIP,
426
						Request: &trainerv1.TrainRequest_TrainMlpRequest{
427
							TrainMlpRequest: &trainerv1.TrainMLPRequest{
428
								Dataset: mockDataset,
429
							},
430
						},
431
					}, nil).Times(1),
432
					mtts.Recv().Return(nil, io.EOF).Times(1),
433
					mtts.SendAndClose(new(emptypb.Empty)).Return(errors.New("send and close error")).Times(1),
434
				)
435

436
				assert := assert.New(t)
437
				assert.EqualError(svc.Train(stream), "send and close error")
438
			},
439
		},
440
		{
441
			name: "training error",
442
			run: func(t *testing.T, svc *V1, stream trainerv1.Trainer_TrainServer, mtts *trainerv1mocks.MockTrainer_TrainServerMockRecorder,
443
				ms *storagemocks.MockStorageMockRecorder, mt *trainingmocks.MockTrainingMockRecorder) {
444
				networktopologyFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "networktopology", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
445
				if err != nil {
446
					t.Fatal(err)
447
				}
448

449
				downloadFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s-%s.%s", "download", mockHostID, "csv")), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
450
				if err != nil {
451
					t.Fatal(err)
452
				}
453

454
				var wg sync.WaitGroup
455
				wg.Add(1)
456
				defer wg.Wait()
457
				gomock.InOrder(
458
					mtts.Recv().Return(&trainerv1.TrainRequest{
459
						Hostname: mockHostName,
460
						Ip:       mockIP,
461
						Request: &trainerv1.TrainRequest_TrainGnnRequest{
462
							TrainGnnRequest: &trainerv1.TrainGNNRequest{
463
								Dataset: mockDataset,
464
							},
465
						},
466
					}, nil).Times(1),
467

468
					ms.OpenNetworkTopology(mockHostID).Return(networktopologyFile, nil).Times(1),
469
					ms.OpenDownload(mockHostID).Return(downloadFile, nil).Times(1),
470
					mtts.Recv().Return(&trainerv1.TrainRequest{
471
						Hostname: mockHostName,
472
						Ip:       mockIP,
473
						Request: &trainerv1.TrainRequest_TrainMlpRequest{
474
							TrainMlpRequest: &trainerv1.TrainMLPRequest{
475
								Dataset: mockDataset,
476
							},
477
						},
478
					}, nil).Times(1),
479
					mtts.Recv().Return(nil, io.EOF).Times(1),
480
					mtts.SendAndClose(new(emptypb.Empty)).Return(nil).Times(1),
481
					mt.Train(context.Background(), mockIP, mockHostName).DoAndReturn(func(ctx context.Context, ip, hostName string) error {
482
						wg.Done()
483
						return errors.New("training error")
484
					}).Times(1),
485
				)
486

487
				assert := assert.New(t)
488
				assert.NoError(svc.Train(stream))
489
			},
490
		},
491
	}
492

493
	for _, tc := range tests {
494
		t.Run(tc.name, func(t *testing.T) {
495
			ctl := gomock.NewController(t)
496
			defer ctl.Finish()
497
			storage := storagemocks.NewMockStorage(ctl)
498
			training := trainingmocks.NewMockTraining(ctl)
499
			stream := trainerv1mocks.NewMockTrainer_TrainServer(ctl)
500

501
			svc := NewV1(config.New(), storage, training)
502
			tc.run(t, svc, stream, stream.EXPECT(), storage.EXPECT(), training.EXPECT())
503
		})
504
	}
505
}
506