e2e/buildfollow_test.go

  1package e2e
  2
  3import (
  4	"encoding/json"
  5	"fmt"
  6	"io"
  7	"net/http"
  8	"os"
  9	"path/filepath"
 10	"strings"
 11	"testing"
 12	"time"
 13)
 14
 15// streamReader collects what r delivers, so a test can wait for text to
 16// arrive while the writer is still going.
 17type streamReader struct {
 18	ch  chan []byte
 19	buf strings.Builder
 20}
 21
 22func newStreamReader(r io.Reader) *streamReader {
 23	s := &streamReader{ch: make(chan []byte, 16)}
 24	go func() {
 25		b := make([]byte, 4096)
 26		for {
 27			n, err := r.Read(b)
 28			if n > 0 {
 29				s.ch <- append([]byte(nil), b[:n]...)
 30			}
 31			if err != nil {
 32				close(s.ch)
 33				return
 34			}
 35		}
 36	}()
 37	return s
 38}
 39
 40func (s *streamReader) waitFor(t *testing.T, want string) string {
 41	t.Helper()
 42	deadline := time.After(20 * time.Second)
 43	for !strings.Contains(s.buf.String(), want) {
 44		select {
 45		case b, ok := <-s.ch:
 46			if !ok {
 47				t.Fatalf("stream ended before %q:\n%s", want, s.buf.String())
 48			}
 49			s.buf.Write(b)
 50		case <-deadline:
 51			t.Fatalf("no %q after 20s:\n%s", want, s.buf.String())
 52		}
 53	}
 54	return s.buf.String()
 55}
 56
 57// queueBuild creates alice/app with one CI job and pushes it, which
 58// queues build 1. No runner is attached, so it stays queued.
 59func queueBuild(t *testing.T, inst *instance, aliceKey string) {
 60	t.Helper()
 61	if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
 62		t.Fatal("repo create failed")
 63	}
 64	work := t.TempDir()
 65	env := inst.gitEnv(aliceKey)
 66	mustGit(t, work, env, "clone", inst.sshURL("alice/app"), "w")
 67	dir := filepath.Join(work, "w")
 68	os.MkdirAll(filepath.Join(dir, ".gitbay"), 0o755)
 69	os.WriteFile(filepath.Join(dir, ".gitbay", "ci.yml"), []byte("jobs:\n  unit:\n    steps:\n      - echo fine\n"), 0o644)
 70	mustGit(t, dir, env, "checkout", "-q", "-b", "main")
 71	mustGit(t, dir, env, "add", ".")
 72	mustGit(t, dir, env, "commit", "-q", "-m", "ci")
 73	mustGit(t, dir, env, "push", "-q", "origin", "main")
 74}
 75
 76// claimBuild claims the oldest pending build with an admin key, as a
 77// runner would, and returns its id for runner log and runner done.
 78func claimBuild(t *testing.T, inst *instance, runnerKey string) string {
 79	t.Helper()
 80	out, errOut, code := inst.ssh(t, runnerKey, "", "runner", "next", "--json")
 81	if code != 0 {
 82		t.Fatalf("runner next: %s", errOut)
 83	}
 84	var claim struct {
 85		Data struct {
 86			ID int64 `json:"id"`
 87		} `json:"data"`
 88	}
 89	if err := json.Unmarshal([]byte(out), &claim); err != nil || claim.Data.ID == 0 {
 90		t.Fatalf("runner next output %q: %v", out, err)
 91	}
 92	return fmt.Sprint(claim.Data.ID)
 93}
 94
 95// A running build is followed over ssh and on its page: output the runner
 96// sends arrives while the build runs, and both end with the outcome.
 97func TestBuildLogFollow(t *testing.T) {
 98	t.Parallel()
 99	inst := startInstance(t)
100	aliceKey := inst.newKey(t, "alice")
101	runnerKey := inst.newKey(t, "ci")
102	inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
103	inst.admin(t, "admin", "user", "create", "ci", "--key", runnerKey+".pub", "--admin")
104	queueBuild(t, inst, aliceKey)
105
106	// Claim build 1 by hand, so the test decides when output arrives.
107	id := claimBuild(t, inst, runnerKey)
108
109	cmd := inst.sshCmd(aliceKey, "build", "log", "alice/app", "1", "--follow")
110	stdout, err := cmd.StdoutPipe()
111	if err != nil {
112		t.Fatal(err)
113	}
114	var stderr strings.Builder
115	cmd.Stderr = &stderr
116	if err := cmd.Start(); err != nil {
117		t.Fatal(err)
118	}
119	// If the test fails before the final cmd.Wait below, kill the follow
120	// instead of leaving it running. Once that Wait has run,
121	// cmd.ProcessState is set and this is a no-op.
122	defer func() {
123		if cmd.ProcessState == nil {
124			cmd.Process.Kill()
125			cmd.Wait()
126		}
127	}()
128	follow := newStreamReader(stdout)
129
130	page, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/alice/app/builds/1", inst.httpPort))
131	if err != nil {
132		t.Fatal(err)
133	}
134	defer page.Body.Close()
135	web := newStreamReader(page.Body)
136	web.waitFor(t, "Live: the log streams here")
137
138	// A static render while the build runs returns at once.
139	static := &http.Client{Timeout: 10 * time.Second}
140	resp, err := static.Get(fmt.Sprintf("http://127.0.0.1:%d/alice/app/builds/1?follow=0", inst.httpPort))
141	if err != nil {
142		t.Fatalf("?follow=0 did not return: %v", err)
143	}
144	body, _ := io.ReadAll(resp.Body)
145	resp.Body.Close()
146	if strings.Contains(string(body), "Live:") {
147		t.Fatalf("?follow=0 rendered the live page:\n%s", body)
148	}
149
150	if _, errOut, code := inst.ssh(t, runnerKey, "hello from the runner <b>\n", "runner", "log", id); code != 0 {
151		t.Fatalf("runner log: %s", errOut)
152	}
153	follow.waitFor(t, "hello from the runner <b>\n")
154	web.waitFor(t, "hello from the runner &lt;b&gt;")
155
156	if _, errOut, code := inst.ssh(t, runnerKey, "", "runner", "done", id, "success"); code != 0 {
157		t.Fatalf("runner done: %s", errOut)
158	}
159	web.waitFor(t, `<p class="notice" role="status">build finished: success</p>`)
160	web.waitFor(t, "</html>")
161	if err := cmd.Wait(); err != nil {
162		t.Fatalf("follow exited: %v\n%s", err, stderr.String())
163	}
164	if got := strings.TrimSpace(stderr.String()); got != "build 1 success" {
165		t.Errorf("follow stderr %q", got)
166	}
167}