package definition import ( "bytes" "encoding/json" "strings" "testing" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" ) func literal(raw string) Value { return Value{Kind: "LITERAL", Literal: json.RawMessage(raw)} } func ref(source string, path ...string) Value { return Value{Kind: "REFERENCE", Reference: &Reference{Source: source, Path: path}} } func scalar(kind string) Type { return Type{Kind: kind} } func block(result Type, value Value, steps ...Step) Block { return Block{Steps: steps, OutputType: result, Output: value} } func testTask() Task { return Task{QueueID: "11111111111111111111111111111111", Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", InputType: scalar("STRING"), Input: ref("input", "operation"), ResultType: scalar("STRING"), TimeoutMS: 10000, Retry: pulse.RetryPolicy{MaxAttempts: 2, AttemptTimeoutMS: 3000, InitialDelayMS: 1000, MaxDelayMS: 2000}, RecoverySafety: pulse.RecoveryReconcileOnly} } func testSpecification() Specification { t := testTask() return Specification{SchemaVersion: 1, TimeoutMS: 60000, FailurePolicy: "REVIEW", InputType: Type{Kind: "OBJECT", Fields: map[string]Type{"operation": scalar("STRING"), "enabled": scalar("BOOLEAN")}}, Body: block(scalar("STRING"), ref("prepare"), Step{Key: "prepare", Kind: "TASK", Task: &t})} } func TestStructuredDefinitionScopesAndBudgets(t *testing.T) { base := testSpecification() left, right := ref("input", "enabled"), literal("true") task := testTask() branch := Branch{Cases: []BranchCase{{When: Predicate{Op: "EQ", Left: &left, Right: &right}, Body: block(scalar("STRING"), ref("chosen"), Step{Key: "chosen", Kind: "TASK", Task: &task})}}, Default: ptrBlock(block(scalar("STRING"), literal(`"skipped"`)))} parallel := Parallel{Branches: []ParallelBranch{{Name: "first", Body: block(scalar("STRING"), ref("choice"))}, {Name: "second", Body: block(scalar("NULL"), ref("timer"), Step{Key: "timer", Kind: "TIMER", Timer: &Timer{DelayMS: 1}})}}} base.Body = block(Type{Kind: "OBJECT", Fields: map[string]Type{"first": scalar("STRING"), "second": scalar("NULL")}}, ref("joined"), Step{Key: "choice", Kind: "BRANCH", Branch: &branch}, Step{Key: "joined", Kind: "PARALLEL", Parallel: ¶llel}) got, err := Check(base) if err != nil || got.Steps != 4 || got.MaxConcurrency != 1 || got.MaxTaskAttempts != 2 || len(got.Queues) != 1 { t.Fatal("structured definition", got, err) } for _, test := range []struct { name string change func(*Specification) }{ {"forward_reference", func(s *Specification) { s.Body.Steps[0].Task.Input = ref("later") }}, {"self_reference", func(s *Specification) { s.Body.Steps[0].Task.Input = ref("prepare") }}, {"missing_property", func(s *Specification) { s.Body.Steps[0].Task.Input = ref("input", "missing") }}, {"wrong_type", func(s *Specification) { s.Body.Steps[0].Task.Input = ref("input", "enabled") }}, {"default_policy_not_frozen", func(s *Specification) { s.Body.Steps[0].Task.Retry.MaxAttempts = 0 }}, {"unknown_kind", func(s *Specification) { s.Body.Steps[0].Kind = "SCRIPT" }}, {"extra_kind_body", func(s *Specification) { s.Body.Steps[0].Timer = &Timer{DelayMS: 1} }}, {"reserved_step_name", func(s *Specification) { s.Body.Steps[0].Key = "input" }}, {"type_cycle", func(s *Specification) { x := scalar("ARRAY"); x.Items = &x; s.InputType = x }}, {"invalid_unicode", func(s *Specification) { s.Body.Steps[0].Task.Input = literal(`"\ud800"`) }}, {"duplicate_literal_key", func(s *Specification) { s.Body.Steps[0].Task.InputType = scalar("JSON") s.Body.Steps[0].Task.Input = literal(`{"a":1,"a":2}`) }}, {"null_task_literal", func(s *Specification) { s.Body.Steps[0].Task.InputType = scalar("JSON") s.Body.Steps[0].Task.Input = literal(" null ") }}, {"empty_task_deadline", func(s *Specification) { s.Body.Steps[0].Task.TimeoutMS = 0 }}, } { t.Run(test.name, func(t *testing.T) { s := testSpecification() test.change(&s) if _, err := Check(s); err == nil { t.Fatal("invalid definition accepted") } }) } branch.Cases[0].Body.Output = ref("joined", "first") if _, err := Check(base); err == nil { t.Fatal("future join result was visible inside a branch") } branch.Cases[0].Body.Output = ref("chosen") base.Body.Output = ref("chosen") if _, err := Check(base); err == nil { t.Fatal("conditional child leaked outside composite") } base.Body.Output = ref("joined") parallel.Branches[1].Body.Output = ref("chosen") if _, err := Check(base); err == nil { t.Fatal("sibling/unselected result leaked") } parallel.Branches[1].Body.Output = ref("timer") branch.Default = nil if _, err := Check(base); err == nil { t.Fatal("nonexhaustive branch accepted") } } func ptrBlock(b Block) *Block { return &b } func TestDefinitionParallelBudgetAndDuplicateIdentity(t *testing.T) { makeParallel := func(prefix string) Step { p := Parallel{} for _, suffix := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} { task := testTask() key := prefix + suffix p.Branches = append(p.Branches, ParallelBranch{Name: suffix, Body: block(scalar("STRING"), ref(key), Step{Key: key, Kind: "TASK", Task: &task})}) } return Step{Key: prefix, Kind: "PARALLEL", Parallel: &p} } s := testSpecification() p := makeParallel("wide") s.Body = block(scalar("JSON"), ref("wide"), p) got, err := Check(s) if err != nil || got.MaxConcurrency != 8 || got.Steps != 9 || got.MaxTaskAttempts != 16 { t.Fatal(got, err) } p.Parallel.Branches[1].Body.Steps[0].Key = "widea" if _, err := Check(s); err == nil { t.Fatal("duplicate step identities accepted") } outer := Parallel{} for _, prefix := range []string{"a", "b", "c", "d", "e"} { child := makeParallel(prefix) outer.Branches = append(outer.Branches, ParallelBranch{Name: prefix, Body: block(scalar("JSON"), ref(prefix), child)}) } s.Body = block(scalar("JSON"), ref("outer"), Step{Key: "outer", Kind: "PARALLEL", Parallel: &outer}) if _, err := Check(s); err == nil { t.Fatal("nested concurrency over 32 accepted") } s = testSpecification() for i := 0; i < 128; i++ { task := testTask() s.Body.Steps = append(s.Body.Steps, Step{Key: "extra" + strings.Repeat("x", i+1), Kind: "TASK", Task: &task}) } if _, err := Check(s); err == nil { t.Fatal("overlong block accepted") } } func TestPollPartitionsAndIndependentBudgets(t *testing.T) { s := testSpecification() task := testTask() task.ResultType = Type{Kind: "OBJECT", Fields: map[string]Type{"state": {Kind: "STRING", Enum: []string{"processing", "done", "failed", "review"}}, "reference": scalar("STRING")}} p := Poll{Task: task, MaxRounds: 3, TimeoutMS: 30000, IntervalMS: 1000, StatusPath: []string{"state"}, Pending: []string{"processing"}, Succeeded: []string{"done"}, Failed: []string{"failed"}, Review: []string{"review"}, FinalSignal: &Signal{Name: "completion", PayloadVersion: "v1", Correlation: ref("input", "operation"), PayloadType: task.ResultType, TimeoutMS: 30000}} s.Body = block(task.ResultType, ref("poll"), Step{Key: "poll", Kind: "POLL", Poll: &p}) got, err := Check(s) if err != nil || got.MaxTaskAttempts != 6 || got.Steps != 1 { t.Fatal(got, err) } for _, change := range []func(*Poll){ func(p *Poll) { p.MaxRounds = 0 }, func(p *Poll) { p.MaxRounds = 1001 }, func(p *Poll) { p.IntervalMS = 0 }, func(p *Poll) { p.IntervalMS = p.TimeoutMS }, func(p *Poll) { p.Review = nil }, func(p *Poll) { p.Failed = []string{"processing"} }, func(p *Poll) { p.StatusPath = []string{"reference"} }, func(p *Poll) { p.MaxRounds = 1000; p.Task.Retry.MaxAttempts = 100 }, func(p *Poll) { p.Task.TimeoutMS = p.TimeoutMS + 1 }, func(p *Poll) { v := *p.FinalSignal; v.TimeoutMS++; p.FinalSignal = &v }, } { copy := p change(©) s.Body.Steps[0].Poll = © if _, err := Check(s); err == nil { t.Fatal("unbounded or ambiguous poll accepted") } } } func TestDefinitionDigestAndPreciseData(t *testing.T) { s := testSpecification() s.Body.Steps[0].Task.InputType = scalar("JSON") s.Body.Steps[0].Task.Input = literal(`{"n":9007199254740993,"decimal":12345678901234567890.1234567890123456789}`) a, err := Check(s) if err != nil { t.Fatal(err) } before, _ := json.Marshal(s) s.Body.Steps[0].Task.Input = literal(`{"decimal":12345678901234567890.123456789012345678900,"n":9007199254740993.0}`) b, err := Check(s) if err != nil || a.Digest != b.Digest { t.Fatal("semantic numeric digest changed", err) } s.Body.Steps[0].Task.Input = literal(`{"n":9007199254740992,"decimal":12345678901234567890.1234567890123456789}`) c, err := Check(s) if err != nil || a.Digest == c.Digest { t.Fatal("large integer rounded", err) } var old Specification if json.Unmarshal(before, &old) != nil { t.Fatal("snapshot") } if _, err := Check(old); err != nil { t.Fatal(err) } after, _ := json.Marshal(old) if !bytes.Equal(before, after) { t.Fatal("check mutated definition") } for _, raw := range []string{`"bad"`, `null`, `{"extra":1}`, `{"operation":"x","enabled":true,"extra":1}`, `{"operation":"x","enabled":1}`, `{"operation":"x","operation":"y","enabled":true}`} { if ValidateData(old.InputType, json.RawMessage(raw)) == nil { t.Fatal("bad instance input accepted") } } if err := ValidateData(old.InputType, json.RawMessage(`{"operation":"x","enabled":true}`)); err != nil { t.Fatal(err) } s = Specification{SchemaVersion: 1, InputType: scalar("NULL"), TimeoutMS: 1, FailurePolicy: "REVIEW", Body: block(scalar("NULL"), literal("null"))} if _, err := Check(s); err != nil { t.Fatal("empty typed block must be valid", err) } } func FuzzDefinitionDataNeverPanics(f *testing.F) { for _, raw := range []string{`{"operation":"x","enabled":true}`, `null`, `{"n":9007199254740993}`, `{"n":1,"n":2}`} { f.Add([]byte(raw)) } f.Fuzz(func(t *testing.T, raw []byte) { if len(raw) > 4096 { return } _ = ValidateData(testSpecification().InputType, raw) }) }