package cibridge
import (
"bufio"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
type seenReq struct {
method, path, auth string
body map[string]any
}
// A runner on its owner's machine reaches the engine's four worker
// routes through the repository's URL; the poll is pinned to the
// repository's group; the connect stream arrives as it is written; and
// the engine's admin bearer is refused on this path.
func TestTheBridgeForwardsTheWorkerRoutesAndPinsTheGroup(t *testing.T) {
seen := make(chan seenReq, 10)
engine := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var body map[string]any
json.NewDecoder(r.Body).Decode(&body)
seen <- seenReq{r.Method, r.URL.Path, r.Header.Get("Authorization"), body}
if r.URL.Path == "/v1/workers/connect" {
w.Header().Set("Content-Type", "text/event-stream")
fl := w.(http.Flusher)
io.WriteString(w, "event: heartbeat\ndata: {}\n\n")
fl.Flush()
time.Sleep(200 * time.Millisecond)
return
}
io.WriteString(w, `{"ok":true}`)
}))
defer engine.Close()
h, err := Handler(engine.URL, "ADMIN", func(repo string) string { return "g-" + repo }, func(repo string) bool { return repo == "alpha" })
if err != nil {
t.Fatal(err)
}
front := httptest.NewServer(h)
defer front.Close()
post := func(path, auth, body string) *http.Response {
t.Helper()
req, _ := http.NewRequest("POST", front.URL+path, strings.NewReader(body))
if auth != "" {
req.Header.Set("Authorization", "Bearer "+auth)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
return resp
}
// poll: pinned to the repository's group, token passed through.
resp := post("/r/alpha/ci/bridge/v1/tasks/poll", "WT", `{"task_types":["t"],"max_tasks":1}`)
resp.Body.Close()
if resp.StatusCode != 200 {
t.Fatalf("FAIL: poll = %d", resp.StatusCode)
}
got := <-seen
if got.path != "/v1/tasks/poll" || got.auth != "Bearer WT" || got.body["worker_group"] != "g-alpha" {
t.Fatalf("FAIL: engine saw %+v", got)
}
// A poll naming another repository's group is refused before the engine.
resp = post("/r/alpha/ci/bridge/v1/tasks/poll", "WT", `{"task_types":["t"],"worker_group":"g-beta"}`)
resp.Body.Close()
if resp.StatusCode != 400 {
t.Errorf("FAIL: a foreign group got %d", resp.StatusCode)
}
// resolve and logs keep the task id.
for _, p := range []string{"resolve", "logs"} {
resp = post("/r/alpha/ci/bridge/v1/tasks/task-7/"+p, "WT", `{}`)
resp.Body.Close()
if got := <-seen; got.path != "/v1/tasks/task-7/"+p {
t.Errorf("FAIL: %s reached %s", p, got.path)
}
}
// connect streams: the first event arrives before the engine finishes.
start := time.Now()
resp = post("/r/alpha/ci/bridge/v1/workers/connect", "WT", `{"worker_id":"w","task_types":["t"],"max_tasks":1}`)
line, _ := bufio.NewReader(resp.Body).ReadString('\n')
resp.Body.Close()
<-seen
if !strings.HasPrefix(line, "event: heartbeat") || time.Since(start) > 150*time.Millisecond {
t.Errorf("FAIL: the stream was held back: %q after %s", line, time.Since(start))
}
// The admin bearer, a missing token, an unknown repository: refused
// here, never forwarded.
for _, c := range []struct {
path, auth string
want int
}{
{"/r/alpha/ci/bridge/v1/tasks/poll", "ADMIN", 403},
{"/r/alpha/ci/bridge/v1/tasks/poll", "", 401},
{"/r/nosuch/ci/bridge/v1/tasks/poll", "WT", 404},
} {
resp = post(c.path, c.auth, `{}`)
resp.Body.Close()
if resp.StatusCode != c.want {
t.Errorf("FAIL: %s with %q = %d, want %d", c.path, c.auth, resp.StatusCode, c.want)
}
}
select {
case r := <-seen:
t.Errorf("FAIL: a refused request reached the engine: %+v", r)
default:
}
}