forked from pingcap/tidb
-
Notifications
You must be signed in to change notification settings - Fork 0
/
mock.go
173 lines (143 loc) · 5.14 KB
/
mock.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://wwm.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.
package ddl
import (
"context"
"sync/atomic"
"time"
"github.com/pingcap/errors"
"github.com/pingcap/failpoint"
"github.com/pingcap/parser/ast"
"github.com/pingcap/parser/charset"
"github.com/pingcap/parser/model"
"github.com/pingcap/tidb/ddl/util"
"github.com/pingcap/tidb/sessionctx"
"go.etcd.io/etcd/clientv3"
)
var _ util.SchemaSyncer = &MockSchemaSyncer{}
const mockCheckVersInterval = 2 * time.Millisecond
// MockSchemaSyncer is a mock schema syncer, it is exported for tesing.
type MockSchemaSyncer struct {
selfSchemaVersion int64
globalVerCh chan clientv3.WatchResponse
mockSession chan struct{}
}
// NewMockSchemaSyncer creates a new mock SchemaSyncer.
func NewMockSchemaSyncer() util.SchemaSyncer {
return &MockSchemaSyncer{}
}
// Init implements SchemaSyncer.Init interface.
func (s *MockSchemaSyncer) Init(ctx context.Context) error {
s.globalVerCh = make(chan clientv3.WatchResponse, 1)
s.mockSession = make(chan struct{}, 1)
return nil
}
// GlobalVersionCh implements SchemaSyncer.GlobalVersionCh interface.
func (s *MockSchemaSyncer) GlobalVersionCh() clientv3.WatchChan {
return s.globalVerCh
}
// WatchGlobalSchemaVer implements SchemaSyncer.WatchGlobalSchemaVer interface.
func (s *MockSchemaSyncer) WatchGlobalSchemaVer(context.Context) {}
// UpdateSelfVersion implements SchemaSyncer.UpdateSelfVersion interface.
func (s *MockSchemaSyncer) UpdateSelfVersion(ctx context.Context, version int64) error {
atomic.StoreInt64(&s.selfSchemaVersion, version)
return nil
}
// Done implements SchemaSyncer.Done interface.
func (s *MockSchemaSyncer) Done() <-chan struct{} {
return s.mockSession
}
// CloseSession mockSession, it is exported for testing.
func (s *MockSchemaSyncer) CloseSession() {
close(s.mockSession)
}
// Restart implements SchemaSyncer.Restart interface.
func (s *MockSchemaSyncer) Restart(_ context.Context) error {
s.mockSession = make(chan struct{}, 1)
return nil
}
// OwnerUpdateGlobalVersion implements SchemaSyncer.OwnerUpdateGlobalVersion interface.
func (s *MockSchemaSyncer) OwnerUpdateGlobalVersion(ctx context.Context, version int64) error {
select {
case s.globalVerCh <- clientv3.WatchResponse{}:
default:
}
return nil
}
// MustGetGlobalVersion implements SchemaSyncer.MustGetGlobalVersion interface.
func (s *MockSchemaSyncer) MustGetGlobalVersion(ctx context.Context) (int64, error) {
return 0, nil
}
// OwnerCheckAllVersions implements SchemaSyncer.OwnerCheckAllVersions interface.
func (s *MockSchemaSyncer) OwnerCheckAllVersions(ctx context.Context, latestVer int64) error {
ticker := time.NewTicker(mockCheckVersInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
failpoint.Inject("checkOwnerCheckAllVersionsWaitTime", func(v failpoint.Value) {
if v.(bool) {
panic("shouldn't happen")
}
})
return errors.Trace(ctx.Err())
case <-ticker.C:
ver := atomic.LoadInt64(&s.selfSchemaVersion)
if ver >= latestVer {
return nil
}
}
}
}
// NotifyCleanExpiredPaths implements SchemaSyncer.NotifyCleanExpiredPaths interface.
func (s *MockSchemaSyncer) NotifyCleanExpiredPaths() bool { return true }
// StartCleanWork implements SchemaSyncer.StartCleanWork interface.
func (s *MockSchemaSyncer) StartCleanWork() {}
// Close implements SchemaSyncer.Close interface.
func (s *MockSchemaSyncer) Close() {}
type mockDelRange struct {
}
// newMockDelRangeManager creates a mock delRangeManager only used for test.
func newMockDelRangeManager() delRangeManager {
return &mockDelRange{}
}
// addDelRangeJob implements delRangeManager interface.
func (dr *mockDelRange) addDelRangeJob(job *model.Job) error {
return nil
}
// removeFromGCDeleteRange implements delRangeManager interface.
func (dr *mockDelRange) removeFromGCDeleteRange(jobID int64, tableIDs []int64) error {
return nil
}
// start implements delRangeManager interface.
func (dr *mockDelRange) start() {}
// clear implements delRangeManager interface.
func (dr *mockDelRange) clear() {}
// MockTableInfo mocks a table info by create table stmt ast and a specified table id.
func MockTableInfo(ctx sessionctx.Context, stmt *ast.CreateTableStmt, tableID int64) (*model.TableInfo, error) {
chs, coll := charset.GetDefaultCharsetAndCollate()
cols, newConstraints, err := buildColumnsAndConstraints(ctx, stmt.Cols, stmt.Constraints, chs, coll)
if err != nil {
return nil, errors.Trace(err)
}
tbl, err := buildTableInfo(ctx, stmt.Table.Name, cols, newConstraints, "", "")
if err != nil {
return nil, errors.Trace(err)
}
tbl.ID = tableID
// The specified charset will be handled in handleTableOptions
if err = handleTableOptions(stmt.Options, tbl); err != nil {
return nil, errors.Trace(err)
}
return tbl, nil
}