Add type hints to synapse/tests/rest/admin

changelog.d/11851.misc

@@ -0,0 +1 @@
+Add type hints to `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

@@ -12,18 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 import urllib.parse
 from http import HTTPStatus
-from unittest.mock import Mock
+from typing import List
 
-from twisted.internet.defer import Deferred
+from parameterized import parameterized
+
+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 +35,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 +56,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 +114,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 +126,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,59 +145,15 @@ 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
-
-        self.storage_path = self.mktemp()
-        self.media_store_path = self.mktemp()
-        os.mkdir(self.storage_path)
-        os.mkdir(self.media_store_path)
-
-        config = self.default_config()
-        config["media_store_path"] = self.media_store_path
-        config["thumbnail_requirements"] = {}
-        config["max_image_pixels"] = 2000000
-
-        provider_config = {
-            "module": "synapse.rest.media.v1.storage_provider.FileStorageProviderBackend",
-            "store_local": True,
-            "store_synchronous": False,
-            "store_remote": True,
-            "config": {"directory": self.storage_path},
-        }
-        config["media_storage_providers"] = [provider_config]
-
-        hs = self.setup_test_homeserver(config=config, federation_http_client=client)
-
-        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,12 +174,18 @@ def _ensure_quarantined(self, admin_user_tok, server_and_media_id):
             ),
         )
 
-    def test_quarantine_media_requires_admin(self):
+    @parameterized.expand(
+        [
+            # Attempt quarantine media APIs as non-admin
+            "/_synapse/admin/v1/media/quarantine/example.org/abcde12345",
+            # And the roomID/userID endpoint
+            "/_synapse/admin/v1/room/!room%3Aexample.com/media/quarantine",
+        ]
+    )
+    def test_quarantine_media_requires_admin(self, url: str) -> None:
         self.register_user("nonadmin", "pass", admin=False)
         non_admin_user_tok = self.login("nonadmin", "pass")
 
-        # Attempt quarantine media APIs as non-admin
-        url = "/_synapse/admin/v1/media/quarantine/example.org/abcde12345"
         channel = self.make_request(
             "POST",
             url.encode("ascii"),
@@ -235,22 +199,7 @@ def test_quarantine_media_requires_admin(self):
             msg="Expected forbidden on quarantining media as a non-admin",
         )
 
-        # And the roomID/userID endpoint
-        url = "/_synapse/admin/v1/room/!room%3Aexample.com/media/quarantine"
-        channel = self.make_request(
-            "POST",
-            url.encode("ascii"),
-            access_token=non_admin_user_tok,
-        )
-
-        # Expect a forbidden error
-        self.assertEqual(
-            HTTPStatus.FORBIDDEN,
-            channel.code,
-            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 +244,15 @@ 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):
+    @parameterized.expand(
+        [
+            # regular API path
+            "/_synapse/admin/v1/room/%s/media/quarantine",
+            # deprecated API path
+            "/_synapse/admin/v1/quarantine_media/%s",
+        ]
+    )
+    def test_quarantine_all_media_in_room(self, url: str) -> None:
         self.register_user("room_admin", "pass", admin=True)
         admin_user_tok = self.login("room_admin", "pass")
 
@@ -333,16 +290,9 @@ def test_quarantine_all_media_in_room(self, override_url_template=None):
             tok=non_admin_user_tok,
         )
 
-        # Quarantine all media in the room
-        if override_url_template:
-            url = override_url_template % urllib.parse.quote(room_id)
-        else:
-            url = "/_synapse/admin/v1/room/%s/media/quarantine" % urllib.parse.quote(
-                room_id
-            )
         channel = self.make_request(
             "POST",
-            url,
+            url % urllib.parse.quote(room_id),
             access_token=admin_user_tok,
         )
         self.pump(1.0)
@@ -359,11 +309,7 @@ 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):
-        # 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 +347,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 +421,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 +434,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.