288 lines
9.2 KiB
Go
288 lines
9.2 KiB
Go
package options_test
|
|
|
|
import (
|
|
"fmt"
|
|
"math"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/go-cmp/cmp"
|
|
"github.com/google/go-cmp/cmp/cmpopts"
|
|
"github.com/influxdata/flux/ast"
|
|
_ "github.com/influxdata/influxdb/v2/fluxinit/static"
|
|
"github.com/influxdata/influxdb/v2/pkg/pointer"
|
|
"github.com/influxdata/influxdb/v2/query/fluxlang"
|
|
"github.com/influxdata/influxdb/v2/task/options"
|
|
)
|
|
|
|
func scriptGenerator(opt options.Options, body string) string {
|
|
taskData := ""
|
|
if opt.Name != "" {
|
|
taskData = fmt.Sprintf("%s name: %q,\n", taskData, opt.Name)
|
|
}
|
|
if opt.Cron != "" {
|
|
taskData = fmt.Sprintf("%s cron: %q,\n", taskData, opt.Cron)
|
|
}
|
|
if !opt.Every.IsZero() {
|
|
taskData = fmt.Sprintf("%s every: %s,\n", taskData, opt.Every.String())
|
|
}
|
|
if opt.Offset != nil && !(*opt.Offset).IsZero() {
|
|
taskData = fmt.Sprintf("%s offset: %s,\n", taskData, opt.Offset.String())
|
|
}
|
|
if opt.Concurrency != nil && *opt.Concurrency != 0 {
|
|
taskData = fmt.Sprintf("%s concurrency: %d,\n", taskData, *opt.Concurrency)
|
|
}
|
|
if opt.Retry != nil && *opt.Retry != 0 {
|
|
taskData = fmt.Sprintf("%s retry: %d,\n", taskData, *opt.Retry)
|
|
}
|
|
if body == "" {
|
|
body = `from(bucket: "test")
|
|
|> range(start:-1h)`
|
|
}
|
|
|
|
return fmt.Sprintf(`option task = {
|
|
%s
|
|
}
|
|
|
|
%s`, taskData, body)
|
|
}
|
|
|
|
func TestNegDurations(t *testing.T) {
|
|
dur := options.MustParseDuration("-1m")
|
|
d, err := dur.DurationFrom(time.Now())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if d != -time.Minute {
|
|
t.Fatalf("expected duration to be -1m but was %s", d)
|
|
}
|
|
}
|
|
|
|
func TestFromScriptAST(t *testing.T) {
|
|
for _, c := range []struct {
|
|
script string
|
|
exp options.Options
|
|
shouldErr bool
|
|
}{
|
|
{script: scriptGenerator(options.Options{Name: "name0", Cron: "* * * * *", Concurrency: pointer.Int64(2), Retry: pointer.Int64(3), Offset: options.MustParseDuration("-1m")}, ""),
|
|
exp: options.Options{Name: "name0",
|
|
Cron: "* * * * *",
|
|
Concurrency: pointer.Int64(2),
|
|
Retry: pointer.Int64(3),
|
|
Offset: options.MustParseDuration("-1m")}},
|
|
{script: scriptGenerator(options.Options{Name: "name1", Every: *(options.MustParseDuration("5s"))}, ""), exp: options.Options{Name: "name1", Every: *(options.MustParseDuration("5s")), Concurrency: pointer.Int64(1), Retry: pointer.Int64(1)}},
|
|
{script: scriptGenerator(options.Options{Name: "name2", Cron: "* * * * *"}, ""), exp: options.Options{Name: "name2", Cron: "* * * * *", Concurrency: pointer.Int64(1), Retry: pointer.Int64(1)}},
|
|
{script: scriptGenerator(options.Options{Name: "name3", Every: *(options.MustParseDuration("1h")), Cron: "* * * * *"}, ""), shouldErr: true},
|
|
{script: scriptGenerator(options.Options{Name: "name4", Concurrency: pointer.Int64(1000), Every: *(options.MustParseDuration("1h"))}, ""), shouldErr: true},
|
|
{script: "option task = {\n name: \"name5\",\n concurrency: 0,\n every: 1m0s,\n\n}\n\nfrom(bucket: \"test\")\n |> range(start:-1h)", shouldErr: true},
|
|
{script: "option task = {\n name: \"name6\",\n concurrency: 1,\n every: 1,\n\n}\n\nfrom(bucket: \"test\")\n |> range(start:-1h)", shouldErr: true},
|
|
{script: scriptGenerator(options.Options{Name: "name7", Retry: pointer.Int64(20), Every: *(options.MustParseDuration("1h"))}, ""), shouldErr: true},
|
|
{script: "option task = {\n name: \"name8\",\n retry: 0,\n every: 1m0s,\n\n}\n\nfrom(bucket: \"test\")\n |> range(start:-1h)", shouldErr: true},
|
|
{script: scriptGenerator(options.Options{Name: "name9"}, ""), shouldErr: true},
|
|
{script: scriptGenerator(options.Options{}, ""), shouldErr: true},
|
|
{script: `option task = {
|
|
name: "name10",
|
|
every: 1d,
|
|
offset: 1m,
|
|
}
|
|
from(bucket: "metrics")
|
|
|> range(start: now(), stop: 8w)
|
|
`,
|
|
exp: options.Options{Name: "name10", Every: *(options.MustParseDuration("1d")), Concurrency: pointer.Int64(1), Retry: pointer.Int64(1), Offset: options.MustParseDuration("1m")},
|
|
},
|
|
{script: `option task = {
|
|
name: "name11",
|
|
every: 1m,
|
|
offset: 1d,
|
|
}
|
|
from(bucket: "metrics")
|
|
|> range(start: now(), stop: 8w)
|
|
|
|
`,
|
|
exp: options.Options{Name: "name11", Every: *(options.MustParseDuration("1m")), Concurrency: pointer.Int64(1), Retry: pointer.Int64(1), Offset: options.MustParseDuration("1d")},
|
|
},
|
|
{script: "option task = {name:\"test_task_smoke_name\", every:30s} from(bucket:\"test_tasks_smoke_bucket_source\") |> range(start: -1h) |> map(fn: (r) => ({r with _time: r._time, _value:r._value, t : \"quality_rocks\"}))|> to(bucket:\"test_tasks_smoke_bucket_dest\", orgID:\"3e73e749495d37d5\")",
|
|
exp: options.Options{Name: "test_task_smoke_name", Every: *(options.MustParseDuration("30s")), Retry: pointer.Int64(1), Concurrency: pointer.Int64(1)}, shouldErr: false}, // TODO(docmerlin): remove this once tasks fully supports all flux duration units.
|
|
|
|
} {
|
|
o, err := options.FromScriptAST(fluxlang.DefaultService, c.script)
|
|
if c.shouldErr && err == nil {
|
|
t.Fatalf("script %q should have errored but didn't", c.script)
|
|
} else if !c.shouldErr && err != nil {
|
|
t.Fatalf("script %q should not have errored, but got %v", c.script, err)
|
|
}
|
|
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
ignoreLocation := cmpopts.IgnoreFields(ast.BaseNode{}, "Loc")
|
|
|
|
if !cmp.Equal(o, c.exp, ignoreLocation) {
|
|
t.Fatalf("script %q got unexpected result -got/+exp\n%s", c.script, cmp.Diff(o, c.exp))
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestValidate(t *testing.T) {
|
|
good := options.Options{Name: "x", Cron: "* * * * *", Concurrency: pointer.Int64(1), Retry: pointer.Int64(1)}
|
|
if err := good.Validate(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
bad := new(options.Options)
|
|
*bad = good
|
|
bad.Name = ""
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for options without name")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Cron = ""
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for options without cron or every")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Every = *options.MustParseDuration("1m")
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for options with both cron and every")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Cron = "not a cron string"
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for options with invalid cron")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Cron = ""
|
|
bad.Every = *options.MustParseDuration("-1m")
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for negative every")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Offset = options.MustParseDuration("1500ms")
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for sub-second delay resolution")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Concurrency = pointer.Int64(0)
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for 0 concurrency")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Concurrency = pointer.Int64(math.MaxInt64)
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for concurrency too large")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Retry = pointer.Int64(0)
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for 0 retry")
|
|
}
|
|
|
|
*bad = good
|
|
bad.Retry = pointer.Int64(math.MaxInt64)
|
|
if err := bad.Validate(); err == nil {
|
|
t.Error("expected error for retry too large")
|
|
}
|
|
|
|
notbad := new(options.Options)
|
|
*notbad = good
|
|
notbad.Cron = ""
|
|
notbad.Every = *options.MustParseDuration("22d")
|
|
if err := notbad.Validate(); err != nil {
|
|
t.Error("expected no error for days every")
|
|
}
|
|
|
|
}
|
|
|
|
func TestEffectiveCronString(t *testing.T) {
|
|
for _, c := range []struct {
|
|
c string
|
|
e options.Duration
|
|
exp string
|
|
}{
|
|
{c: "10 * * * *", exp: "10 * * * *"},
|
|
{e: *(options.MustParseDuration("10s")), exp: "@every 10s"},
|
|
{exp: ""},
|
|
{e: *(options.MustParseDuration("10d")), exp: "@every 10d"},
|
|
} {
|
|
o := options.Options{Cron: c.c, Every: c.e}
|
|
got := o.EffectiveCronString()
|
|
if got != c.exp {
|
|
t.Fatalf("exp cron string %q, got %q for %v", c.exp, got, o)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDurationMarshaling(t *testing.T) {
|
|
t.Run("unmarshalling", func(t *testing.T) {
|
|
now := time.Now().UTC() /* to guarantee 24 hour days*/
|
|
dur1 := options.Duration{}
|
|
if err := dur1.UnmarshalText([]byte("1d1h10m3s")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d1, err1 := dur1.DurationFrom(now)
|
|
if err1 != nil {
|
|
t.Fatal(err1)
|
|
}
|
|
|
|
dur2 := options.Duration{}
|
|
if err := dur2.Parse("1d1h10m3s"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d2, err2 := dur2.DurationFrom(now)
|
|
if err2 != nil {
|
|
t.Fatal(err2)
|
|
}
|
|
|
|
if d1 != d2 || d1 != 25*time.Hour+10*time.Minute+3*time.Second /* we know that this day is 24 hours long because its UTC and go ignores leap seconds*/ {
|
|
t.Fatal("Parse and Marshaling do not give us the same result")
|
|
}
|
|
})
|
|
|
|
t.Run("marshaling", func(t *testing.T) {
|
|
dur := options.Duration{}
|
|
if err := dur.UnmarshalText([]byte("1h10m3s")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if dur.String() != "1h10m3s" {
|
|
t.Fatalf("duration string should be \"1h10m3s\" but was %s", dur.String())
|
|
}
|
|
text, err := dur.MarshalText()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if string(text) != "1h10m3s" {
|
|
t.Fatalf("duration text should be \"1h10m3s\" but was %s", text)
|
|
}
|
|
})
|
|
|
|
t.Run("parse zero", func(t *testing.T) {
|
|
dur := options.Duration{}
|
|
if err := dur.UnmarshalText([]byte("0h0s")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !dur.IsZero() {
|
|
t.Fatalf("expected duration \"0s\" to be zero but was %s", dur.String())
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestDurationMath(t *testing.T) {
|
|
dur := options.MustParseDuration("10s")
|
|
d, err := dur.DurationFrom(time.Now())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if d != 10*time.Second {
|
|
t.Fatalf("expected duration to be 10s but it was %s", d)
|
|
}
|
|
}
|