MSC4308: Thread Subscriptions extension to Sliding Sync tests

.github/workflows/ci.yaml

@@ -42,7 +42,7 @@ jobs:
           - homeserver: Synapse
             repo: element-hq/synapse
             tags: synapse_blacklist
-            packages: ./tests/msc3874 ./tests/msc3902
+            packages: ./tests/msc3874 ./tests/msc3902 ./tests/msc4306
             env: "COMPLEMENT_ENABLE_DIRTY_RUNS=1 COMPLEMENT_SHARE_ENV_PREFIX=PASS_ PASS_SYNAPSE_COMPLEMENT_DATABASE=sqlite"
             timeout: 20m
 

client/client.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"math/rand"
 	"net/http"
 	"net/http/httputil"
@@ -261,12 +262,12 @@ func (c *CSAPI) GetAllPushRules(t ct.TestLike) gjson.Result {
 	return gjson.ParseBytes(pushRulesBytes)
 }
 
-// GetPushRule queries the contents of a client's push rule by scope, kind and rule ID.
+// MustGetPushRule queries the contents of a client's push rule by scope, kind and rule ID.
 // A parsed gjson result is returned. Fails the test if the query to server returns a non-2xx status code.
 //
 // Example of checking that a global underride rule contains the expected actions:
 //
-//	containsDisplayNameRule := c.GetPushRule(t, "global", "underride", ".m.rule.contains_display_name")
+//	containsDisplayNameRule := c.MustGetPushRule(t, "global", "underride", ".m.rule.contains_display_name")
 //	must.MatchGJSON(
 //	  t,
 //	  containsDisplayNameRule,
@@ -276,14 +277,23 @@ func (c *CSAPI) GetAllPushRules(t ct.TestLike) gjson.Result {
 //	    map[string]interface{}{"set_tweak": "highlight"},
 //	  }),
 //	)
-func (c *CSAPI) GetPushRule(t ct.TestLike, scope string, kind string, ruleID string) gjson.Result {
+func (c *CSAPI) MustGetPushRule(t ct.TestLike, scope string, kind string, ruleID string) gjson.Result {
 	t.Helper()
 
-	res := c.MustDo(t, "GET", []string{"_matrix", "client", "v3", "pushrules", scope, kind, ruleID})
+	res := c.GetPushRule(t, scope, kind, ruleID)
+	mustRespond2xx(t, res)
+
 	pushRuleBytes := ParseJSON(t, res)
 	return gjson.ParseBytes(pushRuleBytes)
 }
 
+// GetPushRule queries the contents of a client's push rule by scope, kind and rule ID.
+func (c *CSAPI) GetPushRule(t ct.TestLike, scope string, kind string, ruleID string) *http.Response {
+	t.Helper()
+
+	return c.Do(t, "GET", []string{"_matrix", "client", "v3", "pushrules", scope, kind, ruleID})
+}
+
 // SetPushRule creates a new push rule on the user, or modifies an existing one.
 // If `before` or `after` parameters are not set to an empty string, their values
 // will be set as the `before` and `after` query parameters respectively on the
@@ -310,6 +320,14 @@ func (c *CSAPI) SetPushRule(t ct.TestLike, scope string, kind string, ruleID str
 	return c.MustDo(t, "PUT", []string{"_matrix", "client", "v3", "pushrules", scope, kind, ruleID}, WithJSONBody(t, body), WithQueries(queryParams))
 }
 
+// MustDisablePushRule disables a push rule on the user.
+// Fails the test if response is non-2xx.
+func (c *CSAPI) MustDisablePushRule(t ct.TestLike, scope string, kind string, ruleID string) {
+	c.MustDo(t, "PUT", []string{"_matrix", "client", "v3", "pushrules", scope, kind, ruleID, "enabled"}, WithJSONBody(t, map[string]interface{}{
+		"enabled": false,
+	}))
+}
+
 // Unsafe_SendEventUnsynced sends `e` into the room. This function is UNSAFE as it does not wait
 // for the event to be fully processed. This can cause flakey tests. Prefer `SendEventSynced`.
 // Returns the event ID of the sent event.
@@ -735,6 +753,19 @@ func (t *loggedRoundTripper) RoundTrip(req *http.Request) (*http.Response, error
 	return res, err
 }
 
+// Extracts a JSON object given a search key
+// Caller must check `result.Exists()` to see whether the object actually exists.
+func GetOptionalJSONFieldObject(t ct.TestLike, body []byte, wantKey string) gjson.Result {
+	t.Helper()
+	res := gjson.GetBytes(body, wantKey)
+	if !res.Exists() {
+		log.Printf("OptionalJSONFieldObject: key '%s' absent from %s", wantKey, string(body))
+	} else if !res.IsObject() {
+		ct.Fatalf(t, "OptionalJSONFieldObject: key '%s' is not an object, body: %s", wantKey, string(body))
+	}
+	return res
+}
+
 // GetJSONFieldStr extracts a value from a byte-encoded JSON body given a search key
 func GetJSONFieldStr(t ct.TestLike, body []byte, wantKey string) string {
 	t.Helper()

tests/csapi/thread_notifications_test.go

@@ -51,6 +51,21 @@ func syncHasThreadedReadReceipt(roomID, userID, eventID, threadID string) client
 	})
 }
 
