162 lines
7.0 KiB
Go
162 lines
7.0 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"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).
|
|
|
|
// 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.
|
|
|
|
// 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.
|
|
|
|
// 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 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.
|
|
|
|
// aRunningBucket makes the bucket the builders say the checks they run in, with an age of its own short enough for a test
|
|
// (the mesh's is two minutes): a key not renewed within it is gone.
|
|
func aRunningBucket(t *testing.T, js *broker.JetStream, age time.Duration) 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.CheckRunningBucket, History: 1, TTL: age})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return kv
|
|
}
|
|
|
|
// A builder taking a request for a pull request's head and trunk commit that another builder is running a check of drops it:
|
|
// the bucket of running checks names the head and the trunk, the builder and the build (issue 974).
|
|
|
|
// A key that is not renewed ages out, and the request for its head and trunk commit is then taken: a builder that died
|
|
// running a check leaves nothing that holds the next one back for longer than the bucket's age.
|
|
|
|
// A builder running a check renews its key on each heartbeat, so it outlives the bucket's age, and deletes it when the check
|
|
// ends, whatever the outcome.
|
|
|
|
// A key whose verdict is "error" — a check that could not run, or was stopped — judged nothing: a request for that head and
|
|
// trunk commit must run (issue 974, review 2).
|
|
func TestAnErrorVerdictJudgesNothing(t *testing.T) {
|
|
for verdict, judged := range map[string]bool{"pass": true, "warning": true, "fail": true, "error": false, "": false, "pending": false} {
|
|
k := PullKey{Head: "aaaaaaaaaaaa1111", Trunk: "tttttttttttt1111", Verdict: verdict}
|
|
if got := k.Judged("aaaaaaaaaaaa1111", "tttttttttttt1111"); got != judged {
|
|
t.Errorf("a key with verdict %q judged = %v; want %v", verdict, got, judged)
|
|
}
|
|
}
|
|
}
|
|
|
|
// A check stopped because the trunk moved leaves the newer request, for the head against the new trunk commit, taken and
|
|
// run: the stop is the stopped one's failure, not the key's verdict, and the newer request is not dropped by it.
|
|
|
|
// An unacknowledged check goes to a second builder: taken from a check worker and not acknowledged, it is handed out again
|
|
// when the worker's wait is up.
|
|
|
|
// fetchOne takes the next message of a worker and leaves it unacknowledged, as a builder that died would.
|
|
func fetchOne(js *broker.JetStream, worker string) (*nats.Msg, error) {
|
|
for _, w := range broker.CheckWorkers() {
|
|
if w.Name != worker {
|
|
continue
|
|
}
|
|
sub, err := js.Context().PullSubscribe(w.Filters[0], w.Name, nats.Bind(w.Stream, w.Name), nats.ManualAck())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
got, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return got[0], nil
|
|
}
|
|
return nil, nil
|
|
}
|