Add test for examples/chat server. (ServiceWeaver#367)

Added a way for tests to get the host:port for a listener.

Reduce weavertest time spent waiting for weaver.Main.Main
to exit to make tests run faster.

Run weaver.Main Init before starting test code.

Added examples/chat/server_test.go.

Allow chat server to accept GET requests for new threads and posts
(for testing purposes).

examples/chat/server.go

@@ -254,10 +254,10 @@ func getID(str string) (int64, error) {
 func getImage(r *http.Request) ([]byte, error) {
  img, fhdr, err := r.FormFile("image")
  if err != nil {
- if errors.Is(err, http.ErrMissingFile) {
+ if errors.Is(err, http.ErrMissingFile) || errors.Is(err, http.ErrNotMultipart) {
  return nil, nil
  }
- return nil, fmt.Errorf("bad image form value")
+ return nil, fmt.Errorf("bad image form value %w", err)
  }
 
  // Validate by checking size and parsing the image.

examples/chat/server_test.go

@@ -0,0 +1,131 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http:https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "net/http"
+ "regexp"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/ServiceWeaver/weaver"
+ "github.com/ServiceWeaver/weaver/weavertest"
+)
+
+// db is a fake store for the chat database.
+type db struct {
+ mu sync.Mutex
+ lastPost PostID
+ threads []Thread
+}
+
+var _ SQLStore = &db{}
+
+func (db *db) CreateThread(ctx context.Context, creator string, when time.Time, others []string, text string, image []byte) (ThreadID, error) {
+ db.mu.Lock()
+ defer db.mu.Unlock()
+ db.lastPost++
+ t := Thread{
+ ID: ThreadID(len(db.threads)) + 1,
+ Posts: []Post{{ID: db.lastPost, Creator: creator, When: when, Text: text}}, // image is ignored
+ }
+ db.threads = append(db.threads, t)
+ return t.ID, nil
+}
+
+func (db *db) CreatePost(ctx context.Context, creator string, when time.Time, thread ThreadID, text string) error {
+ db.mu.Lock()
+ defer db.mu.Unlock()
+ db.lastPost++
+ t := db.threads[int(thread-1)]
+ t.Posts = append(t.Posts, Post{ID: db.lastPost, Creator: creator, When: when, Text: text})
+ db.threads[int(thread-1)] = t
+ return nil
+}
+
+func (db *db) GetFeed(ctx context.Context, user string) ([]Thread, error) {
+ db.mu.Lock()
+ defer db.mu.Unlock()
+
+ // Deep copy results
+ threads := make([]Thread, len(db.threads))
+ for i, t := range db.threads {
+ t.Posts = append([]Post(nil), t.Posts...)
+ threads[i] = t
+ }
+ return threads, nil
+}
+
+func (db *db) GetImage(ctx context.Context, _ string, image ImageID) ([]byte, error) {
+ return nil, fmt.Errorf("no images")
+}
+
+func TestServer(t *testing.T) {
+ for _, r := range weavertest.AllRunners() {
+ // Use a fake DB since normal implementation does not work with multiple replicas
+ db := &db{}
+ r.Fakes = []weavertest.FakeComponent{weavertest.Fake[SQLStore](db)}
+ r.Test(t, func(t *testing.T, store weaver.Main) {
+ addr, err := weavertest.ListenerAddress(t, "chat")
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // Login
+ expect(t, "login", httpGet(t, addr, "/"), "Login")
+ expect(t, "feed", httpGet(t, addr, "/?name=user"), "/newthread")
+
+ // Create a new thread
+ r := httpGet(t, addr, "/newthread?name=user&recipients=bar,baz&message=Hello&image=")
+ expect(t, "new thread", r, "Hello")
+ m := regexp.MustCompile(`id="tid(\d+)"`).FindStringSubmatch(r)
+ if m == nil {
+ t.Fatalf("no tid in /newthread respinse:\n%s\n", r)
+ }
+ tid := m[1]
+
+ // Reply to thread
+ r = httpGet(t, addr, fmt.Sprintf("/newpost?name=user&tid=%s&post=Hi!", tid))
+ expect(t, "reply", r, `Hello`)
+ expect(t, "reply", r, `Hi!`)
+ })
+ }
+}
+
+func httpGet(t *testing.T, addr, path string) string {
+ t.Helper()
+ response, err := http.Get("http:https://" + addr + path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer response.Body.Close()
+ result, err := io.ReadAll(response.Body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return string(result)
+}
+
+func expect(t *testing.T, title, subject, re string) {
+ t.Helper()
+ r := regexp.MustCompile(re)
+ if !r.MatchString(subject) {
+ t.Fatalf("did not find %q in %s:\n%s\n", re, title, subject)
+ }
+}

internal/private/private.go

@@ -42,4 +42,9 @@ type App interface {
 
  // Get fetches the component with type t from wlet.
  Get(requester string, t reflect.Type) (any, error)
+
+ // ListenerAddress returns the address (host:port) of the
+ // named listener, waiting for the listener to be created
+ // if necessary.
+ ListenerAddress(name string) (string, error)
 }

weavelet.go

@@ -24,6 +24,7 @@ import (
  "os"
  "reflect"
  "strings"
+ "sync"
  "time"
 
  "github.com/ServiceWeaver/weaver/internal/config"
@@ -63,6 +64,14 @@ type weavelet struct {
 
  componentsByName map[string]*component // component name -> component
  componentsByType map[reflect.Type]*component // component type -> component
+
+ listenersMu sync.Mutex
+ listeners map[string]*listenerState
+}
+
+type listenerState struct {
+ addr string
+ initialized chan struct{} // Closed when addr has been filled
 }
 
 type transport struct {
@@ -228,18 +237,32 @@ func (w *weavelet) start() error {
  }
 
  w.logRolodexCard()
+
+ // Make sure Main is initialized if local.
+ if _, err := w.getMainIfLocal(); err != nil {
+ return err
+ }
  return nil
 }
 
-func (w *weavelet) Wait(ctx context.Context) error {
+// getMainIfLocal returns the weaver.Main implementation if hosted in
+// this weavelet, or nil if weaver.Main is remote.
+func (w *weavelet) getMainIfLocal() (*componentImpl, error) {
  // Note that a weavertest may have RunMain set to true, but no main
  // component registered.
  if m, ok := w.componentsByType[reflect.TypeOf((*Main)(nil)).Elem()]; ok && w.info.RunMain {
- // This process is hosting weaver.Main, so call its Main() method.
- impl, err := w.getImpl(ctx, m)
- if err != nil {
- return err
- }
+ return w.getImpl(w.ctx, m)
+ }
+ return nil, nil
+}
+
+func (w *weavelet) Wait(ctx context.Context) error {
+ // Call weaver.Main.Main if weaver.Main is hosted locally.
+ impl, err := w.getMainIfLocal()
+ if err != nil {
+ return nil
+ }
+ if impl != nil {
  return impl.impl.(Main).Main(ctx)
  }
 
@@ -367,6 +390,13 @@ func (w *weavelet) getListener(name string, opts ListenerOptions) (*Listener, er
  if reply.Error != "" {
  return nil, fmt.Errorf("getListener(%q): %s", name, reply.Error)
  }
+
+ w.listenersMu.Lock()
+ defer w.listenersMu.Unlock()
+ ls := w.getListenerState(name)
+ ls.addr = l.Addr().String()
+ close(ls.initialized) // Mark as initialized
+
  return &Listener{Listener: l, proxyAddr: reply.ProxyAddress}, nil
 }
 
@@ -395,6 +425,29 @@ func (w *weavelet) addHandlers(handlers *call.HandlerMap, c *component) {
  }
 }
 
+func (w *weavelet) ListenerAddress(name string) (string, error) {
+ w.listenersMu.Lock()
+ ls := w.getListenerState(name)
+ w.listenersMu.Unlock()
+
+ <-ls.initialized // Wait until initialized
+ return ls.addr, nil
+}
+
+// REQUIRES: w.listenersMu is held
+func (w *weavelet) getListenerState(name string) *listenerState {
+ l := w.listeners[name]
+ if l != nil {
+ return l
+ }
+ l = &listenerState{initialized: make(chan struct{})}
+ if w.listeners == nil {
+ w.listeners = map[string]*listenerState{}
+ }
+ w.listeners[name] = l
+ return l
+}
+
 // GetLoad implements the WeaveletHandler interface.
 func (w *weavelet) GetLoad(*protos.GetLoadRequest) (*protos.GetLoadReply, error) {
  report := &protos.LoadReport{

weavertest/init.go

@@ -21,6 +21,7 @@ import (
  "os"
  "reflect"
  "runtime"
+ "sync"
  "testing"
  "time"
 
@@ -88,6 +89,44 @@ func Fake[T any](impl any) FakeComponent {
  return FakeComponent{intf: reflect.TypeOf((*T)(nil)).Elem(), impl: impl}
 }
 
+// private.App object per live test or benchmark.
+var (
+ appMu sync.Mutex
+ apps map[testing.TB]private.App
+)
+
+func registerApp(t testing.TB, app private.App) {
+ appMu.Lock()
+ defer appMu.Unlock()
+ if apps == nil {
+ apps = map[testing.TB]private.App{}
+ }
+ apps[t] = app
+}
+
+func unregisterApp(t testing.TB, app private.App) {
+ appMu.Lock()
+ defer appMu.Unlock()
+ delete(apps, t)
+}
+
+func getApp(t testing.TB) private.App {
+ appMu.Lock()
+ defer appMu.Unlock()
+ return apps[t]
+}
+
+// ListenerAddress returns the address (of the form host:port) for the
+// listener with the specified name. This call will block waiting for
+// the listener to be initialized if necessary.
+func ListenerAddress(t testing.TB, name string) (string, error) {
+ app := getApp(t)
+ if app == nil {
+ return "", fmt.Errorf("Service Weaver application is not running")
+ }
+ return app.ListenerAddress(name)
+}
+
 //go:generate ../cmd/weaver/weaver generate
 
 // testMain is the component implementation used in tests.
@@ -240,6 +279,8 @@ func runWeaver(ctx context.Context, t testing.TB, runner Runner, body func(conte
  if err != nil {
  return err
  }
+ registerApp(t, app)
+ t.Cleanup(func() { unregisterApp(t, app) })
 
  // Run wait() in a go routine.
  sub, cancel := context.WithCancel(ctx)
@@ -259,7 +300,7 @@ func runWeaver(ctx context.Context, t testing.TB, runner Runner, body func(conte
  if err != nil && !errors.Is(err, context.Canceled) {
  t.Fatalf("weaver.Main.Main failure: %v", err)
  }
- case <-time.After(time.Second):
+ case <-time.After(time.Millisecond * 100):
  t.Log("weaver.Main.Main not exiting after cancellation")
  }
  return nil