+// Disables push rules that are introduced in MSC4306 (if present),
+// because they interfere with the normal semantics of notifications in threads.
+func disableMsc4306PushRules(t *testing.T, user *client.CSAPI) {
+	rules := []string{".io.element.msc4306.rule.subscribed_thread", ".io.element.msc4306.rule.unsubscribed_thread"}
+	for _, rule := range rules {
+		res := user.GetPushRule(t, "global", "postcontent", rule)
+		if res.StatusCode == 404 {
+			// No push rule to disable
+			continue
+		}
+
+		user.MustDisablePushRule(t, "global", "postcontent", rule)
+	}
+}
+
 // Test behavior of threaded receipts and notifications.
 //
 // 1. Send a series of messages, some of which are in threads.
@@ -79,7 +94,9 @@ func TestThreadedReceipts(t *testing.T) {
 
 	// Create a room with alice and bob.
 	alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{})
+	disableMsc4306PushRules(t, alice)
 	bob := deployment.Register(t, "hs1", helpers.RegistrationOpts{})
+	disableMsc4306PushRules(t, bob)
 
 	roomID := alice.MustCreateRoom(t, map[string]interface{}{"preset": "public_chat"})
 	bob.MustJoinRoom(t, roomID, nil)
@@ -312,7 +329,9 @@ func TestThreadReceiptsInSyncMSC4102(t *testing.T) {
 
 	// Create a room with alice and bob.
 	alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{})
+	disableMsc4306PushRules(t, alice)
 	bob := deployment.Register(t, "hs2", helpers.RegistrationOpts{})
