Add type hints to synapse/tests/rest/admin

mypy.ini

@@ -77,9 +77,6 @@ exclude = (?x)
    |tests/push/test_http.py
    |tests/push/test_presentable_names.py
    |tests/push/test_push_rule_evaluator.py
-   |tests/rest/admin/test_admin.py
-   |tests/rest/admin/test_user.py
-   |tests/rest/admin/test_username_available.py
    |tests/rest/client/test_account.py
    |tests/rest/client/test_events.py
    |tests/rest/client/test_filter.py

tests/rest/admin/test_admin.py

@@ -15,15 +15,16 @@
 import os
 import urllib.parse
 from http import HTTPStatus
-from unittest.mock import Mock
+from typing import List, Optional
 
-from twisted.internet.defer import Deferred
+from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
 from synapse.http.server import JsonResource
-from synapse.logging.context import make_deferred_yieldable
 from synapse.rest.admin import VersionServlet
 from synapse.rest.client import groups, login, room
+from synapse.server import HomeServer
+from synapse.util import Clock
 
 from tests import unittest
 from tests.server import FakeSite, make_request
@@ -33,12 +34,12 @@
 class VersionTestCase(unittest.HomeserverTestCase):
     url = "/_synapse/admin/v1/server_version"
 
-    def create_test_resource(self):
+    def create_test_resource(self) -> JsonResource:
         resource = JsonResource(self.hs)
         VersionServlet(self.hs).register(resource)
         return resource
 
-    def test_version_string(self):
+    def test_version_string(self) -> None:
         channel = self.make_request("GET", self.url, shorthand=False)
 
         self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
@@ -54,14 +55,14 @@ class DeleteGroupTestCase(unittest.HomeserverTestCase):
         groups.register_servlets,
     ]
 
-    def prepare(self, reactor, clock, hs):
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.admin_user = self.register_user("admin", "pass", admin=True)
         self.admin_user_tok = self.login("admin", "pass")
 
         self.other_user = self.register_user("user", "pass")
         self.other_user_token = self.login("user", "pass")
 
-    def test_delete_group(self):
+    def test_delete_group(self) -> None:
         # Create a new group
         channel = self.make_request(
             "POST",
@@ -112,7 +113,7 @@ def test_delete_group(self):
         self.assertNotIn(group_id, self._get_groups_user_is_in(self.admin_user_tok))
         self.assertNotIn(group_id, self._get_groups_user_is_in(self.other_user_token))
 
-    def _check_group(self, group_id, expect_code):
+    def _check_group(self, group_id: str, expect_code: int) -> None:
         """Assert that trying to fetch the given group results in the given
         HTTP status code
         """
@@ -124,7 +125,7 @@ def _check_group(self, group_id, expect_code):
 
         self.assertEqual(expect_code, channel.code, msg=channel.json_body)
 
-    def _get_groups_user_is_in(self, access_token):
+    def _get_groups_user_is_in(self, access_token: str) -> List[str]:
         """Returns the list of groups the user is in (given their access token)"""
         channel = self.make_request("GET", b"/joined_groups", access_token=access_token)
 
@@ -143,34 +144,13 @@ class QuarantineMediaTestCase(unittest.HomeserverTestCase):
         room.register_servlets,
     ]
 
-    def prepare(self, reactor, clock, hs):
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         # Allow for uploading and downloading to/from the media repo
         self.media_repo = hs.get_media_repository_resource()
         self.download_resource = self.media_repo.children[b"download"]
         self.upload_resource = self.media_repo.children[b"upload"]
 
-    def make_homeserver(self, reactor, clock):
-
-        self.fetches = []
-
-        async def get_file(destination, path, output_stream, args=None, max_size=None):
-            """
-            Returns tuple[int,dict,str,int] of file length, response headers,
-            absolute URI, and response code.
-            """
-
-            def write_to(r):
-                data, response = r
-                output_stream.write(data)
-                return response
-
-            d = Deferred()
-            d.addCallback(write_to)
-            self.fetches.append((d, destination, path, args))
-            return await make_deferred_yieldable(d)
-
-        client = Mock()
-        client.get_file = get_file
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
 
         self.storage_path = self.mktemp()
         self.media_store_path = self.mktemp()
