diff --git a/tests/handlers/test_e2e_keys.py b/tests/handlers/test_e2e_keys.py
index 878a54dc34..19f5ed6bce 100644
--- a/tests/handlers/test_e2e_keys.py
+++ b/tests/handlers/test_e2e_keys.py
@@ -14,6 +14,7 @@
# limitations under the License.
import mock
+from synapse.api import errors
from twisted.internet import defer
import synapse.api.errors
@@ -44,3 +45,134 @@ class E2eKeysHandlerTestCase(unittest.TestCase):
local_user = "@boris:" + self.hs.hostname
res = yield self.handler.query_local_devices({local_user: None})
self.assertDictEqual(res, {local_user: {}})
+
+ @defer.inlineCallbacks
+ def test_reupload_one_time_keys(self):
+ """we should be able to re-upload the same keys"""
+ local_user = "@boris:" + self.hs.hostname
+ device_id = "xyz"
+ keys = {
+ "alg1:k1": "key1",
+ "alg2:k2": {
+ "key": "key2",
+ "signatures": {"k1": "sig1"}
+ },
+ "alg2:k3": {
+ "key": "key3",
+ },
+ }
+
+ res = yield self.handler.upload_keys_for_user(
+ local_user, device_id, {"one_time_keys": keys},
+ )
+ self.assertDictEqual(res, {
+ "one_time_key_counts": {"alg1": 1, "alg2": 2}
+ })
+
+ # we should be able to change the signature without a problem
+ keys["alg2:k2"]["signatures"]["k1"] = "sig2"
+ res = yield self.handler.upload_keys_for_user(
+ local_user, device_id, {"one_time_keys": keys},
+ )
+ self.assertDictEqual(res, {
+ "one_time_key_counts": {"alg1": 1, "alg2": 2}
+ })
+
+ @defer.inlineCallbacks
+ def test_change_one_time_keys(self):
+ """attempts to change one-time-keys should be rejected"""
+
+ local_user = "@boris:" + self.hs.hostname
+ device_id = "xyz"
+ keys = {
+ "alg1:k1": "key1",
+ "alg2:k2": {
+ "key": "key2",
+ "signatures": {"k1": "sig1"}
+ },
+ "alg2:k3": {
+ "key": "key3",
+ },
+ }
+
+ res = yield self.handler.upload_keys_for_user(
+ local_user, device_id, {"one_time_keys": keys},
+ )
+ self.assertDictEqual(res, {
+ "one_time_key_counts": {"alg1": 1, "alg2": 2}
+ })
+
+ try:
+ yield self.handler.upload_keys_for_user(
+ local_user, device_id, {"one_time_keys": {"alg1:k1": "key2"}},
+ )
+ self.fail("No error when changing string key")
+ except errors.SynapseError:
+ pass
+
+ try:
+ yield self.handler.upload_keys_for_user(
+ local_user, device_id, {"one_time_keys": {"alg2:k3": "key2"}},
+ )
+ self.fail("No error when replacing dict key with string")
+ except errors.SynapseError:
+ pass
+
+ try:
+ yield self.handler.upload_keys_for_user(
+ local_user, device_id, {
+ "one_time_keys": {"alg1:k1": {"key": "key"}}
+ },
+ )
+ self.fail("No error when replacing string key with dict")
+ except errors.SynapseError:
+ pass
+
+ try:
+ yield self.handler.upload_keys_for_user(
+ local_user, device_id, {
+ "one_time_keys": {
+ "alg2:k2": {
+ "key": "key3",
+ "signatures": {"k1": "sig1"},
+ }
+ },
+ },
+ )
+ self.fail("No error when replacing dict key")
+ except errors.SynapseError:
+ pass
+
+ @unittest.DEBUG
+ @defer.inlineCallbacks
+ def test_claim_one_time_key(self):
+ local_user = "@boris:" + self.hs.hostname
+ device_id = "xyz"
+ keys = {
+ "alg1:k1": "key1",
+ }
+
+ res = yield self.handler.upload_keys_for_user(
+ local_user, device_id, {"one_time_keys": keys},
+ )
+ self.assertDictEqual(res, {
+ "one_time_key_counts": {"alg1": 1}
+ })
+
+ res2 = yield self.handler.claim_one_time_keys({
+ "one_time_keys": {
+ local_user: {
+ device_id: "alg1"
+ }
+ }
+ }, timeout=None)
+ self.assertEqual(res2, {
+ "failures": {},
+ "one_time_keys": {
+ local_user: {
+ device_id: {
+ "alg1:k1": "key1"
+ }
+ }
+ }
+ })
|
