test_devices.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  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()
  26. self.store = hs.get_datastore()
  27. @defer.inlineCallbacks
  28. def test_store_new_device(self):
  29. yield self.store.store_device(
  30. "user_id", "device_id", "display_name"
  31. )
  32. res = yield self.store.get_device("user_id", "device_id")
  33. self.assertDictContainsSubset({
  34. "user_id": "user_id",
  35. "device_id": "device_id",
  36. "display_name": "display_name",
  37. }, res)
  38. @defer.inlineCallbacks
  39. def test_get_devices_by_user(self):
  40. yield self.store.store_device(
  41. "user_id", "device1", "display_name 1"
  42. )
  43. yield self.store.store_device(
  44. "user_id", "device2", "display_name 2"
  45. )
  46. yield self.store.store_device(
  47. "user_id2", "device3", "display_name 3"
  48. )
  49. res = yield self.store.get_devices_by_user("user_id")
  50. self.assertEqual(2, len(res.keys()))
  51. self.assertDictContainsSubset({
  52. "user_id": "user_id",
  53. "device_id": "device1",
  54. "display_name": "display_name 1",
  55. }, res["device1"])
  56. self.assertDictContainsSubset({
  57. "user_id": "user_id",
  58. "device_id": "device2",
  59. "display_name": "display_name 2",
  60. }, res["device2"])
  61. @defer.inlineCallbacks
  62. def test_update_device(self):
  63. yield self.store.store_device(
  64. "user_id", "device_id", "display_name 1"
  65. )
  66. res = yield self.store.get_device("user_id", "device_id")
  67. self.assertEqual("display_name 1", res["display_name"])
  68. # do a no-op first
  69. yield self.store.update_device(
  70. "user_id", "device_id",
  71. )
  72. res = yield self.store.get_device("user_id", "device_id")
  73. self.assertEqual("display_name 1", res["display_name"])
  74. # do the update
  75. yield self.store.update_device(
  76. "user_id", "device_id",
  77. new_display_name="display_name 2",
  78. )
  79. # check it worked
  80. res = yield self.store.get_device("user_id", "device_id")
  81. self.assertEqual("display_name 2", res["display_name"])
  82. @defer.inlineCallbacks
  83. def test_update_unknown_device(self):
  84. with self.assertRaises(synapse.api.errors.StoreError) as cm:
  85. yield self.store.update_device(
  86. "user_id", "unknown_device_id",
  87. new_display_name="display_name 2",
  88. )
  89. self.assertEqual(404, cm.exception.code)