1package oprunner23import (4 "bufio"5 "context"6 "errors"7 "flag"8 "fmt"9 "io"10 "net/http"11 "net/http/httptest"12 "os"13 "strings"14 "sync"15 "testing"16 "time"17)1819// The helper subprocess.20//21// A runner test needs a program whose exact behaviour it wrote: one that says a22// slog record, one that fails, one that runs until it is stopped. Re-running23// this test binary under a marker argument is that program, so the tests need24// nothing on PATH and nothing built beforehand.2526const helperMarker = "-atlas-oprunner-helper"2728func TestHelperProcess(t *testing.T) {29 args := flag.Args()30 if len(args) == 0 || args[0] != helperMarker {31 return // an ordinary run of the suite; this test is not one of the cases32 }33 switch args[1] {34 case "speaks":35 fmt.Fprintln(os.Stderr,36 `{"time":"2026-08-02T10:00:00Z","level":"INFO","msg":"crawl finished","op":"crawl","volume":"tunic","fetched":37}`)37 fmt.Fprintln(os.Stdout, "http://127.0.0.1:6180")38 fmt.Fprintln(os.Stderr, "a line nobody structured")39 case "fails":40 fmt.Fprintln(os.Stderr,41 `{"level":"ERROR","msg":"archive not found","op":"tiles","path":"/nowhere"}`)42 os.Exit(3)43 case "waits":44 fmt.Fprintln(os.Stderr, `{"level":"INFO","msg":"started","op":"crawl"}`)45 time.Sleep(30 * time.Second)46 case "dawdles":47 fmt.Fprintln(os.Stderr, `{"level":"INFO","msg":"started","op":"crawl"}`)48 time.Sleep(1500 * time.Millisecond)49 fmt.Fprintln(os.Stderr, `{"level":"INFO","msg":"finished","op":"crawl"}`)50 }51 os.Exit(0)52}5354func helper(name, mode string) Operation {55 return Operation{56 Name: name,57 Argv: []string{os.Args[0], "-test.run=TestHelperProcess", "--", helperMarker, mode},58 }59}6061// collect runs an operation and gathers its rows.62func collect(t *testing.T, ctx context.Context, r *Runner, op Operation) ([]Row, error) {63 t.Helper()64 var rows []Row65 err := r.Run(ctx, op, func(row Row) error {66 rows = append(rows, row)67 return nil68 })69 return rows, err70}7172func TestRunCarriesTheSubprocessesOwnVoice(t *testing.T) {73 rows, err := collect(t, context.Background(), &Runner{}, helper("crawl", "speaks"))74 if err != nil {75 t.Fatal(err)76 }77 if len(rows) < 4 {78 t.Fatalf("a run that said three things produced %d rows: %+v", len(rows), rows)79 }80 if rows[0].Kind != KindCommand || !strings.Contains(rows[0].Message, helperMarker) {81 t.Errorf("the first row is not the command: %+v", rows[0])82 }83 if rows[0].Attrs[0] != (Attr{Key: "op", Value: "crawl"}) {84 t.Errorf("the command row does not name its operation: %+v", rows[0].Attrs)85 }8687 var event, output, plain *Row88 for at := range rows {89 switch row := &rows[at]; {90 case row.Kind == KindEvent:91 event = row92 case row.Kind == KindOutput && row.Stream == StreamOut:93 output = row94 case row.Kind == KindOutput && row.Stream == StreamErr:95 plain = row96 }97 }98 if event == nil {99 t.Fatalf("no event row: %+v", rows)100 }101 if event.Level != "INFO" || event.Message != "crawl finished" || event.Time != "2026-08-02T10:00:00Z" {102 t.Errorf("the record was not read: %+v", event)103 }104 // The documented vocabulary comes first, whatever order the record was105 // written in, and a count reads as a count.106 want := []Attr{{"op", "crawl"}, {"volume", "tunic"}, {"fetched", "37"}}107 if len(event.Attrs) != len(want) {108 t.Fatalf("attrs %+v, want %+v", event.Attrs, want)109 }110 for at := range want {111 if event.Attrs[at] != want[at] {112 t.Errorf("attr %d is %+v, want %+v", at, event.Attrs[at], want[at])113 }114 }115 if output == nil || output.Message != "http://127.0.0.1:6180" {116 t.Errorf("product output did not arrive as its own row: %+v", output)117 }118 if plain == nil || plain.Message != "a line nobody structured" {119 t.Errorf("an unstructured line was not carried verbatim: %+v", plain)120 }121 last := rows[len(rows)-1]122 if last.Kind != KindResult || last.Failed || last.Message != "done" {123 t.Errorf("the run did not end with a plain result: %+v", last)124 }125 for at, row := range rows {126 if row.Seq != at+1 {127 t.Errorf("row %d carries seq %d", at, row.Seq)128 }129 }130}131132func TestAFailedOperationSaysSoInItsLastRow(t *testing.T) {133 rows, err := collect(t, context.Background(), &Runner{}, helper("tiles", "fails"))134 if err != nil {135 t.Fatal(err)136 }137 last := rows[len(rows)-1]138 if last.Kind != KindResult || !last.Failed {139 t.Fatalf("a program that exited 3 ended with %+v", last)140 }141 if !strings.Contains(last.Message, "3") {142 t.Errorf("the result does not carry the exit status: %q", last.Message)143 }144 var errorRow *Row145 for at := range rows {146 if rows[at].Kind == KindEvent {147 errorRow = &rows[at]148 }149 }150 if errorRow == nil || !errorRow.Failed || errorRow.Message != "archive not found" {151 t.Errorf("the failure's own event did not arrive marked: %+v", errorRow)152 }153}154155func TestAProgramThatWillNotStartIsToldThroughTheStream(t *testing.T) {156 rows, err := collect(t, context.Background(), &Runner{},157 Operation{Name: "crawl", Argv: []string{"/nowhere/atlas", "crawl"}})158 if err != nil {159 t.Fatal(err)160 }161 last := rows[len(rows)-1]162 if last.Kind != KindResult || !last.Failed {163 t.Fatalf("a missing program ended with %+v", last)164 }165}166167func TestOnlyOneOperationRunsAtATime(t *testing.T) {168 runner := &Runner{}169 started := make(chan struct{})170 ctx, stop := context.WithCancel(context.Background())171 defer stop()172173 var wg sync.WaitGroup174 wg.Add(1)175 go func() {176 defer wg.Done()177 var once sync.Once178 _ = runner.Run(ctx, helper("crawl", "waits"), func(row Row) error {179 if row.Kind == KindEvent {180 once.Do(func() { close(started) })181 }182 return nil183 })184 }()185 select {186 case <-started:187 case <-time.After(30 * time.Second):188 t.Fatal("the first operation never spoke")189 }190191 if held := runner.Busy(); held != "crawl" {192 t.Errorf("the runner says it is running %q", held)193 }194 if _, err := collect(t, context.Background(), runner, helper("tiles", "speaks")); !errors.Is(err, ErrBusy) {195 t.Errorf("a second operation was let in: %v", err)196 }197198 stop()199 wg.Wait()200 if held := runner.Busy(); held != "" {201 t.Errorf("the slot was not released; it holds %q", held)202 }203 // And the slot really is free again.204 if _, err := collect(t, context.Background(), runner, helper("tiles", "speaks")); err != nil {205 t.Errorf("the runner stayed busy after its operation ended: %v", err)206 }207}208209func TestAnAbandonedRunStopsItsSubprocess(t *testing.T) {210 ctx, stop := context.WithCancel(context.Background())211 started := make(chan struct{})212 done := make(chan struct{})213 go func() {214 defer close(done)215 var once sync.Once216 _ = (&Runner{}).Run(ctx, helper("crawl", "waits"), func(row Row) error {217 if row.Kind == KindEvent {218 once.Do(func() { close(started) })219 }220 return nil221 })222 }()223 <-started224 stop()225 select {226 case <-done:227 case <-time.After(10 * time.Second):228 t.Fatal("the operation outlived the context that started it")229 }230}231232func TestAnEmitThatFailsEndsTheRun(t *testing.T) {233 refuse := errors.New("the page went away")234 err := (&Runner{}).Run(context.Background(), helper("crawl", "waits"), func(row Row) error {235 if row.Kind == KindEvent {236 return refuse237 }238 return nil239 })240 if !errors.Is(err, refuse) {241 t.Fatalf("run ended with %v, want the emit's own error", err)242 }243}244245func TestOperationValidate(t *testing.T) {246 cases := []struct {247 what string248 op Operation249 ok bool250 }{251 {"a command", Operation{Name: "crawl", Argv: []string{"atlas", "crawl"}}, true},252 {"no name", Operation{Argv: []string{"atlas"}}, false},253 {"no command", Operation{Name: "crawl"}, false},254 {"an empty program", Operation{Name: "crawl", Argv: []string{" "}}, false},255 {"a line break in an argument", Operation{Name: "crawl", Argv: []string{"atlas", "a\nb"}}, false},256 }257 for _, c := range cases {258 if err := c.op.Validate(); (err == nil) != c.ok {259 t.Errorf("%s: %v", c.what, err)260 }261 }262}263264func TestValidTarget(t *testing.T) {265 cases := []struct {266 target string267 pair bool268 ok bool269 }{270 {"cyberpunk-2077", false, true},271 {"bend-or", false, true},272 {"mars", false, true},273 {"cyberpunk-2077/night-city", true, true},274 {"", false, false},275 {"-rf", false, false},276 {"not a target", false, false},277 {"Cyberpunk", false, false},278 {"a/b", false, false},279 {"a", true, false},280 {"a/b/c", true, false},281 {"/night-city", true, false},282 {"cyberpunk-2077/", true, false},283 {"a;rm", false, false},284 }285 for _, c := range cases {286 err := ValidTarget(c.target, c.pair)287 if (err == nil) != c.ok {288 t.Errorf("ValidTarget(%q, pair=%v) = %v", c.target, c.pair, err)289 }290 }291}292293func TestCheckOrigin(t *testing.T) {294 cases := []struct {295 origin string296 ok bool297 }{298 {"", true}, // a plain same-origin form submission sends none299 {"http://127.0.0.1:6180", true},300 {"https://127.0.0.1:6180", true},301 {"http://evil.example", false},302 {"null", false},303 {"http://127.0.0.1:6181", false},304 }305 for _, c := range cases {306 request := httptest.NewRequest(http.MethodPost, "http://127.0.0.1:6180/operations/run", nil)307 request.Host = "127.0.0.1:6180"308 if c.origin != "" {309 request.Header.Set("Origin", c.origin)310 }311 err := CheckOrigin(request)312 if (err == nil) != c.ok {313 t.Errorf("Origin %q: %v", c.origin, err)314 }315 if err != nil && !errors.Is(err, ErrForeignOrigin) {316 t.Errorf("Origin %q refused with an unrecognisable error: %v", c.origin, err)317 }318 }319}320321// rows renders one row the way the workbench does, in miniature.322func rows(w io.Writer, row Row) error {323 _, err := fmt.Fprintf(w, "<p data-kind=%q data-seq=%q>%s</p>\n", row.Kind,324 fmt.Sprint(row.Seq), row.Message)325 return err326}327328func site(t *testing.T, runner *Runner, op func(*http.Request) Operation) *httptest.Server {329 t.Helper()330 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {331 runner.Serve(w, r, op(r), rows)332 }))333 t.Cleanup(server.Close)334 return server335}336337func TestServeStreamsRowsAsTheyArrive(t *testing.T) {338 server := site(t, &Runner{}, func(*http.Request) Operation { return helper("crawl", "dawdles") })339340 started := time.Now()341 response, err := server.Client().Post(server.URL, "application/x-www-form-urlencoded", nil)342 if err != nil {343 t.Fatal(err)344 }345 defer response.Body.Close()346 if response.StatusCode != http.StatusOK {347 t.Fatalf("the operation answered %d", response.StatusCode)348 }349 if got := response.Header.Get("Content-Type"); !strings.HasPrefix(got, "text/html") {350 t.Errorf("rows arrived as %q", got)351 }352 if got := response.Header.Get("X-Content-Type-Options"); got != "nosniff" {353 t.Errorf("the stream is sniffable: %q", got)354 }355356 reader := bufio.NewReader(response.Body)357 first, err := reader.ReadString('\n')358 if err != nil {359 t.Fatal(err)360 }361 // The helper waits a second and a half between its two events. A row that362 // arrives before then arrived because it was flushed, not because the363 // operation was over.364 if waited := time.Since(started); waited > time.Second {365 t.Errorf("the first row took %s: the stream is being buffered", waited)366 }367 if !strings.Contains(first, `data-kind="command"`) {368 t.Errorf("the stream does not open with the command: %q", first)369 }370 rest, err := io.ReadAll(reader)371 if err != nil {372 t.Fatal(err)373 }374 body := first + string(rest)375 for _, want := range []string{"started", "finished", `data-kind="result"`, "done"} {376 if !strings.Contains(body, want) {377 t.Errorf("the stream misses %q: %s", want, body)378 }379 }380}381382func TestServeRefusesAForeignOriginASecondOperationAndANonCommand(t *testing.T) {383 runner := &Runner{}384 var operation func(*http.Request) Operation = func(r *http.Request) Operation {385 if r.URL.Query().Get("empty") != "" {386 return Operation{Name: "crawl"}387 }388 return helper("crawl", "waits")389 }390 server := site(t, runner, operation)391392 post := func(url string, header ...string) *http.Response {393 t.Helper()394 request, err := http.NewRequest(http.MethodPost, url, nil)395 if err != nil {396 t.Fatal(err)397 }398 for at := 0; at+1 < len(header); at += 2 {399 request.Header.Set(header[at], header[at+1])400 }401 response, err := server.Client().Do(request)402 if err != nil {403 t.Fatal(err)404 }405 t.Cleanup(func() { response.Body.Close() })406 return response407 }408409 if got := post(server.URL, "Origin", "http://evil.example").StatusCode; got != http.StatusForbidden {410 t.Errorf("a cross-site POST answered %d", got)411 }412 if got := post(server.URL + "?empty=1").StatusCode; got != http.StatusBadRequest {413 t.Errorf("an operation with no command answered %d", got)414 }415416 // A long operation holds the slot; the second submission is refused417 // outright rather than queued behind it.418 release, ok := runner.Acquire("crawl")419 if !ok {420 t.Fatal("the slot was already held")421 }422 response := post(server.URL)423 if response.StatusCode != http.StatusConflict {424 t.Errorf("a second operation answered %d, want 409", response.StatusCode)425 }426 body, _ := io.ReadAll(response.Body)427 if !strings.Contains(string(body), "crawl") {428 t.Errorf("the refusal does not say what is running: %q", body)429 }430 release()431}