diff options
Diffstat (limited to 'tests/rest/admin/test_registration_tokens.py')
| -rw-r--r-- | tests/rest/admin/test_registration_tokens.py | 77 |
1 files changed, 40 insertions, 37 deletions
diff --git a/tests/rest/admin/test_registration_tokens.py b/tests/rest/admin/test_registration_tokens.py index 63087955f2..350a62dda6 100644 --- a/tests/rest/admin/test_registration_tokens.py +++ b/tests/rest/admin/test_registration_tokens.py @@ -11,14 +11,17 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - import random import string from http import HTTPStatus +from twisted.test.proto_helpers import MemoryReactor + import synapse.rest.admin from synapse.api.errors import Codes from synapse.rest.client import login +from synapse.server import HomeServer +from synapse.util import Clock from tests import unittest @@ -29,7 +32,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): login.register_servlets, ] - def prepare(self, reactor, clock, hs): + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.store = hs.get_datastore() self.admin_user = self.register_user("admin", "pass", admin=True) self.admin_user_tok = self.login("admin", "pass") @@ -39,7 +42,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.url = "/_synapse/admin/v1/registration_tokens" - def _new_token(self, **kwargs): + def _new_token(self, **kwargs) -> str: """Helper function to create a token.""" token = kwargs.get( "token", @@ -61,7 +64,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): # CREATION - def test_create_no_auth(self): + def test_create_no_auth(self) -> None: """Try to create a token without authentication.""" channel = self.make_request("POST", self.url + "/new", {}) self.assertEqual( @@ -71,7 +74,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) - def test_create_requester_not_admin(self): + def test_create_requester_not_admin(self) -> None: """Try to create a token while not an admin.""" channel = self.make_request( "POST", @@ -86,7 +89,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) - def test_create_using_defaults(self): + def test_create_using_defaults(self) -> None: """Create a token using all the defaults.""" channel = self.make_request( "POST", @@ -102,7 +105,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertEqual(channel.json_body["pending"], 0) self.assertEqual(channel.json_body["completed"], 0) - def test_create_specifying_fields(self): + def test_create_specifying_fields(self) -> None: """Create a token specifying the value of all fields.""" # As many of the allowed characters as possible with length <= 64 token = "adefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._~-" @@ -126,7 +129,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertEqual(channel.json_body["pending"], 0) self.assertEqual(channel.json_body["completed"], 0) - def test_create_with_null_value(self): + def test_create_with_null_value(self) -> None: """Create a token specifying unlimited uses and no expiry.""" data = { "uses_allowed": None, @@ -147,7 +150,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertEqual(channel.json_body["pending"], 0) self.assertEqual(channel.json_body["completed"], 0) - def test_create_token_too_long(self): + def test_create_token_too_long(self) -> None: """Check token longer than 64 chars is invalid.""" data = {"token": "a" * 65} @@ -165,7 +168,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM) - def test_create_token_invalid_chars(self): + def test_create_token_invalid_chars(self) -> None: """Check you can't create token with invalid characters.""" data = { "token": "abc/def", @@ -185,7 +188,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM) - def test_create_token_already_exists(self): + def test_create_token_already_exists(self) -> None: """Check you can't create token that already exists.""" data = { "token": "abcd", @@ -208,7 +211,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertEqual(HTTPStatus.BAD_REQUEST, channel2.code, msg=channel2.json_body) self.assertEqual(channel2.json_body["errcode"], Codes.INVALID_PARAM) - def test_create_unable_to_generate_token(self): + def test_create_unable_to_generate_token(self) -> None: """Check right error is raised when server can't generate unique token.""" # Create all possible single character tokens tokens = [] @@ -239,7 +242,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(500, channel.code, msg=channel.json_body) - def test_create_uses_allowed(self): + def test_create_uses_allowed(self) -> None: """Check you can only create a token with good values for uses_allowed.""" # Should work with 0 (token is invalid from the start) channel = self.make_request( @@ -279,7 +282,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM) - def test_create_expiry_time(self): + def test_create_expiry_time(self) -> None: """Check you can't create a token with an invalid expiry_time.""" # Should fail with a time in the past channel = self.make_request( @@ -309,7 +312,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM) - def test_create_length(self): + def test_create_length(self) -> None: """Check you can only generate a token with a valid length.""" # Should work with 64 channel = self.make_request( @@ -379,7 +382,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): # UPDATING - def test_update_no_auth(self): + def test_update_no_auth(self) -> None: """Try to update a token without authentication.""" channel = self.make_request( "PUT", @@ -393,7 +396,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) - def test_update_requester_not_admin(self): + def test_update_requester_not_admin(self) -> None: """Try to update a token while not an admin.""" channel = self.make_request( "PUT", @@ -408,7 +411,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) - def test_update_non_existent(self): + def test_update_non_existent(self) -> None: """Try to update a token that doesn't exist.""" channel = self.make_request( "PUT", @@ -424,7 +427,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND) - def test_update_uses_allowed(self): + def test_update_uses_allowed(self) -> None: """Test updating just uses_allowed.""" # Create new token using default values token = self._new_token() @@ -490,7 +493,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM) - def test_update_expiry_time(self): + def test_update_expiry_time(self) -> None: """Test updating just expiry_time.""" # Create new token using default values token = self._new_token() @@ -547,7 +550,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.INVALID_PARAM) - def test_update_both(self): + def test_update_both(self) -> None: """Test updating both uses_allowed and expiry_time.""" # Create new token using default values token = self._new_token() @@ -569,7 +572,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertEqual(channel.json_body["uses_allowed"], 1) self.assertEqual(channel.json_body["expiry_time"], new_expiry_time) - def test_update_invalid_type(self): + def test_update_invalid_type(self) -> None: """Test using invalid types doesn't work.""" # Create new token using default values token = self._new_token() @@ -595,7 +598,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): # DELETING - def test_delete_no_auth(self): + def test_delete_no_auth(self) -> None: """Try to delete a token without authentication.""" channel = self.make_request( "DELETE", @@ -609,7 +612,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) - def test_delete_requester_not_admin(self): + def test_delete_requester_not_admin(self) -> None: """Try to delete a token while not an admin.""" channel = self.make_request( "DELETE", @@ -624,7 +627,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) - def test_delete_non_existent(self): + def test_delete_non_existent(self) -> None: """Try to delete a token that doesn't exist.""" channel = self.make_request( "DELETE", @@ -640,7 +643,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND) - def test_delete(self): + def test_delete(self) -> None: """Test deleting a token.""" # Create new token using default values token = self._new_token() @@ -656,7 +659,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): # GETTING ONE - def test_get_no_auth(self): + def test_get_no_auth(self) -> None: """Try to get a token without authentication.""" channel = self.make_request( "GET", @@ -670,7 +673,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) - def test_get_requester_not_admin(self): + def test_get_requester_not_admin(self) -> None: """Try to get a token while not an admin.""" channel = self.make_request( "GET", @@ -685,7 +688,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) - def test_get_non_existent(self): + def test_get_non_existent(self) -> None: """Try to get a token that doesn't exist.""" channel = self.make_request( "GET", @@ -701,7 +704,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(channel.json_body["errcode"], Codes.NOT_FOUND) - def test_get(self): + def test_get(self) -> None: """Test getting a token.""" # Create new token using default values token = self._new_token() @@ -722,7 +725,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): # LISTING - def test_list_no_auth(self): + def test_list_no_auth(self) -> None: """Try to list tokens without authentication.""" channel = self.make_request("GET", self.url, {}) self.assertEqual( @@ -732,7 +735,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) - def test_list_requester_not_admin(self): + def test_list_requester_not_admin(self) -> None: """Try to list tokens while not an admin.""" channel = self.make_request( "GET", @@ -747,7 +750,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): ) self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) - def test_list_all(self): + def test_list_all(self) -> None: """Test listing all tokens.""" # Create new token using default values token = self._new_token() @@ -768,7 +771,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertEqual(token_info["pending"], 0) self.assertEqual(token_info["completed"], 0) - def test_list_invalid_query_parameter(self): + def test_list_invalid_query_parameter(self) -> None: """Test with `valid` query parameter not `true` or `false`.""" channel = self.make_request( "GET", @@ -783,7 +786,7 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): msg=channel.json_body, ) - def _test_list_query_parameter(self, valid: str): + def _test_list_query_parameter(self, valid: str) -> None: """Helper used to test both valid=true and valid=false.""" # Create 2 valid and 2 invalid tokens. now = self.hs.get_clock().time_msec() @@ -820,10 +823,10 @@ class ManageRegistrationTokensTestCase(unittest.HomeserverTestCase): self.assertIn(token_info_1["token"], tokens) self.assertIn(token_info_2["token"], tokens) - def test_list_valid(self): + def test_list_valid(self) -> None: """Test listing just valid tokens.""" self._test_list_query_parameter(valid="true") - def test_list_invalid(self): + def test_list_invalid(self) -> None: """Test listing just invalid tokens.""" self._test_list_query_parameter(valid="false") |