Add util tests

userapi/util/notify_test.go

@@ -0,0 +1,89 @@
+package util_test
+
+import (
+	"context"
+	"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]
+
+	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) {
+			// Return empty result, otherwise the call is handled as failed
+			if _, err := w.Write([]byte("{}")); err != nil {
+				t.Error(err)
+			}
+			receivedRequest <- true
+		}))
+		defer srv.Close()
+
+		// Create DB and Dendrite base
+		connStr, close := test.PrepareDBConnectionString(t, dbType)
+		defer close()
+		base, _, _ := testrig.Base(nil)
+		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:   util.RandomString(8),
+			PushKey: util.RandomString(8),
+			Data: map[string]interface{}{
+				"url":      srv.URL,
+				"event_id": dummyEvent.EventID(),
+			},
+		}, 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

@@ -63,7 +63,9 @@ func StartPhoneHomeCollector(startTime time.Time, cfg *config.Dendrite, statsDB
 	}
 
 	// start initial run after 5min
-	time.AfterFunc(time.Minute*5, p.collect)
+	time.AfterFunc(time.Minute*5, func() {
+		p.collect()
+	})
 
 	// run every 3 hours
 	ticker := time.NewTicker(time.Hour * 3)
@@ -97,12 +99,10 @@ func (p *phoneHomeStats) collect() {
 
 	// configuration information
 	p.stats["federation_disabled"] = p.cfg.Global.DisableFederation
-	p.stats["nats_embedded"] = true
-	p.stats["nats_in_memory"] = p.cfg.Global.JetStream.InMemory
-	if len(p.cfg.Global.JetStream.Addresses) > 0 {
-		p.stats["nats_embedded"] = false
-		p.stats["nats_in_memory"] = false // probably
-	}
+	natsEmbedded := len(p.cfg.Global.JetStream.Addresses) == 0
+	p.stats["nats_embedded"] = natsEmbedded
+	p.stats["nats_in_memory"] = p.cfg.Global.JetStream.InMemory && natsEmbedded
+
 	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 database_engine 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.Client()
+
+		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.Errorf("timed out waitigf for response")
+		case <-receivedRequest:
+		}
+	})
+}