mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/element-hq/dendrite.git synced 2025-09-14 05:12:26 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

more fixes

Browse source
This commit is contained in:
Roman Isaev 2025-01-08 12:43:22 +00:00
parent b44a79c159
commit 7311d3e1de
No known key found for this signature in database
GPG key ID: 7BE2B6A6C89AEC7F
1 changed files with 39 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

48
syncapi/syncapi_test.go
View file

@ -11,6 +11,7 @@ import (
"testing" "testing"
"time" "time"
"github.com/element-hq/dendrite/federationapi/statistics"
"github.com/element-hq/dendrite/internal/caching" "github.com/element-hq/dendrite/internal/caching"
"github.com/element-hq/dendrite/internal/httputil" "github.com/element-hq/dendrite/internal/httputil"
"github.com/element-hq/dendrite/internal/sqlutil" "github.com/element-hq/dendrite/internal/sqlutil"
@ -27,6 +28,7 @@ import (
"github.com/element-hq/dendrite/syncapi/storage" "github.com/element-hq/dendrite/syncapi/storage"
"github.com/element-hq/dendrite/syncapi/synctypes" "github.com/element-hq/dendrite/syncapi/synctypes"
"github.com/element-hq/dendrite/clientapi/auth"
"github.com/element-hq/dendrite/clientapi/producers" "github.com/element-hq/dendrite/clientapi/producers"
"github.com/element-hq/dendrite/roomserver" "github.com/element-hq/dendrite/roomserver"
"github.com/element-hq/dendrite/roomserver/api" "github.com/element-hq/dendrite/roomserver/api"
@ -35,9 +37,14 @@ import (
"github.com/element-hq/dendrite/syncapi/types" "github.com/element-hq/dendrite/syncapi/types"
"github.com/element-hq/dendrite/test" "github.com/element-hq/dendrite/test"
"github.com/element-hq/dendrite/test/testrig" "github.com/element-hq/dendrite/test/testrig"
usrapi "github.com/element-hq/dendrite/userapi"
userapi "github.com/element-hq/dendrite/userapi/api" userapi "github.com/element-hq/dendrite/userapi/api"
) )
var testIsBlacklistedOrBackingOff = func(s spec.ServerName) (*statistics.ServerStatistics, error) {
return &statistics.ServerStatistics{}, nil
}
type syncRoomserverAPI struct { type syncRoomserverAPI struct {
rsapi.SyncRoomserverAPI rsapi.SyncRoomserverAPI
rooms []*test.Room rooms []*test.Room
@ -141,12 +148,15 @@ func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions) cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics) caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
natsInstance := jetstream.NATSInstance{} natsInstance := jetstream.NATSInstance{}
userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
defer close() defer close()
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream) jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream) defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
msgs := toNATSMsgs(t, cfg, room.Events()...) 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)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, &userVerifier, caching.DisableMetrics)
testrig.MustPublishMsgs(t, jsctx, msgs...) testrig.MustPublishMsgs(t, jsctx, msgs...)
testCases := []struct { testCases := []struct {
@ -241,12 +251,14 @@ func testSyncEventFormatPowerLevels(t *testing.T, dbType test.DBType) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions) cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics) caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
natsInstance := jetstream.NATSInstance{} natsInstance := jetstream.NATSInstance{}
userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
defer close() defer close()
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream) jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream) defer jetstream.DeleteAllStreams(jsctx, &cfg.Global.JetStream)
msgs := toNATSMsgs(t, cfg, room.Events()...) 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) AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, &userVerifier, caching.DisableMetrics)
testrig.MustPublishMsgs(t, jsctx, msgs...) testrig.MustPublishMsgs(t, jsctx, msgs...)
testCases := []struct { testCases := []struct {
@ -399,7 +411,9 @@ func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
// m.room.history_visibility // m.room.history_visibility
msgs := toNATSMsgs(t, cfg, room.Events()...) msgs := toNATSMsgs(t, cfg, room.Events()...)
sinceTokens := make([]string, len(msgs)) sinceTokens := make([]string, len(msgs))
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, caching.DisableMetrics) userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, &userVerifier, caching.DisableMetrics)
for i, msg := range msgs { for i, msg := range msgs {
testrig.MustPublishMsgs(t, jsctx, msg) testrig.MustPublishMsgs(t, jsctx, msg)
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
@ -487,7 +501,9 @@ func testSyncAPIUpdatePresenceImmediately(t *testing.T, dbType test.DBType) {
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream) jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &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) userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches, &userVerifier, caching.DisableMetrics)
w := httptest.NewRecorder() w := httptest.NewRecorder()
routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{ routers.Client.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
"access_token": alice.AccessToken, "access_token": alice.AccessToken,
@ -609,7 +625,9 @@ func testHistoryVisibility(t *testing.T, dbType test.DBType) {
// Use the actual internal roomserver API // Use the actual internal roomserver API
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics) rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil) rsAPI.SetFederationAPI(nil, nil)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, caching.DisableMetrics) userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, &userVerifier, caching.DisableMetrics)
for _, tc := range testCases { for _, tc := range testCases {
testname := fmt.Sprintf("%s - %s", tc.historyVisibility, userType) testname := fmt.Sprintf("%s - %s", tc.historyVisibility, userType)
@ -879,7 +897,10 @@ func TestGetMembership(t *testing.T) {
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics) rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil) rsAPI.SetFederationAPI(nil, nil)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, caching.DisableMetrics) userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{aliceDev, bobDev}}, rsAPI, caches, &userVerifier, caching.DisableMetrics)
for _, tc := range testCases { for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
@ -946,10 +967,12 @@ func testSendToDevice(t *testing.T, dbType test.DBType) {
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics) caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
defer close() defer close()
natsInstance := jetstream.NATSInstance{} natsInstance := jetstream.NATSInstance{}
userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream) jsctx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsctx, &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) AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, caches, &userVerifier, caching.DisableMetrics)
producer := producers.SyncAPIProducer{ producer := producers.SyncAPIProducer{
TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent), TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
@ -1172,7 +1195,10 @@ func testContext(t *testing.T, dbType test.DBType) {
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics) rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil) rsAPI.SetFederationAPI(nil, nil)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI, caches, caching.DisableMetrics) userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, rsAPI, caches, &userVerifier, caching.DisableMetrics)
room := test.NewRoom(t, user) room := test.NewRoom(t, user)
@ -1352,8 +1378,11 @@ func TestRemoveEditedEventFromSearchIndex(t *testing.T) {
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics) rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.DisableMetrics)
rsAPI.SetFederationAPI(nil, nil) rsAPI.SetFederationAPI(nil, nil)
userAPI := usrapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, nil, nil, caching.DisableMetrics, testIsBlacklistedOrBackingOff)
userVerifier := auth.DefaultUserVerifier{UserAPI: userAPI}
room := test.NewRoom(t, user) room := test.NewRoom(t, user)
AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, caching.DisableMetrics) AddPublicRoutes(processCtx, routers, cfg, cm, &natsInstance, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, caches, &userVerifier, caching.DisableMetrics)
if err := api.SendEvents(processCtx.Context(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil { if err := api.SendEvents(processCtx.Context(), rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err) t.Fatalf("failed to send events: %v", err)
@ -1416,6 +1445,7 @@ func searchRequest(t *testing.T, router *mux.Router, accessToken, searchTerm str
assert.NoError(t, err) assert.NoError(t, err)
return body return body
} }
func syncUntil(t *testing.T, func syncUntil(t *testing.T,
routers httputil.Routers, accessToken string, routers httputil.Routers, accessToken string,
skip bool, skip bool,