@@ -191,11 +171,13 @@ def write_to(r):
         }
         config["media_storage_providers"] = [provider_config]
 
-        hs = self.setup_test_homeserver(config=config, federation_http_client=client)
+        hs = self.setup_test_homeserver(config=config)
 
         return hs
 
-    def _ensure_quarantined(self, admin_user_tok, server_and_media_id):
+    def _ensure_quarantined(
+        self, admin_user_tok: str, server_and_media_id: str
+    ) -> None:
         """Ensure a piece of media is quarantined when trying to access it."""
         channel = make_request(
             self.reactor,
@@ -216,7 +198,7 @@ def _ensure_quarantined(self, admin_user_tok, server_and_media_id):
             ),
         )
 
-    def test_quarantine_media_requires_admin(self):
+    def test_quarantine_media_requires_admin(self) -> None:
         self.register_user("nonadmin", "pass", admin=False)
         non_admin_user_tok = self.login("nonadmin", "pass")
 
@@ -250,7 +232,7 @@ def test_quarantine_media_requires_admin(self):
             msg="Expected forbidden on quarantining media as a non-admin",
         )
 
-    def test_quarantine_media_by_id(self):
+    def test_quarantine_media_by_id(self) -> None:
         self.register_user("id_admin", "pass", admin=True)
         admin_user_tok = self.login("id_admin", "pass")
 
@@ -295,7 +277,9 @@ def test_quarantine_media_by_id(self):
         # Attempt to access the media
         self._ensure_quarantined(admin_user_tok, server_name_and_media_id)
 
-    def test_quarantine_all_media_in_room(self, override_url_template=None):
+    def test_quarantine_all_media_in_room(
+        self, override_url_template: Optional[str] = None
+    ) -> None:
         self.register_user("room_admin", "pass", admin=True)
         admin_user_tok = self.login("room_admin", "pass")
 
@@ -359,11 +343,11 @@ def test_quarantine_all_media_in_room(self, override_url_template=None):
         self._ensure_quarantined(admin_user_tok, server_and_media_id_1)
         self._ensure_quarantined(admin_user_tok, server_and_media_id_2)
 
-    def test_quarantine_all_media_in_room_deprecated_api_path(self):
+    def test_quarantine_all_media_in_room_deprecated_api_path(self) -> None:
         # Perform the above test with the deprecated API path
         self.test_quarantine_all_media_in_room("/_synapse/admin/v1/quarantine_media/%s")
 
-    def test_quarantine_all_media_by_user(self):
+    def test_quarantine_all_media_by_user(self) -> None:
         self.register_user("user_admin", "pass", admin=True)
         admin_user_tok = self.login("user_admin", "pass")
 
@@ -401,7 +385,7 @@ def test_quarantine_all_media_by_user(self):
         self._ensure_quarantined(admin_user_tok, server_and_media_id_1)
         self._ensure_quarantined(admin_user_tok, server_and_media_id_2)
 
-    def test_cannot_quarantine_safe_media(self):
+    def test_cannot_quarantine_safe_media(self) -> None:
         self.register_user("user_admin", "pass", admin=True)
         admin_user_tok = self.login("user_admin", "pass")
 
@@ -475,7 +459,7 @@ class PurgeHistoryTestCase(unittest.HomeserverTestCase):
         room.register_servlets,
     ]
 
-    def prepare(self, reactor, clock, hs):
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.admin_user = self.register_user("admin", "pass", admin=True)
         self.admin_user_tok = self.login("admin", "pass")
 
@@ -488,7 +472,7 @@ def prepare(self, reactor, clock, hs):
         self.url = f"/_synapse/admin/v1/purge_history/{self.room_id}"
         self.url_status = "/_synapse/admin/v1/purge_history_status/"
 
-    def test_purge_history(self):
+    def test_purge_history(self) -> None:
         """
         Simple test of purge history API.
         Test only that is is possible to call, get status HTTPStatus.OK and purge_id.