Skip to content
quarry Artifact [2cb3ec687b]
Login

Artifact [2cb3ec687b]

Artifact 2cb3ec687b852b07d4c4ef220a0c8f141b1d54d4976a8d02012bae9ef1e7749c:


package runner

import (
	"context"
	"encoding/json"
	"net/http/httptest"
	"strings"
	"testing"
	"time"
)

// A STEP THAT DEPENDS ON ANOTHER STILL KNOWS ITS CHECK-IN.
//
// dagnats hands a step with one dependency that step's output as its
// input, and a step with several a map of their outputs. pikchr-studio's
// deploy, the first quarry pipeline with depends_on, failed with "the
// task names no check-in" because the runner's output was "ok". The
// output is now the task's subject, so it flows down the graph.
func TestAStepAfterAnotherGetsTheCheckin(t *testing.T) {
	subjectIn := func(checkin string) json.RawMessage {
		b, _ := json.Marshal(map[string]string{"repo": "alpha", "checkin": checkin, "branch": "trunk"})
		return b
	}
	b := &fakeBridge{queue: map[string][]polled{
		"t-test":   {{TaskID: "test", RunID: "r", StepID: "test", Input: subjectIn("c1")}},
		"t-deploy": nil,
	}, resolved: map[string]map[string]any{}, logs: map[string]string{}}
	srv := httptest.NewServer(b.handler())
	defer srv.Close()
	manifest := `{"tasks":{"t-test":["true"],"t-deploy":["sh","-c","echo deployed"]}}`
	r, err := New(Config{
		Bridge: srv.URL, Token: "WT", Tasks: []string{"t-test", "t-deploy"}, MaxTasks: 1,
		WorkDir: t.TempDir(), Checkout: fakeCheckout{map[string]string{"c1": manifest}},
		Heartbeat: 50 * time.Millisecond, Logf: t.Logf,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { r.Run(ctx); close(done) }()
	wait := func(id string) map[string]any {
		for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(20 * time.Millisecond) {
			b.mu.Lock()
			got, ok := b.resolved[id]
			b.mu.Unlock()
			if ok {
				return got
			}
		}
		t.Fatalf("FAIL: task %s never resolved", id)
		return nil
	}
	// The engine's hand-off, done here by hand: the test's output becomes
	// the deploy's input (one dependency), then a fan-in of two.
	out := wait("test")
	if out["action"] != "complete" {
		t.Fatalf("FAIL: the test task resolved %v", out)
	}
	carried, _ := json.Marshal(out["output"])
	fanIn, _ := json.Marshal(map[string]json.RawMessage{"test": carried, "worker": subjectIn("c1")})
	split, _ := json.Marshal(map[string]json.RawMessage{"test": carried, "worker": subjectIn("c2")})
	b.mu.Lock()
	b.queue["t-deploy"] = []polled{
		{TaskID: "one", RunID: "r", StepID: "deploy", Input: carried},
		{TaskID: "fan", RunID: "r", StepID: "deploy", Input: fanIn},
		{TaskID: "split", RunID: "r", StepID: "deploy", Input: split},
	}
	b.mu.Unlock()
	one, fan, bad := wait("one"), wait("fan"), wait("split")
	cancel()
	<-done
	if one["action"] != "complete" || !strings.Contains(b.logs["one"], "deployed") {
		t.Errorf("FAIL: a step after one dependency resolved %v, log %q; want it run at c1", one, b.logs["one"])
	}
	if fan["action"] != "complete" {
		t.Errorf("FAIL: a step after two dependencies on c1 resolved %v; want it run", fan)
	}
	if bad["action"] != "fail" || !strings.Contains(bad["error"].(string), "names no check-in") {
		t.Errorf("FAIL: dependencies naming two check-ins resolved %v; want a refusal", bad)
	}
	t.Log("ok: the check-in flows through one dependency and a fan-in, and a disagreement is refused")
}