1
0

test_registration.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2014-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 tests import unittest
  16. from twisted.internet import defer
  17. from synapse.api.errors import StoreError
  18. from synapse.util import stringutils
  19. from tests.utils import setup_test_homeserver
  20. class RegistrationStoreTestCase(unittest.TestCase):
  21. @defer.inlineCallbacks
  22. def setUp(self):
  23. hs = yield setup_test_homeserver()
  24. self.db_pool = hs.get_db_pool()
  25. self.store = hs.get_datastore()
  26. self.user_id = "@my-user:test"
  27. self.tokens = [
  28. "AbCdEfGhIjKlMnOpQrStUvWxYz",
  29. "BcDeFgHiJkLmNoPqRsTuVwXyZa"
  30. ]
  31. self.pwhash = "{xx1}123456789"
  32. @defer.inlineCallbacks
  33. def test_register(self):
  34. yield self.store.register(self.user_id, self.tokens[0], self.pwhash)
  35. self.assertEquals(
  36. # TODO(paul): Surely this field should be 'user_id', not 'name'
  37. # Additionally surely it shouldn't come in a 1-element list
  38. {"name": self.user_id, "password_hash": self.pwhash, "is_guest": 0},
  39. (yield self.store.get_user_by_id(self.user_id))
  40. )
  41. result = yield self.store.get_user_by_access_token(self.tokens[0])
  42. self.assertDictContainsSubset(
  43. {
  44. "name": self.user_id,
  45. },
  46. result
  47. )
  48. self.assertTrue("token_id" in result)
  49. @defer.inlineCallbacks
  50. def test_add_tokens(self):
  51. yield self.store.register(self.user_id, self.tokens[0], self.pwhash)
  52. yield self.store.add_access_token_to_user(self.user_id, self.tokens[1])
  53. result = yield self.store.get_user_by_access_token(self.tokens[1])
  54. self.assertDictContainsSubset(
  55. {
  56. "name": self.user_id,
  57. },
  58. result
  59. )
  60. self.assertTrue("token_id" in result)
  61. @defer.inlineCallbacks
  62. def test_exchange_refresh_token_valid(self):
  63. uid = stringutils.random_string(32)
  64. generator = TokenGenerator()
  65. last_token = generator.generate(uid)
  66. self.db_pool.runQuery(
  67. "INSERT INTO refresh_tokens(user_id, token) VALUES(?,?)",
  68. (uid, last_token,))
  69. (found_user_id, refresh_token) = yield self.store.exchange_refresh_token(
  70. last_token, generator.generate)
  71. self.assertEqual(uid, found_user_id)
  72. rows = yield self.db_pool.runQuery(
  73. "SELECT token FROM refresh_tokens WHERE user_id = ?", (uid, ))
  74. self.assertEqual([(refresh_token,)], rows)
  75. # We issued token 1, then exchanged it for token 2
  76. expected_refresh_token = u"%s-%d" % (uid, 2,)
  77. self.assertEqual(expected_refresh_token, refresh_token)
  78. @defer.inlineCallbacks
  79. def test_exchange_refresh_token_none(self):
  80. uid = stringutils.random_string(32)
  81. generator = TokenGenerator()
  82. last_token = generator.generate(uid)
  83. with self.assertRaises(StoreError):
  84. yield self.store.exchange_refresh_token(last_token, generator.generate)
  85. @defer.inlineCallbacks
  86. def test_exchange_refresh_token_invalid(self):
  87. uid = stringutils.random_string(32)
  88. generator = TokenGenerator()
  89. last_token = generator.generate(uid)
  90. wrong_token = "%s-wrong" % (last_token,)
  91. self.db_pool.runQuery(
  92. "INSERT INTO refresh_tokens(user_id, token) VALUES(?,?)",
  93. (uid, wrong_token,))
  94. with self.assertRaises(StoreError):
  95. yield self.store.exchange_refresh_token(last_token, generator.generate)
  96. class TokenGenerator:
  97. def __init__(self):
  98. self._last_issued_token = 0
  99. def generate(self, user_id):
  100. self._last_issued_token += 1
  101. return u"%s-%d" % (user_id, self._last_issued_token,)