Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-07-21 21:48:47 -04:00
Code Issues Projects Releases Packages Wiki Activity

Add type hints to synapse/tests/rest/admin (#11501)

Browse source
This commit is contained in:
Dirk Klimpel 2021-12-03 14:57:13 +01:00 • committed by GitHub
parent 8cd68b8102
commit e5f426cd54
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 257 additions and 228 deletions
Download patch file Download diff file Expand all files Collapse all files

77
tests/rest/admin/test_registration_tokens.py
View file

@ -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")