jschoubben
00b52e3c1f
A holder says ask-taken when an ask starts, step-progress every minute while it runs (phase fetch, build or run) and step-ended for an ask outside a pipeline, each on the seat's subject of its node (SayOn); the asker is the last token of the subject the ask came on (Pulled.Subject). Builds: design 47 G §496, §602, §603, §481 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
94 lines
2.9 KiB
Go
94 lines
2.9 KiB
Go
package steprun
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
|
|
)
|
|
|
|
type said struct {
|
|
event, id string
|
|
body map[string]any
|
|
}
|
|
|
|
type sayer struct {
|
|
mu sync.Mutex
|
|
got []said
|
|
}
|
|
|
|
func (s *sayer) say(event string, body any, id string) error {
|
|
raw, _ := json.Marshal(body)
|
|
var m map[string]any
|
|
_ = json.Unmarshal(raw, &m)
|
|
s.mu.Lock()
|
|
s.got = append(s.got, said{event, id, m})
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
func (s *sayer) of(event string) []said {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
var out []said
|
|
for _, g := range s.got {
|
|
if g.event == event {
|
|
out = append(out, g)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Design 47 G §496, §602 (BG4, BN4, BI4): a holder says ask-taken when an ask starts, step-progress while it runs and
|
|
// step-ended when an ask outside a pipeline ends, whatever its asker.
|
|
func TestPullSaysAskTakenProgressAndEnd(t *testing.T) {
|
|
for _, c := range []struct {
|
|
name string
|
|
runErr error
|
|
outcome string
|
|
}{{"passed", nil, sdkstep.OutcomePassed}, {"failed", errors.New("the build broke"), sdkstep.OutcomeFailed}} {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
src, st, sy := NewFakeAsks(), newSteps(), &sayer{}
|
|
a := testAsk(sdkstep.AskDelivery)
|
|
_ = src.PublishAsk(a)
|
|
release, entered := make(chan struct{}), make(chan struct{})
|
|
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
|
|
PullWait: 10 * time.Millisecond, Heartbeat: time.Hour, Say: sy.say, ProgressEvery: 5 * time.Millisecond,
|
|
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
|
|
Phase(ctx, "build m")
|
|
close(entered)
|
|
<-release
|
|
return nil, c.runErr
|
|
}}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go p.Pull(ctx)
|
|
<-entered
|
|
waitFor(t, "a step-progress in the build", func() bool {
|
|
pr := sy.of("step-progress")
|
|
return len(pr) > 0 && pr[len(pr)-1].body["phase"] == "build"
|
|
})
|
|
close(release)
|
|
waitFor(t, "step-ended", func() bool { return len(sy.of("step-ended")) == 1 })
|
|
taken := sy.of("ask-taken")
|
|
if len(taken) != 1 || taken[0].body["ask"] != a.ID || taken[0].body["node"] != "shanks" || taken[0].id != "ask-taken."+a.ID {
|
|
t.Fatalf("ask-taken = %+v, want one for %s on shanks", taken, a.ID)
|
|
}
|
|
if pr := sy.of("step-progress")[len(sy.of("step-progress"))-1].body; pr["ask"] != a.ID || pr["phase"] != "build" || pr["asker"] != a.Asker {
|
|
t.Fatalf("step-progress = %v, want the ask, its asker and the phase build", pr)
|
|
}
|
|
end := sy.of("step-ended")[0].body
|
|
if end["outcome"] != c.outcome || end["ask"] != a.ID || end["asker"] != a.Asker || end["node"] != "shanks" || end["at"] == nil {
|
|
t.Fatalf("step-ended = %v, want outcome %s of ask %s", end, c.outcome, a.ID)
|
|
}
|
|
if c.runErr != nil && end["said"] != "the build broke" {
|
|
t.Fatalf("step-ended said %v, want why", end["said"])
|
|
}
|
|
})
|
|
}
|
|
}
|