Add UserAPI util tests (#2907)

This adds some `userapi/util` tests.

test/testrig/base.go

@@ -108,7 +108,7 @@ func Base(cfg *config.Dendrite) (*base.BaseDendrite, nats.JetStreamContext, *nat
 	cfg.Global.JetStream.InMemory = true
 	cfg.SyncAPI.Fulltext.InMemory = true
 	cfg.FederationAPI.KeyPerspectives = nil
-	base := base.NewBaseDendrite(cfg, "Tests")
+	base := base.NewBaseDendrite(cfg, "Tests", base.DisableMetrics)
 	js, jc := base.NATS.Prepare(base.ProcessContext, &cfg.Global.JetStream)
 	return base, js, jc
 }

userapi/util/notify_test.go

@@ -0,0 +1,119 @@
+package util_test
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+	"golang.org/x/crypto/bcrypt"
+
+	"github.com/matrix-org/dendrite/internal/pushgateway"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/dendrite/test/testrig"
+	"github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/dendrite/userapi/storage"
+	userUtil "github.com/matrix-org/dendrite/userapi/util"
+)
+
+func TestNotifyUserCountsAsync(t *testing.T) {
+	alice := test.NewUser(t)
+	aliceLocalpart, serverName, err := gomatrixserverlib.SplitID('@', alice.ID)
+	if err != nil {
+		t.Error(err)
+	}
+	ctx := context.Background()
+
+	// Create a test room, just used to provide events
+	room := test.NewRoom(t, alice)
+	dummyEvent := room.Events()[len(room.Events())-1]
+
+	appID := util.RandomString(8)
+	pushKey := util.RandomString(8)
+
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		receivedRequest := make(chan bool, 1)
+		// create a test server which responds to our /notify call
+		srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var data pushgateway.NotifyRequest
+			if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
+				t.Error(err)
+			}
+			notification := data.Notification
+			// Validate the request
+			if notification.Counts == nil {
+				t.Fatal("no unread notification counts in request")
+			}
+			if unread := notification.Counts.Unread; unread != 1 {
+				t.Errorf("expected one unread notification, got %d", unread)
+			}
+
+			if len(notification.Devices) == 0 {
+				t.Fatal("expected devices in request")
+			}
+
+			// We only created one push device, so access it directly
+			device := notification.Devices[0]
+			if device.AppID != appID {
+				t.Errorf("unexpected app_id: %s, want %s", device.AppID, appID)
+			}
+			if device.PushKey != pushKey {
+				t.Errorf("unexpected push_key: %s, want %s", device.PushKey, pushKey)
+			}
+
+			// Return empty result, otherwise the call is handled as failed
+			if _, err := w.Write([]byte("{}")); err != nil {
+				t.Error(err)
+			}
+			close(receivedRequest)
+		}))
+		defer srv.Close()
+
+		// Create DB and Dendrite base
+		connStr, close := test.PrepareDBConnectionString(t, dbType)
+		defer close()
+		base, _, _ := testrig.Base(nil)
+		defer base.Close()
+		db, err := storage.NewUserAPIDatabase(base, &config.DatabaseOptions{
+			ConnectionString: config.DataSource(connStr),
+		}, "test", bcrypt.MinCost, 0, 0, "")
+		if err != nil {
+			t.Error(err)
+		}
+
+		// Prepare pusher with our test server URL
+		if err := db.UpsertPusher(ctx, api.Pusher{
+			Kind:    api.HTTPKind,
+			AppID:   appID,
+			PushKey: pushKey,
+			Data: map[string]interface{}{
+				"url": srv.URL,
+			},
+		}, aliceLocalpart, serverName); err != nil {
+			t.Error(err)
+		}
+
+		// Insert a dummy event
+		if err := db.InsertNotification(ctx, aliceLocalpart, serverName, dummyEvent.EventID(), 0, nil, &api.Notification{
+			Event: gomatrixserverlib.HeaderedToClientEvent(dummyEvent, gomatrixserverlib.FormatAll),
+		}); err != nil {
+			t.Error(err)
+		}
+
+		// Notify the user about a new notification
+		if err := userUtil.NotifyUserCountsAsync(ctx, pushgateway.NewHTTPClient(true), aliceLocalpart, serverName, db); err != nil {
+			t.Error(err)
+		}
+		select {
+		case <-time.After(time.Second * 5):
+			t.Error("timed out waiting for response")
+		case <-receivedRequest:
+		}
+	})
+
+}

