Initial commit: DSH Web Manager

internal/api/config_body_test.go

@@ -0,0 +1,205 @@
+package api
+
+import (
+	"bytes"
+	"io"
+	"mime/multipart"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.misaka.ren/m1saka/dsh-web/internal/store"
+)
+
+type configPausedBody struct {
+	io.Reader
+	started chan struct{}
+	release chan struct{}
+	once    sync.Once
+}
+
+func (b *configPausedBody) Read(p []byte) (int, error) {
+	b.once.Do(func() { close(b.started) })
+	<-b.release
+	return b.Reader.Read(p)
+}
+
+func TestConfigSlowBodyRevalidatesWithoutLocks(t *testing.T) {
+	for _, action := range []string{"profile", "backups", "restore", "multipart"} {
+		for _, change := range []string{"lease", "deleted"} {
+			t.Run(action+"/"+change, func(t *testing.T) {
+				a, i, _ := configTestAPI(t, false)
+				a.workspaceReaders = make(map[string]int)
+				body := &configPausedBody{Reader: strings.NewReader(`{}`), started: make(chan struct{}), release: make(chan struct{})}
+				r := httptest.NewRequest("POST", "/", body)
+				if action == "profile" {
+					r.Method = "PUT"
+				}
+				target := action
+				if action == "multipart" {
+					target = "backups"
+					var buf bytes.Buffer
+					form := multipart.NewWriter(&buf)
+					part, _ := form.CreateFormFile("file", "test.dshbak")
+					part.Write(bytes.Repeat([]byte("x"), 2<<20))
+					form.Close()
+					body.Reader = &buf
+					r.Header.Set("Content-Type", form.FormDataContentType())
+					t.Setenv("TMPDIR", t.TempDir())
+				}
+				w := httptest.NewRecorder()
+				done := make(chan struct{})
+				go func() { defer close(done); a.config(w, r, i, target) }()
+				defer func() {
+					select {
+					case <-body.release:
+					default:
+						close(body.release)
+					}
+					<-done
+				}()
+				select {
+				case <-body.started:
+				case <-time.After(3 * time.Second):
+					t.Fatal("body read did not start")
+				}
+				if !a.mu.TryLock() {
+					t.Fatal("slow body holds API lock")
+				}
+				if change == "lease" {
+					a.workspaceReaders[i.ID]++
+				}
+				a.mu.Unlock()
+				if !workspaceMu.TryLock() {
+					t.Fatal("slow body holds workspace lock")
+				}
+				workspaceMu.Unlock()
+				want := 409
+				if change == "deleted" {
+					if _, err := a.s.DB.Exec("DELETE FROM instances WHERE id=?", i.ID); err != nil {
+						t.Fatal(err)
+					}
+					want = 404
+				}
+				close(body.release)
+				select {
+				case <-done:
+				case <-time.After(3 * time.Second):
+					t.Fatal("config did not finish")
+				}
+				if w.Code != want {
+					t.Fatalf("status %d: %s", w.Code, w.Body.String())
+				}
+				if action == "multipart" {
+					entries, err := os.ReadDir(os.Getenv("TMPDIR"))
+					if err != nil || len(entries) != 0 {
+						t.Fatalf("multipart temp files remain: %v %v", entries, err)
+					}
+				}
+			})
+		}
+	}
+}
+
+type configDeadlineWriter struct {
+	*httptest.ResponseRecorder
+	deadlines []time.Time
+}
+
+func (w *configDeadlineWriter) SetReadDeadline(deadline time.Time) error {
+	w.deadlines = append(w.deadlines, deadline)
+	return nil
+}
+
+type configErrorBody struct{}
+
+func (configErrorBody) Read([]byte) (int, error) { return 0, os.ErrDeadlineExceeded }
+
+func TestConfigBodyErrorsReturnWithoutLocks(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		body io.Reader
+		want int
+	}{
+		{"oversized", strings.NewReader(strings.Repeat("x", 4097)), 413},
+		{"timeout", configErrorBody{}, 408},
+		{"multipart", strings.NewReader("bad multipart"), 400},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			a := &API{}
+			r := httptest.NewRequest("POST", "/", tc.body)
+			r.ContentLength = -1
+			if tc.name == "multipart" {
+				r.Header.Set("Content-Type", "multipart/form-data; boundary=test")
+			}
+			w := &configDeadlineWriter{ResponseRecorder: httptest.NewRecorder()}
+			a.mu.Lock()
+			workspaceMu.Lock()
+			done := make(chan struct{})
+			go func() { defer close(done); a.config(w, r, store.Instance{}, "backups") }()
+			select {
+			case <-done:
+				workspaceMu.Unlock()
+				a.mu.Unlock()
+			case <-time.After(3 * time.Second):
+				workspaceMu.Unlock()
+				a.mu.Unlock()
+				t.Fatal("body error waited for locks")
+			}
+			if w.Code != tc.want {
+				t.Fatalf("status %d: %s", w.Code, w.Body.String())
+			}
+			if len(w.deadlines) != 2 || w.deadlines[0].IsZero() || !w.deadlines[1].IsZero() {
+				t.Fatalf("deadlines: %v", w.deadlines)
+			}
+		})
+	}
+}
+
+func TestConfigBodySlotsRejectWithoutReading(t *testing.T) {
+	for n := 0; n < cap(configBodySlots); n++ {
+		configBodySlots <- struct{}{}
+	}
+	defer func() {
+		for n := 0; n < cap(configBodySlots); n++ {
+			<-configBodySlots
+		}
+	}()
+	w := httptest.NewRecorder()
+	(&API{}).config(w, httptest.NewRequest("POST", "/", configErrorBody{}), store.Instance{}, "backups")
+	if w.Code != 503 {
+		t.Fatalf("status %d", w.Code)
+	}
+}
+
+func TestConfigSlowResponseReleasesLocks(t *testing.T) {
+	for _, tc := range []struct{ method, action string }{{"POST", "restore"}, {"GET", "profile"}, {"GET", "backups"}} {
+		t.Run(tc.method+"/"+tc.action, func(t *testing.T) {
+			a, i, _ := configTestAPI(t, false)
+			w := &pausedDownload{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), release: make(chan struct{})}
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				a.config(w, httptest.NewRequest(tc.method, "/", strings.NewReader(`{}`)), i, tc.action)
+			}()
+			defer func() { close(w.release); <-done }()
+			select {
+			case <-w.started:
+			case <-time.After(3 * time.Second):
+				t.Fatal("response write did not start")
+			}
+			if !a.mu.TryLock() {
+				t.Fatal("slow response holds API lock")
+			}
+			a.mu.Unlock()
+			if !workspaceMu.TryLock() {
+				t.Fatal("slow response holds workspace lock")
+			}
+			workspaceMu.Unlock()
+
+		})
+	}
+}