package app_test import ( "bufio" "context" "errors" "io" "log" "net" "net/http" "strings" "sync/atomic" "testing" "time" omni "git.yuntuo.hk/k2safe/go-omni" ) func TestHTTPConfiguredStandardServer(t *testing.T) { cfg := omni.HTTPConfig{ReadHeaderTimeout: time.Second, ReadTimeout: 2 * time.Second, WriteTimeout: 3 * time.Second, IdleTimeout: 4 * time.Second, MaxHeaderBytes: 8192, ErrorLog: log.New(io.Discard, "", 0)} a := omni.New(omni.WithHTTPConfig(cfg)) s := a.NewServer(":9000") if s.ReadHeaderTimeout != cfg.ReadHeaderTimeout || s.ReadTimeout != cfg.ReadTimeout || s.WriteTimeout != cfg.WriteTimeout || s.IdleTimeout != cfg.IdleTimeout || s.MaxHeaderBytes != cfg.MaxHeaderBytes || s.ErrorLog != cfg.ErrorLog || s.Addr != ":9000" { t.Fatalf("server options: %+v", s) } s = omni.New().NewServer("") if s.ReadHeaderTimeout != 0 || s.ReadTimeout != 0 || s.WriteTimeout != 0 || s.IdleTimeout != 0 || s.MaxHeaderBytes != 0 || s.ErrorLog != nil { t.Fatal("default behavior changed") } } func TestHTTPInvalidStartupReleasesResources(t *testing.T) { a := omni.New(omni.WithHTTPConfig(omni.HTTPConfig{ReadTimeout: -time.Second})) called := 0 cleanupErr := errors.New("cleanup") a.OnShutdown(func() error { called++; return cleanupErr }) err := a.RunContext(context.Background(), "bad address") if err == nil || !strings.Contains(err.Error(), "http.readTimeout") || !errors.Is(err, cleanupErr) || called != 1 { t.Fatalf("validation/cleanup: %v, %d", err, called) } defer func() { if recover() == nil { t.Error("NewServer accepted invalid config") } }() a.NewServer("") } type httpConfigListener struct{ net.Listener } func (l httpConfigListener) Accept() (net.Conn, error) { c, err := l.Listener.Accept() if err == nil { if tcp, ok := c.(*net.TCPConn); ok { _ = tcp.SetWriteBuffer(32 << 10) } } return c, err } func httpConfigConnection(t *testing.T, transport string, cfg omni.HTTPConfig, handler omni.Handler) net.Conn { t.Helper() a := omni.New(omni.WithTransport(transport), omni.WithHTTPConfig(cfg), omni.WithShutdownTimeout(time.Second)) a.GET("/", handler) a.POST("/", handler) l, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- a.ServeContext(ctx, httpConfigListener{l}) }() t.Cleanup(func() { cancel() select { case err := <-done: if err != nil { t.Errorf("server shutdown: %v", err) } case <-time.After(3 * time.Second): t.Error("server did not stop") } }) c, err := net.DialTimeout("tcp", l.Addr().String(), time.Second) if err != nil { t.Fatal(err) } t.Cleanup(func() { c.Close() }) _ = c.SetDeadline(time.Now().Add(3 * time.Second)) return c } func TestHTTPConfiguredNetworkLimits(t *testing.T) { for _, transport := range []string{omni.TransportNetHTTP, omni.TransportNative} { t.Run(transport, func(t *testing.T) { empty := func(c *omni.Context) error { w := c.Raw(); w.WriteHeader(204); return nil } t.Run("slowHeaders", func(t *testing.T) { c := httpConfigConnection(t, transport, omni.HTTPConfig{ReadHeaderTimeout: 60 * time.Millisecond}, empty) if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: localhost\r\nX-Incomplete:"); err != nil { t.Fatal(err) } _, err := io.ReadAll(c) if e, ok := err.(net.Error); ok && e.Timeout() { t.Fatal("server read-header timeout not applied") } }) t.Run("slowBody", func(t *testing.T) { var accepted atomic.Bool c := httpConfigConnection(t, transport, omni.HTTPConfig{ReadTimeout: 60 * time.Millisecond}, func(c *omni.Context) error { _, err := io.ReadAll(c.Request().Body) accepted.Store(err == nil) return nil }) if _, err := io.WriteString(c, "POST / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nContent-Length: 10\r\n\r\nx"); err != nil { t.Fatal(err) } _, err := io.ReadAll(c) if e, ok := err.(net.Error); ok && e.Timeout() { t.Fatal("read timeout not applied") } if accepted.Load() { t.Fatal("incomplete body accepted") } }) t.Run("idle", func(t *testing.T) { c := httpConfigConnection(t, transport, omni.HTTPConfig{IdleTimeout: 60 * time.Millisecond}, empty) if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); err != nil { t.Fatal(err) } r, err := http.ReadResponse(bufio.NewReader(c), nil) if err != nil { t.Fatal(err) } r.Body.Close() if r.StatusCode != 204 { t.Fatal(r.StatusCode) } _, err = c.Read(make([]byte, 1)) if e, ok := err.(net.Error); ok && e.Timeout() { t.Fatal("idle timeout not applied") } if err == nil { t.Fatal("idle connection not closed") } }) t.Run("writeBackpressure", func(t *testing.T) { done := make(chan error, 1) c := httpConfigConnection(t, transport, omni.HTTPConfig{WriteTimeout: 60 * time.Millisecond}, func(c *omni.Context) error { w := c.Raw() w.WriteHeader(200) if err := http.NewResponseController(w).Flush(); err != nil { done <- err return nil } _, err := io.Copy(w, io.LimitReader(repeatingHTTPReader{}, 64<<20)) done <- err return nil }) if tcp, ok := c.(*net.TCPConn); ok { _ = tcp.SetReadBuffer(1024) } if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); err != nil { t.Fatal(err) } select { case err := <-done: if err == nil { t.Fatal("write unexpectedly completed") } case <-time.After(2 * time.Second): t.Fatal("write timeout not applied") } }) t.Run("headerLimit", func(t *testing.T) { c := httpConfigConnection(t, transport, omni.HTTPConfig{MaxHeaderBytes: 1024}, empty) if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: localhost\r\nX-Large: "+strings.Repeat("a", 8192)+"\r\n\r\n"); err != nil { t.Fatal(err) } r, err := http.ReadResponse(bufio.NewReader(c), nil) if err != nil { t.Fatal(err) } r.Body.Close() if r.StatusCode != 431 { t.Fatalf("oversized header status: %d", r.StatusCode) } }) }) } } type repeatingHTTPReader struct{} func (repeatingHTTPReader) Read(b []byte) (int, error) { for i := range b { b[i] = 'x' } return len(b), nil }