test_registration_tokens.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723
  1. # Copyright 2021 Callum Brown
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. import random
  15. import string
  16. from typing import Optional
  17. from twisted.test.proto_helpers import MemoryReactor
  18. import synapse.rest.admin
  19. from synapse.api.errors import Codes
  20. from synapse.rest.client import login
  21. from synapse.server import HomeServer
  22. from synapse.util import Clock
  23. from tests import unittest
  24. class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase):
  25. servlets = [
  26. synapse.rest.admin.register_servlets,
  27. login.register_servlets,
  28. ]
  29. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  30. self.store = hs.get_datastores().main
  31. self.admin_user = self.register_user("admin", "pass", admin=True)
  32. self.admin_user_tok = self.login("admin", "pass")
  33. self.other_user = self.register_user("user", "pass")
  34. self.other_user_tok = self.login("user", "pass")
  35. self.url = "/_synapse/admin/v1/registration_tokens"
  36. def _new_token(
  37. self,
  38. token: Optional[str] = None,
  39. uses_allowed: Optional[int] = None,
  40. pending: int = 0,
  41. completed: int = 0,
  42. expiry_time: Optional[int] = None,
  43. ) -> str:
  44. """Helper function to create a token."""
  45. if token is None:
  46. token = "".join(random.choices(string.ascii_letters, k=8))
  47. self.get_success(
  48. self.store.db_pool.simple_insert(
  49. "registration_tokens",
  50. {
  51. "token": token,
  52. "uses_allowed": uses_allowed,
  53. "pending": pending,
  54. "completed": completed,
  55. "expiry_time": expiry_time,
  56. },
  57. )
  58. )
  59. return token
  60. # CREATION
  61. def test_create_no_auth(self) -> None:
  62. """Try to create a token without authentication."""
  63. channel = self.make_request("POST", self.url + "/new", {})
  64. self.assertEqual(401, channel.code, msg=channel.json_body)
  65. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  66. def test_create_requester_not_admin(self) -> None:
  67. """Try to create a token while not an admin."""
  68. channel = self.make_request(
  69. "POST",
  70. self.url + "/new",
  71. {},
  72. access_token=self.other_user_tok,
  73. )
  74. self.assertEqual(403, channel.code, msg=channel.json_body)
  75. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  76. def test_create_using_defaults(self) -> None:
  77. """Create a token using all the defaults."""
  78. channel = self.make_request(
  79. "POST",
  80. self.url + "/new",
  81. {},
  82. access_token=self.admin_user_tok,
  83. )
  84. self.assertEqual(200, channel.code, msg=channel.json_body)
  85. self.assertEqual(len(channel.json_body["token"]), 16)
  86. self.assertIsNone(channel.json_body["uses_allowed"])
  87. self.assertIsNone(channel.json_body["expiry_time"])
  88. self.assertEqual(channel.json_body["pending"], 0)
  89. self.assertEqual(channel.json_body["completed"], 0)
  90. def test_create_specifying_fields(self) -> None:
  91. """Create a token specifying the value of all fields."""
  92. # As many of the allowed characters as possible with length <= 64
  93. token = "adefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._~-"
  94. data = {
  95. "token": token,
  96. "uses_allowed": 1,
  97. "expiry_time": self.clock.time_msec() + 1000000,
  98. }
  99. channel = self.make_request(
  100. "POST",
  101. self.url + "/new",
  102. data,
  103. access_token=self.admin_user_tok,
  104. )
  105. self.assertEqual(200, channel.code, msg=channel.json_body)
  106. self.assertEqual(channel.json_body["token"], token)
  107. self.assertEqual(channel.json_body["uses_allowed"], 1)
  108. self.assertEqual(channel.json_body["expiry_time"], data["expiry_time"])
  109. self.assertEqual(channel.json_body["pending"], 0)
  110. self.assertEqual(channel.json_body["completed"], 0)
  111. def test_create_with_null_value(self) -> None:
  112. """Create a token specifying unlimited uses and no expiry."""
  113. data = {
  114. "uses_allowed": None,
  115. "expiry_time": None,
  116. }
  117. channel = self.make_request(
  118. "POST",
  119. self.url + "/new",
  120. data,
  121. access_token=self.admin_user_tok,
  122. )
  123. self.assertEqual(200, channel.code, msg=channel.json_body)
  124. self.assertEqual(len(channel.json_body["token"]), 16)
  125. self.assertIsNone(channel.json_body["uses_allowed"])
  126. self.assertIsNone(channel.json_body["expiry_time"])
  127. self.assertEqual(channel.json_body["pending"], 0)
  128. self.assertEqual(channel.json_body["completed"], 0)
  129. def test_create_token_too_long(self) -> None:
  130. """Check token longer than 64 chars is invalid."""
  131. data = {"token": "a" * 65}
  132. channel = self.make_request(
  133. "POST",
  134. self.url + "/new",
  135. data,
  136. access_token=self.admin_user_tok,
  137. )
  138. self.assertEqual(400, channel.code, msg=channel.json_body)
  139. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  140. def test_create_token_invalid_chars(self) -> None:
  141. """Check you can't create token with invalid characters."""
  142. data = {
  143. "token": "abc/def",
  144. }
  145. channel = self.make_request(
  146. "POST",
  147. self.url + "/new",
  148. data,
  149. access_token=self.admin_user_tok,
  150. )
  151. self.assertEqual(400, channel.code, msg=channel.json_body)
  152. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  153. def test_create_token_already_exists(self) -> None:
  154. """Check you can't create token that already exists."""
  155. data = {
  156. "token": "abcd",
  157. }
  158. channel1 = self.make_request(
  159. "POST",
  160. self.url + "/new",
  161. data,
  162. access_token=self.admin_user_tok,
  163. )
  164. self.assertEqual(200, channel1.code, msg=channel1.json_body)
  165. channel2 = self.make_request(
  166. "POST",
  167. self.url + "/new",
  168. data,
  169. access_token=self.admin_user_tok,
  170. )
  171. self.assertEqual(400, channel2.code, msg=channel2.json_body)
  172. self.assertEqual(channel2.json_body["errcode"], Codes.INVALID_PARAM)
  173. def test_create_unable_to_generate_token(self) -> None:
  174. """Check right error is raised when server can't generate unique token."""
  175. # Create all possible single character tokens
  176. tokens = []
  177. for c in string.ascii_letters + string.digits + "._~-":
  178. tokens.append((c, None, 0, 0, None))
  179. self.get_success(
  180. self.store.db_pool.simple_insert_many(
  181. "registration_tokens",
  182. keys=("token", "uses_allowed", "pending", "completed", "expiry_time"),
  183. values=tokens,
  184. desc="create_all_registration_tokens",
  185. )
  186. )
  187. # Check creating a single character token fails with a 500 status code
  188. channel = self.make_request(
  189. "POST",
  190. self.url + "/new",
  191. {"length": 1},
  192. access_token=self.admin_user_tok,
  193. )
  194. self.assertEqual(500, channel.code, msg=channel.json_body)
  195. def test_create_uses_allowed(self) -> None:
  196. """Check you can only create a token with good values for uses_allowed."""
  197. # Should work with 0 (token is invalid from the start)
  198. channel = self.make_request(
  199. "POST",
  200. self.url + "/new",
  201. {"uses_allowed": 0},
  202. access_token=self.admin_user_tok,
  203. )
  204. self.assertEqual(200, channel.code, msg=channel.json_body)
  205. self.assertEqual(channel.json_body["uses_allowed"], 0)
  206. # Should fail with negative integer
  207. channel = self.make_request(
  208. "POST",
  209. self.url + "/new",
  210. {"uses_allowed": -5},
  211. access_token=self.admin_user_tok,
  212. )
  213. self.assertEqual(
  214. 400,
  215. channel.code,
  216. msg=channel.json_body,
  217. )
  218. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  219. # Should fail with float
  220. channel = self.make_request(
  221. "POST",
  222. self.url + "/new",
  223. {"uses_allowed": 1.5},
  224. access_token=self.admin_user_tok,
  225. )
  226. self.assertEqual(400, channel.code, msg=channel.json_body)
  227. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  228. def test_create_expiry_time(self) -> None:
  229. """Check you can't create a token with an invalid expiry_time."""
  230. # Should fail with a time in the past
  231. channel = self.make_request(
  232. "POST",
  233. self.url + "/new",
  234. {"expiry_time": self.clock.time_msec() - 10000},
  235. access_token=self.admin_user_tok,
  236. )
  237. self.assertEqual(400, channel.code, msg=channel.json_body)
  238. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  239. # Should fail with float
  240. channel = self.make_request(
  241. "POST",
  242. self.url + "/new",
  243. {"expiry_time": self.clock.time_msec() + 1000000.5},
  244. access_token=self.admin_user_tok,
  245. )
  246. self.assertEqual(400, channel.code, msg=channel.json_body)
  247. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  248. def test_create_length(self) -> None:
  249. """Check you can only generate a token with a valid length."""
  250. # Should work with 64
  251. channel = self.make_request(
  252. "POST",
  253. self.url + "/new",
  254. {"length": 64},
  255. access_token=self.admin_user_tok,
  256. )
  257. self.assertEqual(200, channel.code, msg=channel.json_body)
  258. self.assertEqual(len(channel.json_body["token"]), 64)
  259. # Should fail with 0
  260. channel = self.make_request(
  261. "POST",
  262. self.url + "/new",
  263. {"length": 0},
  264. access_token=self.admin_user_tok,
  265. )
  266. self.assertEqual(400, channel.code, msg=channel.json_body)
  267. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  268. # Should fail with a negative integer
  269. channel = self.make_request(
  270. "POST",
  271. self.url + "/new",
  272. {"length": -5},
  273. access_token=self.admin_user_tok,
  274. )
  275. self.assertEqual(400, channel.code, msg=channel.json_body)
  276. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  277. # Should fail with a float
  278. channel = self.make_request(
  279. "POST",
  280. self.url + "/new",
  281. {"length": 8.5},
  282. access_token=self.admin_user_tok,
  283. )
  284. self.assertEqual(400, channel.code, msg=channel.json_body)
  285. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  286. # Should fail with 65
  287. channel = self.make_request(
  288. "POST",
  289. self.url + "/new",
  290. {"length": 65},
  291. access_token=self.admin_user_tok,
  292. )
  293. self.assertEqual(400, channel.code, msg=channel.json_body)
  294. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  295. # UPDATING
  296. def test_update_no_auth(self) -> None:
  297. """Try to update a token without authentication."""
  298. channel = self.make_request(
  299. "PUT",
  300. self.url + "/1234", # Token doesn't exist but that doesn't matter
  301. {},
  302. )
  303. self.assertEqual(401, channel.code, msg=channel.json_body)
  304. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  305. def test_update_requester_not_admin(self) -> None:
  306. """Try to update a token while not an admin."""
  307. channel = self.make_request(
  308. "PUT",
  309. self.url + "/1234", # Token doesn't exist but that doesn't matter
  310. {},
  311. access_token=self.other_user_tok,
  312. )
  313. self.assertEqual(403, channel.code, msg=channel.json_body)
  314. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  315. def test_update_non_existent(self) -> None:
  316. """Try to update a token that doesn't exist."""
  317. channel = self.make_request(
  318. "PUT",
  319. self.url + "/1234",
  320. {"uses_allowed": 1},
  321. access_token=self.admin_user_tok,
  322. )
  323. self.assertEqual(404, channel.code, msg=channel.json_body)
  324. self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND)
  325. def test_update_uses_allowed(self) -> None:
  326. """Test updating just uses_allowed."""
  327. # Create new token using default values
  328. token = self._new_token()
  329. # Should succeed with 1
  330. channel = self.make_request(
  331. "PUT",
  332. self.url + "/" + token,
  333. {"uses_allowed": 1},
  334. access_token=self.admin_user_tok,
  335. )
  336. self.assertEqual(200, channel.code, msg=channel.json_body)
  337. self.assertEqual(channel.json_body["uses_allowed"], 1)
  338. self.assertIsNone(channel.json_body["expiry_time"])
  339. # Should succeed with 0 (makes token invalid)
  340. channel = self.make_request(
  341. "PUT",
  342. self.url + "/" + token,
  343. {"uses_allowed": 0},
  344. access_token=self.admin_user_tok,
  345. )
  346. self.assertEqual(200, channel.code, msg=channel.json_body)
  347. self.assertEqual(channel.json_body["uses_allowed"], 0)
  348. self.assertIsNone(channel.json_body["expiry_time"])
  349. # Should succeed with null
  350. channel = self.make_request(
  351. "PUT",
  352. self.url + "/" + token,
  353. {"uses_allowed": None},
  354. access_token=self.admin_user_tok,
  355. )
  356. self.assertEqual(200, channel.code, msg=channel.json_body)
  357. self.assertIsNone(channel.json_body["uses_allowed"])
  358. self.assertIsNone(channel.json_body["expiry_time"])
  359. # Should fail with a float
  360. channel = self.make_request(
  361. "PUT",
  362. self.url + "/" + token,
  363. {"uses_allowed": 1.5},
  364. access_token=self.admin_user_tok,
  365. )
  366. self.assertEqual(400, channel.code, msg=channel.json_body)
  367. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  368. # Should fail with a negative integer
  369. channel = self.make_request(
  370. "PUT",
  371. self.url + "/" + token,
  372. {"uses_allowed": -5},
  373. access_token=self.admin_user_tok,
  374. )
  375. self.assertEqual(400, channel.code, msg=channel.json_body)
  376. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  377. def test_update_expiry_time(self) -> None:
  378. """Test updating just expiry_time."""
  379. # Create new token using default values
  380. token = self._new_token()
  381. new_expiry_time = self.clock.time_msec() + 1000000
  382. # Should succeed with a time in the future
  383. channel = self.make_request(
  384. "PUT",
  385. self.url + "/" + token,
  386. {"expiry_time": new_expiry_time},
  387. access_token=self.admin_user_tok,
  388. )
  389. self.assertEqual(200, channel.code, msg=channel.json_body)
  390. self.assertEqual(channel.json_body["expiry_time"], new_expiry_time)
  391. self.assertIsNone(channel.json_body["uses_allowed"])
  392. # Should succeed with null
  393. channel = self.make_request(
  394. "PUT",
  395. self.url + "/" + token,
  396. {"expiry_time": None},
  397. access_token=self.admin_user_tok,
  398. )
  399. self.assertEqual(200, channel.code, msg=channel.json_body)
  400. self.assertIsNone(channel.json_body["expiry_time"])
  401. self.assertIsNone(channel.json_body["uses_allowed"])
  402. # Should fail with a time in the past
  403. past_time = self.clock.time_msec() - 10000
  404. channel = self.make_request(
  405. "PUT",
  406. self.url + "/" + token,
  407. {"expiry_time": past_time},
  408. access_token=self.admin_user_tok,
  409. )
  410. self.assertEqual(400, channel.code, msg=channel.json_body)
  411. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  412. # Should fail a float
  413. channel = self.make_request(
  414. "PUT",
  415. self.url + "/" + token,
  416. {"expiry_time": new_expiry_time + 0.5},
  417. access_token=self.admin_user_tok,
  418. )
  419. self.assertEqual(400, channel.code, msg=channel.json_body)
  420. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  421. def test_update_both(self) -> None:
  422. """Test updating both uses_allowed and expiry_time."""
  423. # Create new token using default values
  424. token = self._new_token()
  425. new_expiry_time = self.clock.time_msec() + 1000000
  426. data = {
  427. "uses_allowed": 1,
  428. "expiry_time": new_expiry_time,
  429. }
  430. channel = self.make_request(
  431. "PUT",
  432. self.url + "/" + token,
  433. data,
  434. access_token=self.admin_user_tok,
  435. )
  436. self.assertEqual(200, channel.code, msg=channel.json_body)
  437. self.assertEqual(channel.json_body["uses_allowed"], 1)
  438. self.assertEqual(channel.json_body["expiry_time"], new_expiry_time)
  439. def test_update_invalid_type(self) -> None:
  440. """Test using invalid types doesn't work."""
  441. # Create new token using default values
  442. token = self._new_token()
  443. data = {
  444. "uses_allowed": False,
  445. "expiry_time": "1626430124000",
  446. }
  447. channel = self.make_request(
  448. "PUT",
  449. self.url + "/" + token,
  450. data,
  451. access_token=self.admin_user_tok,
  452. )
  453. self.assertEqual(400, channel.code, msg=channel.json_body)
  454. self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM)
  455. # DELETING
  456. def test_delete_no_auth(self) -> None:
  457. """Try to delete a token without authentication."""
  458. channel = self.make_request(
  459. "DELETE",
  460. self.url + "/1234", # Token doesn't exist but that doesn't matter
  461. {},
  462. )
  463. self.assertEqual(401, channel.code, msg=channel.json_body)
  464. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  465. def test_delete_requester_not_admin(self) -> None:
  466. """Try to delete a token while not an admin."""
  467. channel = self.make_request(
  468. "DELETE",
  469. self.url + "/1234", # Token doesn't exist but that doesn't matter
  470. {},
  471. access_token=self.other_user_tok,
  472. )
  473. self.assertEqual(403, channel.code, msg=channel.json_body)
  474. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  475. def test_delete_non_existent(self) -> None:
  476. """Try to delete a token that doesn't exist."""
  477. channel = self.make_request(
  478. "DELETE",
  479. self.url + "/1234",
  480. {},
  481. access_token=self.admin_user_tok,
  482. )
  483. self.assertEqual(404, channel.code, msg=channel.json_body)
  484. self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND)
  485. def test_delete(self) -> None:
  486. """Test deleting a token."""
  487. # Create new token using default values
  488. token = self._new_token()
  489. channel = self.make_request(
  490. "DELETE",
  491. self.url + "/" + token,
  492. {},
  493. access_token=self.admin_user_tok,
  494. )
  495. self.assertEqual(200, channel.code, msg=channel.json_body)
  496. # GETTING ONE
  497. def test_get_no_auth(self) -> None:
  498. """Try to get a token without authentication."""
  499. channel = self.make_request(
  500. "GET",
  501. self.url + "/1234", # Token doesn't exist but that doesn't matter
  502. {},
  503. )
  504. self.assertEqual(401, channel.code, msg=channel.json_body)
  505. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  506. def test_get_requester_not_admin(self) -> None:
  507. """Try to get a token while not an admin."""
  508. channel = self.make_request(
  509. "GET",
  510. self.url + "/1234", # Token doesn't exist but that doesn't matter
  511. {},
  512. access_token=self.other_user_tok,
  513. )
  514. self.assertEqual(
  515. 403,
  516. channel.code,
  517. msg=channel.json_body,
  518. )
  519. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  520. def test_get_non_existent(self) -> None:
  521. """Try to get a token that doesn't exist."""
  522. channel = self.make_request(
  523. "GET",
  524. self.url + "/1234",
  525. {},
  526. access_token=self.admin_user_tok,
  527. )
  528. self.assertEqual(404, channel.code, msg=channel.json_body)
  529. self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND)
  530. def test_get(self) -> None:
  531. """Test getting a token."""
  532. # Create new token using default values
  533. token = self._new_token()
  534. channel = self.make_request(
  535. "GET",
  536. self.url + "/" + token,
  537. {},
  538. access_token=self.admin_user_tok,
  539. )
  540. self.assertEqual(200, channel.code, msg=channel.json_body)
  541. self.assertEqual(channel.json_body["token"], token)
  542. self.assertIsNone(channel.json_body["uses_allowed"])
  543. self.assertIsNone(channel.json_body["expiry_time"])
  544. self.assertEqual(channel.json_body["pending"], 0)
  545. self.assertEqual(channel.json_body["completed"], 0)
  546. # LISTING
  547. def test_list_no_auth(self) -> None:
  548. """Try to list tokens without authentication."""
  549. channel = self.make_request("GET", self.url, {})
  550. self.assertEqual(401, channel.code, msg=channel.json_body)
  551. self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"])
  552. def test_list_requester_not_admin(self) -> None:
  553. """Try to list tokens while not an admin."""
  554. channel = self.make_request(
  555. "GET",
  556. self.url,
  557. {},
  558. access_token=self.other_user_tok,
  559. )
  560. self.assertEqual(403, channel.code, msg=channel.json_body)
  561. self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"])
  562. def test_list_all(self) -> None:
  563. """Test listing all tokens."""
  564. # Create new token using default values
  565. token = self._new_token()
  566. channel = self.make_request(
  567. "GET",
  568. self.url,
  569. {},
  570. access_token=self.admin_user_tok,
  571. )
  572. self.assertEqual(200, channel.code, msg=channel.json_body)
  573. self.assertEqual(len(channel.json_body["registration_tokens"]), 1)
  574. token_info = channel.json_body["registration_tokens"][0]
  575. self.assertEqual(token_info["token"], token)
  576. self.assertIsNone(token_info["uses_allowed"])
  577. self.assertIsNone(token_info["expiry_time"])
  578. self.assertEqual(token_info["pending"], 0)
  579. self.assertEqual(token_info["completed"], 0)
  580. def test_list_invalid_query_parameter(self) -> None:
  581. """Test with `valid` query parameter not `true` or `false`."""
  582. channel = self.make_request(
  583. "GET",
  584. self.url + "?valid=x",
  585. {},
  586. access_token=self.admin_user_tok,
  587. )
  588. self.assertEqual(400, channel.code, msg=channel.json_body)
  589. def _test_list_query_parameter(self, valid: str) -> None:
  590. """Helper used to test both valid=true and valid=false."""
  591. # Create 2 valid and 2 invalid tokens.
  592. now = self.hs.get_clock().time_msec()
  593. # Create always valid token
  594. valid1 = self._new_token()
  595. # Create token that hasn't been used up
  596. valid2 = self._new_token(uses_allowed=1)
  597. # Create token that has expired
  598. invalid1 = self._new_token(expiry_time=now - 10000)
  599. # Create token that has been used up but hasn't expired
  600. invalid2 = self._new_token(
  601. uses_allowed=2,
  602. pending=1,
  603. completed=1,
  604. expiry_time=now + 1000000,
  605. )
  606. if valid == "true":
  607. tokens = [valid1, valid2]
  608. else:
  609. tokens = [invalid1, invalid2]
  610. channel = self.make_request(
  611. "GET",
  612. self.url + "?valid=" + valid,
  613. {},
  614. access_token=self.admin_user_tok,
  615. )
  616. self.assertEqual(200, channel.code, msg=channel.json_body)
  617. self.assertEqual(len(channel.json_body["registration_tokens"]), 2)
  618. token_info_1 = channel.json_body["registration_tokens"][0]
  619. token_info_2 = channel.json_body["registration_tokens"][1]
  620. self.assertIn(token_info_1["token"], tokens)
  621. self.assertIn(token_info_2["token"], tokens)
  622. def test_list_valid(self) -> None:
  623. """Test listing just valid tokens."""
  624. self._test_list_query_parameter(valid="true")
  625. def test_list_invalid(self) -> None:
  626. """Test listing just invalid tokens."""
  627. self._test_list_query_parameter(valid="false")