test_devices.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. from twisted.internet import defer
  16. import synapse.api.errors
  17. import tests.unittest
  18. import tests.utils
  19. class DeviceStoreTestCase(tests.unittest.TestCase):
  20. def __init__(self, *args, **kwargs):
  21. super(DeviceStoreTestCase, self).__init__(*args, **kwargs)
  22. self.store = None # type: synapse.storage.DataStore
  23. @defer.inlineCallbacks
  24. def setUp(self):
  25. hs = yield tests.utils.setup_test_homeserver(self.addCleanup)
  26. self.store = hs.get_datastore()
  27. @defer.inlineCallbacks
  28. def test_store_new_device(self):
  29. yield self.store.store_device("user_id", "device_id", "display_name")
  30. res = yield self.store.get_device("user_id", "device_id")
  31. self.assertDictContainsSubset(
  32. {
  33. "user_id": "user_id",
  34. "device_id": "device_id",
  35. "display_name": "display_name",
  36. },
  37. res,
  38. )
  39. @defer.inlineCallbacks
  40. def test_get_devices_by_user(self):
  41. yield self.store.store_device("user_id", "device1", "display_name 1")
  42. yield self.store.store_device("user_id", "device2", "display_name 2")
  43. yield self.store.store_device("user_id2", "device3", "display_name 3")
  44. res = yield self.store.get_devices_by_user("user_id")
  45. self.assertEqual(2, len(res.keys()))
  46. self.assertDictContainsSubset(
  47. {
  48. "user_id": "user_id",
  49. "device_id": "device1",
  50. "display_name": "display_name 1",
  51. },
  52. res["device1"],
  53. )
  54. self.assertDictContainsSubset(
  55. {
  56. "user_id": "user_id",
  57. "device_id": "device2",
  58. "display_name": "display_name 2",
  59. },
  60. res["device2"],
  61. )
  62. @defer.inlineCallbacks
  63. def test_update_device(self):
  64. yield self.store.store_device("user_id", "device_id", "display_name 1")
  65. res = yield self.store.get_device("user_id", "device_id")
  66. self.assertEqual("display_name 1", res["display_name"])
  67. # do a no-op first
  68. yield self.store.update_device("user_id", "device_id")
  69. res = yield self.store.get_device("user_id", "device_id")
  70. self.assertEqual("display_name 1", res["display_name"])
  71. # do the update
  72. yield self.store.update_device(
  73. "user_id", "device_id", new_display_name="display_name 2"
  74. )
  75. # check it worked
  76. res = yield self.store.get_device("user_id", "device_id")
  77. self.assertEqual("display_name 2", res["display_name"])
  78. @defer.inlineCallbacks
  79. def test_update_unknown_device(self):
  80. with self.assertRaises(synapse.api.errors.StoreError) as cm:
  81. yield self.store.update_device(
  82. "user_id", "unknown_device_id", new_display_name="display_name 2"
  83. )
  84. self.assertEqual(404, cm.exception.code)