userapi/util/phonehomestats.go

@@ -97,12 +97,10 @@ func (p *phoneHomeStats) collect() {
 
 	// configuration information
 	p.stats["federation_disabled"] = p.cfg.Global.DisableFederation
-	p.stats["nats_embedded"] = true
+	natsEmbedded := len(p.cfg.Global.JetStream.Addresses) == 0
-	p.stats["nats_in_memory"] = p.cfg.Global.JetStream.InMemory
+	p.stats["nats_embedded"] = natsEmbedded
-	if len(p.cfg.Global.JetStream.Addresses) > 0 {
+	p.stats["nats_in_memory"] = p.cfg.Global.JetStream.InMemory && natsEmbedded
-		p.stats["nats_embedded"] = false
+
-		p.stats["nats_in_memory"] = false // probably
-	}
 	if len(p.cfg.Logging) > 0 {
 		p.stats["log_level"] = p.cfg.Logging[0].Level
 	} else {

userapi/util/phonehomestats_test.go

@@ -0,0 +1,84 @@
+package util
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"golang.org/x/crypto/bcrypt"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/dendrite/test/testrig"
+	"github.com/matrix-org/dendrite/userapi/storage"
+)
+
+func TestCollect(t *testing.T) {
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		b, _, _ := testrig.Base(nil)
+		connStr, closeDB := test.PrepareDBConnectionString(t, dbType)
+		defer closeDB()
+		db, err := storage.NewUserAPIDatabase(b, &config.DatabaseOptions{
+			ConnectionString: config.DataSource(connStr),
+		}, "localhost", bcrypt.MinCost, 1000, 1000, "")
+		if err != nil {
+			t.Error(err)
+		}
+
+		receivedRequest := make(chan struct{}, 1)
+		// create a test server which responds to our call
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var data map[string]interface{}
+			if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
+				t.Error(err)
+			}
+			defer r.Body.Close()
+			if _, err := w.Write([]byte("{}")); err != nil {
+				t.Error(err)
+			}
+
+			// verify the received data matches our expectations
+			dbEngine, ok := data["database_engine"]
+			if !ok {
+				t.Errorf("missing database_engine in JSON request: %+v", data)
+			}
+			version, ok := data["version"]
+			if !ok {
+				t.Errorf("missing version in JSON request: %+v", data)
+			}
+			if version != internal.VersionString() {
+				t.Errorf("unexpected version: %q, expected %q", version, internal.VersionString())
+			}
+			switch {
+			case dbType == test.DBTypeSQLite && dbEngine != "SQLite":
+				t.Errorf("unexpected database_engine: %s", dbEngine)
+			case dbType == test.DBTypePostgres && dbEngine != "Postgres":
+				t.Errorf("unexpected database_engine: %s", dbEngine)
+			}
+			close(receivedRequest)
+		}))
+		defer srv.Close()
+
+		b.Cfg.Global.ReportStats.Endpoint = srv.URL
+		stats := phoneHomeStats{
+			prevData:   timestampToRUUsage{},
+			serverName: "localhost",
+			startTime:  time.Now(),
+			cfg:        b.Cfg,
+			db:         db,
+			isMonolith: false,
+			client:     &http.Client{Timeout: time.Second},
+		}
+
+		stats.collect()
+
+		select {
+		case <-time.After(time.Second * 5):
+			t.Error("timed out waiting for response")
+		case <-receivedRequest:
+		}
+	})
+}