+	disableMsc4306PushRules(t, bob)
 	roomID := alice.MustCreateRoom(t, map[string]interface{}{"preset": "public_chat"})
 	bob.MustJoinRoom(t, roomID, []spec.ServerName{
 		deployment.GetFullyQualifiedHomeserverName(t, "hs1"),

tests/msc3930/msc3930_test.go

@@ -39,7 +39,7 @@ func TestPollsLocalPushRules(t *testing.T) {
 		// Request each of the push rule IDs defined by MSC3930 and verify their structure.
 
 		// This push rule silences all poll responses.
-		pollResponseRule := alice.GetPushRule(t, "global", "override", pollResponseRuleID)
+		pollResponseRule := alice.MustGetPushRule(t, "global", "override", pollResponseRuleID)
 		must.MatchGJSON(
 			t,
 			pollResponseRule,
@@ -55,7 +55,7 @@ func TestPollsLocalPushRules(t *testing.T) {
 		)
 
 		// This push rule creates a sound and notifies the user when a poll is started in a one-to-one room.
-		pollStartOneToOneRule := alice.GetPushRule(t, "global", "underride", pollStartOneToOneRuleID)
+		pollStartOneToOneRule := alice.MustGetPushRule(t, "global", "underride", pollStartOneToOneRuleID)
 		must.MatchGJSON(
 			t,
 			pollStartOneToOneRule,
@@ -78,7 +78,7 @@ func TestPollsLocalPushRules(t *testing.T) {
 		)
 
 		// This push rule creates a sound and notifies the user when a poll is ended in a one-to-one room.
-		pollEndOneToOneRule := alice.GetPushRule(t, "global", "underride", pollEndOneToOneRuleID)
+		pollEndOneToOneRule := alice.MustGetPushRule(t, "global", "underride", pollEndOneToOneRuleID)
 		must.MatchGJSON(
 			t,
 			pollEndOneToOneRule,
@@ -101,7 +101,7 @@ func TestPollsLocalPushRules(t *testing.T) {
 		)
 
 		// This push rule notifies the user when a poll is started in any room.
-		pollStartRule := alice.GetPushRule(t, "global", "underride", pollStartRuleID)
+		pollStartRule := alice.MustGetPushRule(t, "global", "underride", pollStartRuleID)
 		must.MatchGJSON(
 			t,
 			pollStartRule,
@@ -118,7 +118,7 @@ func TestPollsLocalPushRules(t *testing.T) {
 		)
 
 		// This push rule notifies the user when a poll is ended in any room.
-		pollEndRule := alice.GetPushRule(t, "global", "underride", pollEndRuleID)
+		pollEndRule := alice.MustGetPushRule(t, "global", "underride", pollEndRuleID)
 		must.MatchGJSON(
 			t,
 			pollEndRule,

tests/msc4306/main_test.go

@@ -0,0 +1,11 @@
+package tests
+
+import (
+	"testing"
+
+	"github.com/matrix-org/complement"
+)
+
+func TestMain(m *testing.M) {
+	complement.TestMain(m, "msc4306")
+}

tests/msc4306/thread_subscriptions_sliding_sync_test.go

@@ -0,0 +1,97 @@
+package tests
+
+import (
+	"testing"
+
+	"github.com/matrix-org/complement"
+	"github.com/matrix-org/complement/b"
+	"github.com/matrix-org/complement/client"
+	"github.com/matrix-org/complement/helpers"
+	"github.com/matrix-org/complement/match"
+	"github.com/matrix-org/complement/must"
+	"github.com/tidwall/gjson"
+)
+
+func MustDoSlidingSync(t *testing.T, user *client.CSAPI, pos string, thread_subs_ext map[string]interface{}) (string, gjson.Result) {
+	body := map[string]interface{}{
+		"extensions": map[string]interface{}{
+			"io.element.msc4308.thread_subscriptions": thread_subs_ext,
+		},
+	}
+	if pos != "" {
+		body["pos"] = pos
+	}
+	resp := user.MustDo(t, "POST", []string{"_matrix", "client", "unstable", "org.matrix.simplified_msc3575", "sync"}, client.WithJSONBody(t, body))
+	respBody := client.ParseJSON(t, resp)
+
+	newPos := client.GetJSONFieldStr(t, respBody, "pos")
+	extension := client.GetOptionalJSONFieldObject(t, respBody, "extensions.io\\.element\\.msc4308\\.thread_subscriptions")
+	if !extension.Exists() {
+		// Missing extension is semantically the same as an empty one
+		// So eliminate the nil for simplicity
+		extension = gjson.Parse("{}")
+	}
+	return newPos, extension
+}
+
+func TestThreadSubscriptionsSlidingSync(t *testing.T) {
+	deployment := complement.Deploy(t, 1)
+	defer deployment.Destroy(t)
+
+	t.Run("Receives thread subscriptions over initial sliding sync", func(t *testing.T) {
+		alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{})
+		roomID := alice.MustCreateRoom(t, map[string]interface{}{})
+		threadRootID := alice.SendEventSynced(t, roomID, b.Event{
+			Type: "m.room.message",
+			Content: map[string]interface{}{
+				"msgtype": "m.text",
+				"body":    "what do you think? reply in a thread!",
+			},
+		})
+
+		// Subscribe to the thread manually
+		alice.MustDo(t, "PUT", []string{"_matrix", "client", "unstable", "io.element.msc4306", "rooms", roomID, "thread", threadRootID, "subscription"}, client.WithJSONBody(t, map[string]interface{}{}))
+
+		_, ext := MustDoSlidingSync(t, alice, "", map[string]interface{}{
+			"enabled": true,
+			"limit":   2,
+		})
+
+		must.MatchGJSON(t, ext,
+			match.JSONKeyTypeEqual("subscribed."+gjson.Escape(roomID)+"."+gjson.Escape(threadRootID)+".bump_stamp", gjson.Number),
+			match.JSONKeyEqual("subscribed."+gjson.Escape(roomID)+"."+gjson.Escape(threadRootID)+".automatic", false),
+		)
+	})
+
+	t.Run("Receives thread subscriptions over incremental sliding sync", func(t *testing.T) {
+		alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{})
+		roomID := alice.MustCreateRoom(t, map[string]interface{}{})
+		threadRootID := alice.SendEventSynced(t, roomID, b.Event{
+			Type: "m.room.message",
+			Content: map[string]interface{}{
+				"msgtype": "m.text",
+				"body":    "what do you think? reply in a thread!",
+			},
+		})
+
+		newPos, ext := MustDoSlidingSync(t, alice, "", map[string]interface{}{
+			"enabled": true,
+			"limit":   2,
+		})
+		must.MatchGJSON(t, ext,
+			match.JSONKeyMissing("subscribed"))
+
+		// Subscribe to the thread manually
+		alice.MustDo(t, "PUT", []string{"_matrix", "client", "unstable", "io.element.msc4306", "rooms", roomID, "thread", threadRootID, "subscription"}, client.WithJSONBody(t, map[string]interface{}{}))
+
+		_, ext = MustDoSlidingSync(t, alice, newPos, map[string]interface{}{
+			"enabled": true,
+			"limit":   2,
+		})
+
+		must.MatchGJSON(t, ext,
+			match.JSONKeyTypeEqual("subscribed."+gjson.Escape(roomID)+"."+gjson.Escape(threadRootID)+".bump_stamp", gjson.Number),
+			match.JSONKeyEqual("subscribed."+gjson.Escape(roomID)+"."+gjson.Escape(threadRootID)+".automatic", false),
+		)
+	})
+}