dendrite/syncapi/syncapi_test.go
kegsay 72285b2659
refactor: update GMSL (#3058)
Sister PR to https://github.com/matrix-org/gomatrixserverlib/pull/364

Read this commit by commit to avoid going insane.
2023-04-19 15:50:33 +01:00

1249 lines
41 KiB
Go

package syncapi
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"testing"
"time"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/gomatrixserverlib/spec"
"github.com/nats-io/nats.go"
"github.com/tidwall/gjson"
"github.com/matrix-org/dendrite/syncapi/routing"
"github.com/matrix-org/dendrite/syncapi/storage"
"github.com/matrix-org/dendrite/syncapi/synctypes"
"github.com/matrix-org/dendrite/clientapi/producers"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
userapi "github.com/matrix-org/dendrite/userapi/api"
)
type syncRoomserverAPI struct {
rsapi.SyncRoomserverAPI
rooms []*test.Room
}
func (s *syncRoomserverAPI) QueryLatestEventsAndState(ctx context.Context, req *rsapi.QueryLatestEventsAndStateRequest, res *rsapi.QueryLatestEventsAndStateResponse) error {
var room *test.Room
for _, r := range s.rooms {
if r.ID == req.RoomID {
room = r
break
}
}
if room == nil {
res.RoomExists = false
return nil
}
res.RoomVersion = room.Version
return nil // TODO: return state
}
func (s *syncRoomserverAPI) QuerySharedUsers(ctx context.Context, req *rsapi.QuerySharedUsersRequest, res *rsapi.QuerySharedUsersResponse) error {
res.UserIDsToCount = make(map[string]int)
return nil
}
func (s *syncRoomserverAPI) QueryBulkStateContent(ctx context.Context, req *rsapi.QueryBulkStateContentRequest, res *rsapi.QueryBulkStateContentResponse) error {
return nil
}
func (s *syncRoomserverAPI) QueryMembershipForUser(ctx context.Context, req *rsapi.QueryMembershipForUserRequest, res *rsapi.QueryMembershipForUserResponse) error {
res.IsRoomForgotten = false
res.RoomExists = true
return nil
}
func (s *syncRoomserverAPI) QueryMembershipAtEvent(ctx context.Context, req *rsapi.QueryMembershipAtEventRequest, res *rsapi.QueryMembershipAtEventResponse) error {
return nil
}
type syncUserAPI struct {
userapi.SyncUserAPI
accounts []userapi.Device
}
func (s *syncUserAPI) QueryAccessToken(ctx context.Context, req *userapi.QueryAccessTokenRequest, res *userapi.QueryAccessTokenResponse) error {
for _, acc := range s.accounts {
if acc.AccessToken == req.AccessToken {
res.Device = &acc
return nil
}
}
res.Err = "unknown user"
return nil
}
func (s *syncUserAPI) QueryKeyChanges(ctx context.Context, req *userapi.QueryKeyChangesRequest, res *userapi.QueryKeyChangesResponse) error {
return nil
}
func (s *syncUserAPI) QueryOneTimeKeys(ctx context.Context, req *userapi.QueryOneTimeKeysRequest, res *userapi.QueryOneTimeKeysResponse) error {
return nil
}
func (s *syncUserAPI) PerformLastSeenUpdate(ctx context.Context, req *userapi.PerformLastSeenUpdateRequest, res *userapi.PerformLastSeenUpdateResponse) error {
return nil
}
func TestSyncAPIAccessTokens(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
testSyncAccessTokens(t, dbType)
})
}
func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
user := test.NewUser(t)
room := test.NewRoom(t, user)
alice := userapi.Device{
ID: "ALICEID",
UserID: user.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "Alice",
AccountType: userapi.AccountTypeUser,
}
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
natsInstance := jetstream.NATSInstance{}
defer close()
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
msgs := toNATSMsgs(t, cfg, room.Events()...)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, caching.DisableMetrics)
testrig.MustPublishMsgs(t, jsctx, msgs...)
testCases := []struct {
name string
req *http.Request
wantCode int
wantJoinedRooms []string
}{
{
name: "missing access token",
req: test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"timeout": "0",
})),
wantCode: 401,
},
{
name: "unknown access token",
req: test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": "foo",
"timeout": "0",
})),
wantCode: 401,
},
{
name: "valid access token",
req: test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
})),
wantCode: 200,
wantJoinedRooms: []string{room.ID},
},
}
syncUntil(t, routers, alice.AccessToken, false, func(syncBody string) bool {
// wait for the last sent eventID to come down sync
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(event_id=="%s")`, room.ID, room.Events()[len(room.Events())-1].EventID())
return gjson.Get(syncBody, path).Exists()
})
for _, tc := range testCases {
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, tc.req)
if w.Code != tc.wantCode {
t.Fatalf("%s: got HTTP %d want %d", tc.name, w.Code, tc.wantCode)
}
if tc.wantJoinedRooms != nil {
var res types.Response
if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
t.Fatalf("%s: failed to decode response body: %s", tc.name, err)
}
if len(res.Rooms.Join) != len(tc.wantJoinedRooms) {
t.Errorf("%s: got %v joined rooms, want %v.\nResponse: %+v", tc.name, len(res.Rooms.Join), len(tc.wantJoinedRooms), res)
}
t.Logf("res: %+v", res.Rooms.Join[room.ID])
gotEventIDs := make([]string, len(res.Rooms.Join[room.ID].Timeline.Events))
for i, ev := range res.Rooms.Join[room.ID].Timeline.Events {
gotEventIDs[i] = ev.EventID
}
test.AssertEventIDsEqual(t, gotEventIDs, room.Events())
}
}
}
// Tests what happens when we create a room and then /sync before all events from /createRoom have
// been sent to the syncapi
func TestSyncAPICreateRoomSyncEarly(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
testSyncAPICreateRoomSyncEarly(t, dbType)
})
}
func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
t.Skip("Skipped, possibly fixed")
user := test.NewUser(t)
room := test.NewRoom(t, user)
alice := userapi.Device{
ID: "ALICEID",
UserID: user.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "Alice",
AccountType: userapi.AccountTypeUser,
}
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// order is:
// m.room.create
// m.room.member
// m.room.power_levels
// m.room.join_rules
// m.room.history_visibility
msgs := toNATSMsgs(t, cfg, room.Events()...)
sinceTokens := make([]string, len(msgs))
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, caching.DisableMetrics)
for i, msg := range msgs {
testrig.MustPublishMsgs(t, jsctx, msg)
time.Sleep(100 * time.Millisecond)
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
})))
if w.Code != 200 {
t.Errorf("got HTTP %d want 200", w.Code)
continue
}
var res types.Response
if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
t.Errorf("failed to decode response body: %s", err)
}
sinceTokens[i] = res.NextBatch.String()
if i == 0 { // create event does not produce a room section
if len(res.Rooms.Join) != 0 {
t.Fatalf("i=%v got %d joined rooms, want 0", i, len(res.Rooms.Join))
}
} else { // we should have that room somewhere
if len(res.Rooms.Join) != 1 {
t.Fatalf("i=%v got %d joined rooms, want 1", i, len(res.Rooms.Join))
}
}
}
// sync with no token "" and with the penultimate token and this should neatly return room events in the timeline block
sinceTokens = append([]string{""}, sinceTokens[:len(sinceTokens)-1]...)
t.Logf("waited for events to be consumed; syncing with %v", sinceTokens)
for i, since := range sinceTokens {
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
"since": since,
})))
if w.Code != 200 {
t.Errorf("since=%s got HTTP %d want 200", since, w.Code)
}
var res types.Response
if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
t.Errorf("failed to decode response body: %s", err)
}
if len(res.Rooms.Join) != 1 {
t.Fatalf("since=%s got %d joined rooms, want 1", since, len(res.Rooms.Join))
}
t.Logf("since=%s res state:%+v res timeline:%+v", since, res.Rooms.Join[room.ID].State.Events, res.Rooms.Join[room.ID].Timeline.Events)
gotEventIDs := make([]string, len(res.Rooms.Join[room.ID].Timeline.Events))
for j, ev := range res.Rooms.Join[room.ID].Timeline.Events {
gotEventIDs[j] = ev.EventID
}
test.AssertEventIDsEqual(t, gotEventIDs, room.Events()[i:])
}
}
// Test that if we hit /sync we get back presence: online, regardless of whether messages get delivered
// via NATS. Regression test for a flakey test "User sees their own presence in a sync"
func TestSyncAPIUpdatePresenceImmediately(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
testSyncAPIUpdatePresenceImmediately(t, dbType)
})
}
func testSyncAPIUpdatePresenceImmediately(t *testing.T, dbType test.DBType) {
user := test.NewUser(t)
alice := userapi.Device{
ID: "ALICEID",
UserID: user.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "Alice",
AccountType: userapi.AccountTypeUser,
}
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
cfg.Global.Presence.EnableOutbound = true
cfg.Global.Presence.EnableInbound = true
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches, caching.DisableMetrics)
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"timeout": "0",
"set_presence": "online",
})))
if w.Code != 200 {
t.Fatalf("got HTTP %d want %d", w.Code, 200)
}
var res types.Response
if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
t.Errorf("failed to decode response body: %s", err)
}
if len(res.Presence.Events) != 1 {
t.Fatalf("expected 1 presence events, got: %+v", res.Presence.Events)
}
if res.Presence.Events[0].Sender != alice.UserID {
t.Errorf("sender: got %v want %v", res.Presence.Events[0].Sender, alice.UserID)
}
if res.Presence.Events[0].Type != "m.presence" {
t.Errorf("type: got %v want %v", res.Presence.Events[0].Type, "m.presence")
}
if gjson.ParseBytes(res.Presence.Events[0].Content).Get("presence").Str != "online" {
t.Errorf("content: not online, got %v", res.Presence.Events[0].Content)
}
}
// This is mainly what Sytest is doing in "test_history_visibility"
func TestMessageHistoryVisibility(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
testHistoryVisibility(t, dbType)
})
}
func testHistoryVisibility(t *testing.T, dbType test.DBType) {
type result struct {
seeWithoutJoin bool
seeBeforeJoin bool
seeAfterInvite bool
}
// create the users
alice := test.NewUser(t)
aliceDev := userapi.Device{
ID: "ALICEID",
UserID: alice.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "ALICE",
}
bob := test.NewUser(t)
bobDev := userapi.Device{
ID: "BOBID",
UserID: bob.ID,
AccessToken: "BOD_BEARER_TOKEN",
DisplayName: "BOB",
}
ctx := context.Background()
// check guest and normal user accounts
for _, accType := range []userapi.AccountType{userapi.AccountTypeGuest, userapi.AccountTypeUser} {
testCases := []struct {
historyVisibility gomatrixserverlib.HistoryVisibility
wantResult result
}{
{
historyVisibility: gomatrixserverlib.HistoryVisibilityWorldReadable,
wantResult: result{
seeWithoutJoin: true,
seeBeforeJoin: true,
seeAfterInvite: true,
},
},
{
historyVisibility: gomatrixserverlib.HistoryVisibilityShared,
wantResult: result{
seeWithoutJoin: false,
seeBeforeJoin: true,
seeAfterInvite: true,
},
},
{
historyVisibility: gomatrixserverlib.HistoryVisibilityInvited,
wantResult: result{
seeWithoutJoin: false,
seeBeforeJoin: false,
seeAfterInvite: true,
},
},
{
historyVisibility: gomatrixserverlib.HistoryVisibilityJoined,
wantResult: result{
seeWithoutJoin: false,
seeBeforeJoin: false,
seeAfterInvite: false,
},
},
}
bobDev.AccountType = accType
userType := "guest"
if accType == userapi.AccountTypeUser {
userType = "real user"
}
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use the actual internal roomserver API
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, caching.DisableMetrics)
for _, tc := range testCases {
testname := fmt.Sprintf("%s - %s", tc.historyVisibility, userType)
t.Run(testname, func(t *testing.T) {
// create a room with the given visibility
room := test.NewRoom(t, alice, test.RoomHistoryVisibility(tc.historyVisibility))
// send the events/messages to NATS to create the rooms
beforeJoinBody := fmt.Sprintf("Before invite in a %s room", tc.historyVisibility)
beforeJoinEv := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": beforeJoinBody})
eventsToSend := append(room.Events(), beforeJoinEv)
if err := api.SendEvents(ctx, rsAPI, api.KindNew, eventsToSend, "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
syncUntil(t, routers, aliceDev.AccessToken, false,
func(syncBody string) bool {
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(content.body=="%s")`, room.ID, beforeJoinBody)
return gjson.Get(syncBody, path).Exists()
},
)
// There is only one event, we expect only to be able to see this, if the room is world_readable
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
"access_token": bobDev.AccessToken,
"dir": "b",
"filter": `{"lazy_load_members":true}`, // check that lazy loading doesn't break history visibility
})))
if w.Code != 200 {
t.Logf("%s", w.Body.String())
t.Fatalf("got HTTP %d want %d", w.Code, 200)
}
// We only care about the returned events at this point
var res struct {
Chunk []synctypes.ClientEvent `json:"chunk"`
}
if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
t.Errorf("failed to decode response body: %s", err)
}
verifyEventVisible(t, tc.wantResult.seeWithoutJoin, beforeJoinEv, res.Chunk)
// Create invite, a message, join the room and create another message.
inviteEv := room.CreateAndInsert(t, alice, "m.room.member", map[string]interface{}{"membership": "invite"}, test.WithStateKey(bob.ID))
afterInviteEv := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": fmt.Sprintf("After invite in a %s room", tc.historyVisibility)})
joinEv := room.CreateAndInsert(t, bob, "m.room.member", map[string]interface{}{"membership": "join"}, test.WithStateKey(bob.ID))
afterJoinBody := fmt.Sprintf("After join in a %s room", tc.historyVisibility)
msgEv := room.CreateAndInsert(t, alice, "m.room.message", map[string]interface{}{"body": afterJoinBody})
eventsToSend = append([]*gomatrixserverlib.HeaderedEvent{}, inviteEv, afterInviteEv, joinEv, msgEv)
if err := api.SendEvents(ctx, rsAPI, api.KindNew, eventsToSend, "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
syncUntil(t, routers, aliceDev.AccessToken, false,
func(syncBody string) bool {
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(content.body=="%s")`, room.ID, afterJoinBody)
return gjson.Get(syncBody, path).Exists()
},
)
// Verify the messages after/before invite are visible or not
w = httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/messages", room.ID), test.WithQueryParams(map[string]string{
"access_token": bobDev.AccessToken,
"dir": "b",
})))
if w.Code != 200 {
t.Logf("%s", w.Body.String())
t.Fatalf("got HTTP %d want %d", w.Code, 200)
}
if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
t.Errorf("failed to decode response body: %s", err)
}
// verify results
verifyEventVisible(t, tc.wantResult.seeBeforeJoin, beforeJoinEv, res.Chunk)
verifyEventVisible(t, tc.wantResult.seeAfterInvite, afterInviteEv, res.Chunk)
})
}
}
}
func verifyEventVisible(t *testing.T, wantVisible bool, wantVisibleEvent *gomatrixserverlib.HeaderedEvent, chunk []synctypes.ClientEvent) {
t.Helper()
if wantVisible {
for _, ev := range chunk {
if ev.EventID == wantVisibleEvent.EventID() {
return
}
}
t.Fatalf("expected to see event %s but didn't: %+v", wantVisibleEvent.EventID(), chunk)
} else {
for _, ev := range chunk {
if ev.EventID == wantVisibleEvent.EventID() {
t.Fatalf("expected not to see event %s: %+v", wantVisibleEvent.EventID(), string(ev.Content))
}
}
}
}
func TestGetMembership(t *testing.T) {
alice := test.NewUser(t)
aliceDev := userapi.Device{
ID: "ALICEID",
UserID: alice.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "Alice",
AccountType: userapi.AccountTypeUser,
}
bob := test.NewUser(t)
bobDev := userapi.Device{
ID: "BOBID",
UserID: bob.ID,
AccessToken: "notjoinedtoanyrooms",
}
testCases := []struct {
name string
roomID string
additionalEvents func(t *testing.T, room *test.Room)
request func(t *testing.T, room *test.Room) *http.Request
wantOK bool
wantMemberCount int
useSleep bool // :/
}{
{
name: "/members - Alice joined",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
}))
},
wantOK: true,
wantMemberCount: 1,
},
{
name: "/members - Bob never joined",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": bobDev.AccessToken,
}))
},
wantOK: false,
},
{
name: "/joined_members - Bob never joined",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/joined_members", room.ID), test.WithQueryParams(map[string]string{
"access_token": bobDev.AccessToken,
}))
},
wantOK: false,
},
{
name: "/joined_members - Alice joined",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/joined_members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
}))
},
wantOK: true,
},
{
name: "Alice leaves before Bob joins, should not be able to see Bob",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
}))
},
additionalEvents: func(t *testing.T, room *test.Room) {
room.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": "leave",
}, test.WithStateKey(alice.ID))
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
},
useSleep: true,
wantOK: true,
wantMemberCount: 1,
},
{
name: "Alice leaves after Bob joins, should be able to see Bob",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
}))
},
additionalEvents: func(t *testing.T, room *test.Room) {
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
room.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": "leave",
}, test.WithStateKey(alice.ID))
},
useSleep: true,
wantOK: true,
wantMemberCount: 2,
},
{
name: "/joined_members - Alice leaves, shouldn't be able to see members ",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/joined_members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
}))
},
additionalEvents: func(t *testing.T, room *test.Room) {
room.CreateAndInsert(t, alice, spec.MRoomMember, map[string]interface{}{
"membership": "leave",
}, test.WithStateKey(alice.ID))
},
useSleep: true,
wantOK: false,
},
{
name: "'at' specified, returns memberships before Bob joins",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
"at": "t2_5",
}))
},
additionalEvents: func(t *testing.T, room *test.Room) {
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
},
useSleep: true,
wantOK: true,
wantMemberCount: 1,
},
{
name: "'membership=leave' specified, returns no memberships",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
"membership": "leave",
}))
},
wantOK: true,
wantMemberCount: 0,
},
{
name: "'not_membership=join' specified, returns no memberships",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
"not_membership": "join",
}))
},
wantOK: true,
wantMemberCount: 0,
},
{
name: "'not_membership=leave' & 'membership=join' specified, returns correct memberships",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", room.ID), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
"not_membership": "leave",
"membership": "join",
}))
},
additionalEvents: func(t *testing.T, room *test.Room) {
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
room.CreateAndInsert(t, bob, spec.MRoomMember, map[string]interface{}{
"membership": "leave",
}, test.WithStateKey(bob.ID))
},
wantOK: true,
wantMemberCount: 1,
},
{
name: "non-existent room ID",
request: func(t *testing.T, room *test.Room) *http.Request {
return test.NewRequest(t, "GET", fmt.Sprintf("/_matrix/client/v3/rooms/%s/members", "!notavalidroom:test"), test.WithQueryParams(map[string]string{
"access_token": aliceDev.AccessToken,
}))
},
wantOK: false,
},
}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
// Use an actual roomserver for this
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, caching.DisableMetrics)
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
room := test.NewRoom(t, alice)
t.Cleanup(func() {
t.Logf("running cleanup for %s", tc.name)
})
// inject additional events
if tc.additionalEvents != nil {
tc.additionalEvents(t, room)
}
if err := api.SendEvents(context.Background(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// wait for the events to come down sync
if tc.useSleep {
time.Sleep(time.Millisecond * 100)
} else {
syncUntil(t, routers, aliceDev.AccessToken, false, func(syncBody string) bool {
// wait for the last sent eventID to come down sync
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(event_id=="%s")`, room.ID, room.Events()[len(room.Events())-1].EventID())
return gjson.Get(syncBody, path).Exists()
})
}
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, tc.request(t, room))
if w.Code != 200 && tc.wantOK {
t.Logf("%s", w.Body.String())
t.Fatalf("got HTTP %d want %d", w.Code, 200)
}
t.Logf("[%s] Resp: %s", tc.name, w.Body.String())
// check we got the expected events
if tc.wantOK {
memberCount := len(gjson.GetBytes(w.Body.Bytes(), "chunk").Array())
if memberCount != tc.wantMemberCount {
t.Fatalf("expected %d members, got %d", tc.wantMemberCount, memberCount)
}
}
})
}
})
}
func TestSendToDevice(t *testing.T) {
test.WithAllDatabases(t, testSendToDevice)
}
func testSendToDevice(t *testing.T, dbType test.DBType) {
user := test.NewUser(t)
alice := userapi.Device{
ID: "ALICEID",
UserID: user.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "Alice",
AccountType: userapi.AccountTypeUser,
}
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
natsInstance := jetstream.NATSInstance{}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches, caching.DisableMetrics)
producer := producers.SyncAPIProducer{
TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
JetStream: jsctx,
}
msgCounter := 0
testCases := []struct {
name string
since string
want []string
sendMessagesCount int
}{
{
name: "initial sync, no messages",
want: []string{},
},
{
name: "initial sync, one new message",
sendMessagesCount: 1,
want: []string{
"message 1",
},
},
{
name: "initial sync, two new messages", // we didn't advance the since token, so we'll receive two messages
sendMessagesCount: 1,
want: []string{
"message 1",
"message 2",
},
},
{
name: "incremental sync, one message", // this deletes message 1, as we advanced the since token
since: types.StreamingToken{SendToDevicePosition: 1}.String(),
want: []string{
"message 2",
},
},
{
name: "failed incremental sync, one message", // didn't advance since, so still the same message
since: types.StreamingToken{SendToDevicePosition: 1}.String(),
want: []string{
"message 2",
},
},
{
name: "incremental sync, no message", // this should delete message 2
since: types.StreamingToken{SendToDevicePosition: 2}.String(), // next_batch from previous sync
want: []string{},
},
{
name: "incremental sync, three new messages",
since: types.StreamingToken{SendToDevicePosition: 2}.String(),
sendMessagesCount: 3,
want: []string{
"message 3", // message 2 was deleted in the previous test
"message 4",
"message 5",
},
},
{
name: "initial sync, three messages", // we expect three messages, as we didn't go beyond "2"
want: []string{
"message 3",
"message 4",
"message 5",
},
},
{
name: "incremental sync, no messages", // advance the sync token, no new messages
since: types.StreamingToken{SendToDevicePosition: 5}.String(),
want: []string{},
},
}
ctx := context.Background()
for _, tc := range testCases {
// Send to-device messages of type "m.dendrite.test" with content `{"dummy":"message $counter"}`
for i := 0; i < tc.sendMessagesCount; i++ {
msgCounter++
msg := json.RawMessage(fmt.Sprintf(`{"dummy":"message %d"}`, msgCounter))
if err := producer.SendToDevice(ctx, user.ID, user.ID, alice.ID, "m.dendrite.test", msg); err != nil {
t.Fatalf("unable to send to device message: %v", err)
}
}
syncUntil(t, routers, alice.AccessToken,
len(tc.want) == 0,
func(body string) bool {
return gjson.Get(body, fmt.Sprintf(`to_device.events.#(content.dummy=="message %d")`, msgCounter)).Exists()
},
)
// Execute a /sync request, recording the response
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken,
"since": tc.since,
})))
// Extract the to_device.events, # gets all values of an array, in this case a string slice with "message $counter" entries
events := gjson.Get(w.Body.String(), "to_device.events.#.content.dummy").Array()
got := make([]string, len(events))
for i := range events {
got[i] = events[i].String()
}
// Ensure the messages we received are as we expect them to be
if !reflect.DeepEqual(got, tc.want) {
t.Logf("[%s|since=%s]: Sync: %s", tc.name, tc.since, w.Body.String())
t.Fatalf("[%s|since=%s]: got: %+v, want: %+v", tc.name, tc.since, got, tc.want)
}
}
}
func TestContext(t *testing.T) {
test.WithAllDatabases(t, testContext)
}
func testContext(t *testing.T, dbType test.DBType) {
tests := []struct {
name string
roomID string
eventID string
params map[string]string
wantError bool
wantStateLength int
wantBeforeLength int
wantAfterLength int
}{
{
name: "invalid filter",
params: map[string]string{
"filter": "{",
},
wantError: true,
},
{
name: "invalid limit",
params: map[string]string{
"limit": "abc",
},
wantError: true,
},
{
name: "high limit",
params: map[string]string{
"limit": "100000",
},
},
{
name: "fine limit",
params: map[string]string{
"limit": "10",
},
},
{
name: "last event without lazy loading",
wantStateLength: 5,
},
{
name: "last event with lazy loading",
params: map[string]string{
"filter": `{"lazy_load_members":true}`,
},
wantStateLength: 1,
},
{
name: "invalid room",
roomID: "!doesnotexist",
wantError: true,
},
{
name: "invalid eventID",
eventID: "$doesnotexist",
wantError: true,
},
{
name: "state is limited",
params: map[string]string{
"limit": "1",
},
wantStateLength: 1,
},
{
name: "events are not limited",
wantBeforeLength: 7,
},
{
name: "all events are limited",
params: map[string]string{
"limit": "1",
},
wantStateLength: 1,
wantBeforeLength: 1,
wantAfterLength: 1,
},
}
user := test.NewUser(t)
alice := userapi.Device{
ID: "ALICEID",
UserID: user.ID,
AccessToken: "ALICE_BEARER_TOKEN",
DisplayName: "Alice",
AccountType: userapi.AccountTypeUser,
}
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
routers := httputil.NewRouters()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close()
// Use an actual roomserver for this
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI, caches, caching.DisableMetrics)
room := test.NewRoom(t, user)
room.CreateAndInsert(t, user, "m.room.message", map[string]interface{}{"body": "hello world 1!"})
room.CreateAndInsert(t, user, "m.room.message", map[string]interface{}{"body": "hello world 2!"})
thirdMsg := room.CreateAndInsert(t, user, "m.room.message", map[string]interface{}{"body": "hello world3!"})
room.CreateAndInsert(t, user, "m.room.message", map[string]interface{}{"body": "hello world4!"})
if err := api.SendEvents(context.Background(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
syncUntil(t, routers, alice.AccessToken, false, func(syncBody string) bool {
// wait for the last sent eventID to come down sync
path := fmt.Sprintf(`rooms.join.%s.timeline.events.#(event_id=="%s")`, room.ID, thirdMsg.EventID())
return gjson.Get(syncBody, path).Exists()
})
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
params := map[string]string{
"access_token": alice.AccessToken,
}
w := httptest.NewRecorder()
// test overrides
roomID := room.ID
if tc.roomID != "" {
roomID = tc.roomID
}
eventID := thirdMsg.EventID()
if tc.eventID != "" {
eventID = tc.eventID
}
requestPath := fmt.Sprintf("/_matrix/client/v3/rooms/%s/context/%s", roomID, eventID)
if tc.params != nil {
for k, v := range tc.params {
params[k] = v
}
}
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", requestPath, test.WithQueryParams(params)))
if tc.wantError && w.Code == 200 {
t.Fatalf("Expected an error, but got none")
}
t.Log(w.Body.String())
resp := routing.ContextRespsonse{}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if tc.wantStateLength > 0 && tc.wantStateLength != len(resp.State) {
t.Fatalf("expected %d state events, got %d", tc.wantStateLength, len(resp.State))
}
if tc.wantBeforeLength > 0 && tc.wantBeforeLength != len(resp.EventsBefore) {
t.Fatalf("expected %d before events, got %d", tc.wantBeforeLength, len(resp.EventsBefore))
}
if tc.wantAfterLength > 0 && tc.wantAfterLength != len(resp.EventsAfter) {
t.Fatalf("expected %d after events, got %d", tc.wantAfterLength, len(resp.EventsAfter))
}
if !tc.wantError && resp.Event.EventID != eventID {
t.Fatalf("unexpected eventID %s, expected %s", resp.Event.EventID, eventID)
}
})
}
}
func TestUpdateRelations(t *testing.T) {
testCases := []struct {
name string
eventContent map[string]interface{}
eventType string
}{
{
name: "empty event content should not error",
},
{
name: "unable to unmarshal event should not error",
eventContent: map[string]interface{}{
"m.relates_to": map[string]interface{}{
"event_id": map[string]interface{}{}, // this should be a string and not struct
},
},
},
{
name: "empty event ID is ignored",
eventContent: map[string]interface{}{
"m.relates_to": map[string]interface{}{
"event_id": "",
},
},
},
{
name: "empty rel_type is ignored",
eventContent: map[string]interface{}{
"m.relates_to": map[string]interface{}{
"event_id": "$randomEventID",
"rel_type": "",
},
},
},
{
name: "redactions are ignored",
eventType: spec.MRoomRedaction,
eventContent: map[string]interface{}{
"m.relates_to": map[string]interface{}{
"event_id": "$randomEventID",
"rel_type": "m.replace",
},
},
},
{
name: "valid event is correctly written",
eventContent: map[string]interface{}{
"m.relates_to": map[string]interface{}{
"event_id": "$randomEventID",
"rel_type": "m.replace",
},
},
},
}
ctx := context.Background()
alice := test.NewUser(t)
room := test.NewRoom(t, alice)
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
cfg, processCtx, close := testrig.CreateConfig(t, dbType)
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
t.Cleanup(close)
db, err := storage.NewSyncServerDatasource(processCtx.Context(), cm, &cfg.SyncAPI.Database)
if err != nil {
t.Fatal(err)
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
evType := "m.room.message"
if tc.eventType != "" {
evType = tc.eventType
}
ev := room.CreateEvent(t, alice, evType, tc.eventContent)
err = db.UpdateRelations(ctx, ev)
if err != nil {
t.Fatal(err)
}
})
}
})
}
func syncUntil(t *testing.T,
routers httputil.Routers, accessToken string,
skip bool,
checkFunc func(syncBody string) bool,
) {
t.Helper()
if checkFunc == nil {
t.Fatalf("No checkFunc defined")
}
if skip {
return
}
// loop on /sync until we receive the last send message or timeout after 5 seconds, since we don't know if the message made it
// to the syncAPI when hitting /sync
done := make(chan bool)
defer close(done)
go func() {
for {
w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": accessToken,
"timeout": "1000",
})))
if checkFunc(w.Body.String()) {
done <- true
return
}
}
}()
select {
case <-done:
case <-time.After(time.Second * 5):
t.Fatalf("Timed out waiting for messages")
}
}
func toNATSMsgs(t *testing.T, cfg *config.Dendrite, input ...*gomatrixserverlib.HeaderedEvent) []*nats.Msg {
result := make([]*nats.Msg, len(input))
for i, ev := range input {
var addsStateIDs []string
if ev.StateKey() != nil {
addsStateIDs = append(addsStateIDs, ev.EventID())
}
result[i] = testrig.NewOutputEventMsg(t, cfg, ev.RoomID(), api.OutputEvent{
Type: rsapi.OutputTypeNewRoomEvent,
NewRoomEvent: &rsapi.OutputNewRoomEvent{
Event: ev,
AddsStateEventIDs: addsStateIDs,
HistoryVisibility: ev.Visibility,
},
})
}
return result
}