310 lines
12 KiB
Go
310 lines
12 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"reflect"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
)
|
|
|
|
// The checks of pull requests are queued by the pull request (novox/hq ADR 0406 §3, §4, issue 974).
|
|
|
|
func aPullCheck(id string, number int, head, trunk string, ready bool) BuildRequest {
|
|
return BuildRequest{ID: id, Repository: "https://forge.example/novox/a.git", Ref: head,
|
|
Check: &CheckRequest{Owner: "novox", Repo: "a", Number: number, Trunk: trunk, Ready: ready}}
|
|
}
|
|
|
|
// The subject a check is asked on says its queue and its pull request: ready or other, `<owner>.<repo>.<number>` with
|
|
// a dot in a name made an underscore, `group.<name>` for a group's, the id for a try; a build is on the seat's own queue.
|
|
func TestACheckIsAskedOnItsQueueBySubject(t *testing.T) {
|
|
for _, c := range []struct {
|
|
r BuildRequest
|
|
want string
|
|
}{
|
|
{aPullCheck("b1", 7, "h", "t", false), "mesh.seat.node-build-agent.check.other.novox.a.7"},
|
|
{aPullCheck("b2", 7, "h", "t", true), "mesh.seat.node-build-agent.check.ready.novox.a.7"},
|
|
{BuildRequest{ID: "b3", Check: &CheckRequest{Owner: "no.vox", Repo: "a.b", Number: 12}}, "mesh.seat.node-build-agent.check.other.no_vox.a_b.12"},
|
|
{BuildRequest{ID: "b4", Check: &CheckRequest{Owner: "novox", Repo: "a", Group: "g.1", Ready: true}}, "mesh.seat.node-build-agent.check.ready.group.g_1"},
|
|
{BuildRequest{ID: "b5", Check: &CheckRequest{Owner: "novox", Repo: "a", Try: true}}, "mesh.seat.node-build-agent.check.other.try.b5"},
|
|
{BuildRequest{ID: "b6", Repository: "/r"}, "mesh.seat.node-build-agent.accept.build"},
|
|
{BuildRequest{ID: "b7", Repository: "/r", DryRun: true}, "mesh.seat.node-build-agent.accept.build"},
|
|
} {
|
|
if got := WorkSubjectOf(TheBuildMachine, c.r); got != c.want {
|
|
t.Errorf("%s is asked on %s; want %s", c.r.ID, got, c.want)
|
|
}
|
|
}
|
|
// The retired role has no check queues: its checks wait with its builds.
|
|
if got := WorkSubjectOf(TheBuildMachineBefore, aPullCheck("b8", 7, "h", "t", true)); got != "mesh.seat.mesh-build-machine.accept.build" {
|
|
t.Errorf("a check of the retired role is asked on %s", got)
|
|
}
|
|
}
|
|
|
|
func queueOf(t *testing.T, js *broker.JetStream) Queue {
|
|
t.Helper()
|
|
q, err := ReadQueue(context.Background(), js, TheBuildMachine)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return q
|
|
}
|
|
|
|
// A newer request for a pull request replaces the one still waiting, by the stream's rule, and each open pull request is
|
|
// left with one queued check (acceptance 5).
|
|
func TestANewerCheckRequestReplacesOneStillWaiting(t *testing.T) {
|
|
js := aBusWithTheBuildRole(t)
|
|
ctx := t.Context()
|
|
asks := BuildsOn(js, TheBuildMachine)
|
|
for _, r := range []BuildRequest{
|
|
aPullCheck("check-7-a", 7, "head-a-0000000", "trunk-0-00000", false),
|
|
aPullCheck("check-8-a", 8, "head-b-0000000", "trunk-0-00000", false),
|
|
aPullCheck("check-7-b", 7, "head-a-0000000", "trunk-1-00000", false), // a merge moved the trunk
|
|
aPullCheck("check-8-b", 8, "head-b-0000000", "trunk-1-00000", false),
|
|
aPullCheck("check-7-c", 7, "head-a-0000000", "trunk-2-00000", false), // and a second merge within the minute
|
|
} {
|
|
if err := asks.Ask(ctx, r); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
var got []string
|
|
for _, a := range queueOf(t, js).Asks {
|
|
if a.State != AskWaiting {
|
|
t.Errorf("%s is %s", a.ID, a.State)
|
|
}
|
|
got = append(got, a.ID)
|
|
}
|
|
if want := []string{"check-8-b", "check-7-c"}; !reflect.DeepEqual(got, want) {
|
|
t.Fatalf("queued %v; want one check a pull request, the newest: %v", got, want)
|
|
}
|
|
}
|
|
|
|
// A delivery build is taken before a check of a ready pull request, that before any other check, and the others in the
|
|
// order they were asked.
|
|
func TestADeliveryBuildIsTakenBeforeAReadyCheckAndThatBeforeAnyOther(t *testing.T) {
|
|
js := aBusWithTheBuildRole(t)
|
|
ctx, stop := context.WithCancel(t.Context())
|
|
defer stop()
|
|
asks := BuildsOn(js, TheBuildMachine)
|
|
for _, r := range []BuildRequest{
|
|
aPullCheck("other-1", 1, "h1", "t", false),
|
|
aPullCheck("other-2", 2, "h2", "t", false),
|
|
aPullCheck("ready-3", 3, "h3", "t", true),
|
|
{ID: "delivery-4", Repository: "/d"},
|
|
aPullCheck("other-5", 5, "h5", "t", false),
|
|
aPullCheck("ready-6", 6, "h6", "t", true),
|
|
} {
|
|
if err := asks.Ask(ctx, r); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
took := make(chan string, 8)
|
|
machine := MachineOverNATS(js, "anchor")
|
|
defer machine.Close()
|
|
go func() {
|
|
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
|
|
took <- work.Request().ID
|
|
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
|
|
_ = work.Done()
|
|
})
|
|
}()
|
|
var order []string
|
|
for len(order) < 6 {
|
|
select {
|
|
case id := <-took:
|
|
order = append(order, id)
|
|
case <-time.After(15 * time.Second):
|
|
t.Fatalf("taken so far: %v", order)
|
|
}
|
|
}
|
|
if want := []string{"delivery-4", "ready-3", "ready-6", "other-1", "other-2", "other-5"}; !reflect.DeepEqual(order, want) {
|
|
t.Fatalf("taken as %v; want %v", order, want)
|
|
}
|
|
}
|
|
|
|
// aKeyBucket makes mesh-delivery's bucket of pull requests' keys, which the bus here has not been given by that module.
|
|
func aKeyBucket(t *testing.T, js *broker.JetStream) jetstream.KeyValue {
|
|
t.Helper()
|
|
api, err := jetstream.New(js.Conn())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
kv, err := api.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: broker.PullsBucket})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return kv
|
|
}
|
|
|
|
func putKey(t *testing.T, kv jetstream.KeyValue, number int, k PullKey) {
|
|
t.Helper()
|
|
body, _ := json.Marshal(k)
|
|
if _, err := kv.Put(t.Context(), PullKeyOf("novox", "a", number), body); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// A request left on the third queue for a head and a trunk commit its key says were judged is acknowledged and dropped when a
|
|
// builder takes it; one for a head the key does not say judged is run.
|
|
func TestARequestLeftBehindForAJudgedHeadIsDroppedWhenTaken(t *testing.T) {
|
|
js := aBusWithTheBuildRole(t)
|
|
kv := aKeyBucket(t, js)
|
|
ctx, stop := context.WithCancel(t.Context())
|
|
defer stop()
|
|
putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-1-000000", Verdict: "pass"})
|
|
putKey(t, kv, 8, PullKey{Head: "head-8-000000000", Trunk: "trunk-1-000000"}) // asked, not judged
|
|
putKey(t, kv, 9, PullKey{Head: "head-9-000000000", Trunk: "trunk-2-000000", Verdict: "pass"}) // judged against another trunk
|
|
asks := BuildsOn(js, TheBuildMachine)
|
|
for _, r := range []BuildRequest{
|
|
aPullCheck("left-7", 7, "head-7-000000000", "trunk-1-000000", false),
|
|
aPullCheck("fresh-8", 8, "head-8-000000000", "trunk-1-000000", false),
|
|
aPullCheck("moved-9", 9, "head-9-000000000", "trunk-1-000000", false),
|
|
} {
|
|
if err := asks.Ask(ctx, r); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
took := make(chan string, 8)
|
|
machine := MachineOverNATS(js, "anchor")
|
|
defer machine.Close()
|
|
go func() {
|
|
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
|
|
took <- work.Request().ID
|
|
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
|
|
_ = work.Done()
|
|
})
|
|
}()
|
|
var ran []string
|
|
for len(ran) < 2 {
|
|
select {
|
|
case id := <-took:
|
|
ran = append(ran, id)
|
|
case <-time.After(15 * time.Second):
|
|
t.Fatalf("run so far: %v", ran)
|
|
}
|
|
}
|
|
if want := []string{"fresh-8", "moved-9"}; !reflect.DeepEqual(ran, want) {
|
|
t.Fatalf("run %v; want %v, the judged one dropped", ran, want)
|
|
}
|
|
time.Sleep(300 * time.Millisecond)
|
|
if n := len(queueOf(t, js).Asks); n != 0 {
|
|
t.Errorf("%d ask(s) are left in the queues; the dropped one was to be acknowledged", n)
|
|
}
|
|
}
|
|
|
|
// A running check stops when its key's head moves or its trunk commit does, not when the key only gains its verdict, and a
|
|
// group's check, which a member's merge in the group's order writes nothing to the key of, is not stopped at all.
|
|
func TestARunningCheckStopsWhenItsKeysHeadMovesAndAGroupsDoesNot(t *testing.T) {
|
|
js := aBusWithTheBuildRole(t)
|
|
kv := aKeyBucket(t, js)
|
|
ctx, stop := context.WithCancel(t.Context())
|
|
defer stop()
|
|
moved := make(chan PullKey, 4)
|
|
req := aPullCheck("check-7", 7, "head-7-000000000", "trunk-1-000000", false)
|
|
end, err := WatchPull(ctx, js.Conn(), req, func(k PullKey) { moved <- k })
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer end()
|
|
putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-1-000000"})
|
|
putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-1-000000", Verdict: "pass"})
|
|
select {
|
|
case k := <-moved:
|
|
t.Fatalf("stopped by a key that moved nothing: %+v", k)
|
|
case <-time.After(500 * time.Millisecond):
|
|
}
|
|
putKey(t, kv, 7, PullKey{Head: "head-7-111111111", Trunk: "trunk-1-000000"})
|
|
select {
|
|
case k := <-moved:
|
|
if k.Head != "head-7-111111111" {
|
|
t.Errorf("stopped by %+v", k)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("a key whose head moved did not stop the check")
|
|
}
|
|
|
|
// And the trunk commit.
|
|
req2 := aPullCheck("check-8", 8, "head-8-000000000", "trunk-1-000000", false)
|
|
end2, err := WatchPull(ctx, js.Conn(), req2, func(k PullKey) { moved <- k })
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer end2()
|
|
putKey(t, kv, 8, PullKey{Head: "head-8-000000000", Trunk: "trunk-2-000000"})
|
|
select {
|
|
case <-moved:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("a key whose trunk commit moved did not stop the check")
|
|
}
|
|
|
|
// A group's check watches nothing: a sibling's merge in the group's order writes nothing to a key, and a key it does
|
|
// have is the pull request's own.
|
|
group := BuildRequest{ID: "group-1", Ref: "head-7-000000000",
|
|
Check: &CheckRequest{Owner: "novox", Repo: "a", Number: 7, Group: "g1"}}
|
|
end3, err := WatchPull(ctx, js.Conn(), group, func(k PullKey) { moved <- k })
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer end3()
|
|
putKey(t, kv, 7, PullKey{Head: "head-7-222222222", Trunk: "trunk-9-000000"})
|
|
select {
|
|
case k := <-moved:
|
|
t.Fatalf("a group's check was stopped by %+v", k)
|
|
case <-time.After(500 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func TestAPullKeyIsMovedByHeadOrTrunkAndJudgedByVerdict(t *testing.T) {
|
|
k := PullKey{Head: "aaaaaaaaaaaa1111", Trunk: "tttttttttttt1111", Verdict: "pass"}
|
|
if k.Moved("aaaaaaaaaaaa", "tttttttttttt") || !k.Moved("bbbbbbbbbbbbbbbb", "tttttttttttt") || !k.Moved("aaaaaaaaaaaa", "uuuuuuuuuuuuuu") {
|
|
t.Error("moved is wrong")
|
|
}
|
|
if !k.Judged("aaaaaaaaaaaa1111", "tttttttttttt1111") || k.Judged("aaaaaaaaaaaa1111", "") || (PullKey{Head: "a", Trunk: "t"}).Judged("a", "t") {
|
|
t.Error("judged is wrong")
|
|
}
|
|
}
|
|
|
|
// The queue is read across the three streams in the order they are handed out, each ask says the stream it waits in, and an
|
|
// ask is removed from that stream.
|
|
func TestTheQueueIsReadAcrossTheThreeStreamsAndAnAskIsDeletedFromItsOwn(t *testing.T) {
|
|
js := aBusWithTheBuildRole(t)
|
|
ctx := t.Context()
|
|
asks := BuildsOn(js, TheBuildMachine)
|
|
for _, r := range []BuildRequest{
|
|
aPullCheck("other-1", 1, "h1", "t", false),
|
|
aPullCheck("ready-2", 2, "h2", "t", true),
|
|
{ID: "delivery-3", Repository: "/d"},
|
|
} {
|
|
if err := asks.Ask(ctx, r); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
q := queueOf(t, js)
|
|
var ids, streams []string
|
|
for _, a := range q.Asks {
|
|
ids, streams = append(ids, a.ID), append(streams, a.Stream)
|
|
}
|
|
if want := []string{"delivery-3", "ready-2", "other-1"}; !reflect.DeepEqual(ids, want) {
|
|
t.Fatalf("read %v; want the order they are handed out, %v", ids, want)
|
|
}
|
|
if want := []string{"SEAT_NODE_BUILD_AGENT", broker.ChecksReadyStream, broker.ChecksOtherStream}; !reflect.DeepEqual(streams, want) {
|
|
t.Fatalf("streams %v; want %v", streams, want)
|
|
}
|
|
ready, _ := q.Find("ready-2")
|
|
if w := WorkerOf(TheBuildMachine, ready); w.Name != broker.ChecksReadyWorker {
|
|
t.Errorf("the ready check is handed out by %s", w.Name)
|
|
}
|
|
if removed, err := DeleteAsk(ctx, js, TheBuildMachine, ready); err != nil || removed != ready.Seq {
|
|
t.Fatalf("deleted %d, %v", removed, err)
|
|
}
|
|
if removed, err := DeleteAsk(ctx, js, TheBuildMachine, ready); err != nil || removed != 0 {
|
|
t.Fatalf("deleted again: %d, %v", removed, err)
|
|
}
|
|
if n := len(queueOf(t, js).Asks); n != 2 {
|
|
t.Errorf("%d asks left; want 2", n)
|
|
}
|
|
}
|