曾志威
5e5f2aa7fe
Add the React console, strengthen synchronization and retry handling, and export WES MySQL and MongoDB configuration packages.
103 lines
3.7 KiB
Go
103 lines
3.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type blockedDataInspection struct {
|
|
tableSkipAdapter
|
|
inspections atomic.Int32
|
|
blockedReads atomic.Int32
|
|
entered chan struct{}
|
|
}
|
|
|
|
func (a *blockedDataInspection) Inspect(ctx context.Context, object string) (ObjectInfo, error) {
|
|
if object == "blocked" && a.inspections.Add(1) == 2 {
|
|
close(a.entered)
|
|
<-ctx.Done()
|
|
return ObjectInfo{}, ctx.Err()
|
|
}
|
|
return a.databaseSchemaAdapter.Inspect(ctx, object)
|
|
}
|
|
|
|
func (a *blockedDataInspection) Read(ctx context.Context, object string, options ReadOptions, consume func([]Record) error) error {
|
|
if object == "blocked" {
|
|
a.blockedReads.Add(1)
|
|
}
|
|
return a.tableSkipAdapter.Read(ctx, object, options, consume)
|
|
}
|
|
|
|
func TestDataInspectionHonorsTableSkipTimeoutAndTaskCancellation(t *testing.T) {
|
|
for _, action := range []string{"skip", "timeout", "cancel"} {
|
|
t.Run(action, func(t *testing.T) {
|
|
s := databaseOperationServer(t)
|
|
reader := &blockedDataInspection{entered: make(chan struct{})}
|
|
var connections atomic.Int32
|
|
s.adapterFactory = func(context.Context, DataSource, func(SQLAudit)) (Adapter, error) {
|
|
if connections.Add(1) == 1 {
|
|
return reader, nil
|
|
}
|
|
return &tableSkipAdapter{}, nil
|
|
}
|
|
task := Task{ID: "task", Objects: []string{"blocked", "good"}, SchemaWorkers: 1, MaxRows: 100, BatchSize: 10, TimeoutSeconds: 1}
|
|
s.store.Tasks[task.ID] = task
|
|
s.store.Executions["exec"] = Execution{ID: "exec", TaskID: task.ID, Status: "RUNNING"}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
s.runTask(ctx, task, s.store.Sources["source"], s.store.Sources["target"], "exec")
|
|
}()
|
|
select {
|
|
case <-reader.entered:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("data inspection did not start")
|
|
}
|
|
if action == "skip" {
|
|
if s.skipTable("exec", "good") || !s.skipTable("exec", "blocked") || s.skipTable("exec", "blocked") {
|
|
t.Fatal("active inspection was not skippable or duplicate skip accepted")
|
|
}
|
|
} else if action == "cancel" {
|
|
cancel()
|
|
}
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("data inspection did not terminate")
|
|
}
|
|
execution := s.store.Executions["exec"]
|
|
if reader.blockedReads.Load() != 0 {
|
|
t.Fatal("failed inspection reached a data reader")
|
|
}
|
|
if execution.SchemaStatuses["blocked"] != "COMPLETED" || execution.SchemaStatuses["good"] != "COMPLETED" {
|
|
t.Fatalf("data cancellation changed schema result: %+v", execution)
|
|
}
|
|
s.jobsMu.Lock()
|
|
active := len(s.tableRuns)
|
|
s.jobsMu.Unlock()
|
|
if active != 0 || s.skipTable("exec", "blocked") {
|
|
t.Fatal("finished inspection remained active")
|
|
}
|
|
if action == "cancel" {
|
|
if execution.Status != "CANCELLED" || execution.ObjectStatuses["blocked"] != "FAILED" || execution.ObjectErrors["blocked"] != "operationCancelled" || execution.ObjectStatuses["good"] != "PENDING" || execution.Written != 0 {
|
|
t.Fatalf("cancelled inspection did not preserve waiting table: %+v", execution)
|
|
}
|
|
} else {
|
|
if execution.Status != "PARTIAL" || execution.ObjectStatuses["good"] != "COMPLETED" || execution.Written != 1 || execution.ObjectFinishedAt["blocked"].IsZero() {
|
|
t.Fatalf("inspection did not continue to next table: %+v", execution)
|
|
}
|
|
if action == "skip" && execution.ObjectStatuses["blocked"] != "SKIPPED" {
|
|
t.Fatalf("inspection skip result missing: %+v", execution)
|
|
}
|
|
if action == "timeout" && (execution.ObjectStatuses["blocked"] != "FAILED" || execution.ObjectErrors["blocked"] != "operationTimeout") {
|
|
t.Fatalf("inspection timeout result missing: %+v", execution)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|