Convert status codes to HTTPStatus in tests.rest.admin

changelog.d/11455.misc

@@ -0,0 +1 @@
+Convert status codes to `HTTPStatus` in `synapse.rest.admin`.

tests/rest/admin/test_admin.py

@@ -12,9 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json
 import os
 import urllib.parse
+from http import HTTPStatus
 from unittest.mock import Mock
 
 from twisted.internet.defer import Deferred
@@ -41,7 +41,7 @@ def create_test_resource(self):
     def test_version_string(self):
         channel = self.make_request("GET", self.url, shorthand=False)
 
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
         self.assertEqual(
             {"server_version", "python_version"}, set(channel.json_body.keys())
         )
@@ -70,25 +70,25 @@ def test_delete_group(self):
             content={"localpart": "test"},
         )
 
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
         group_id = channel.json_body["group_id"]
 
-        self._check_group(group_id, expect_code=200)
+        self._check_group(group_id, expect_code=HTTPStatus.OK)
 
         # Invite/join another user
 
         url = "/groups/%s/admin/users/invite/%s" % (group_id, self.other_user)
         channel = self.make_request(
             "PUT", url.encode("ascii"), access_token=self.admin_user_tok, content={}
         )
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
         url = "/groups/%s/self/accept_invite" % (group_id,)
         channel = self.make_request(
             "PUT", url.encode("ascii"), access_token=self.other_user_token, content={}
         )
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
         # Check other user knows they're in the group
         self.assertIn(group_id, self._get_groups_user_is_in(self.admin_user_tok))
@@ -103,10 +103,10 @@ def test_delete_group(self):
             content={"localpart": "test"},
         )
 
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
-        # Check group returns 404
-        self._check_group(group_id, expect_code=404)
+        # Check group returns HTTPStatus.NOT_FOUND
+        self._check_group(group_id, expect_code=HTTPStatus.NOT_FOUND)
 
         # Check users don't think they're in the group
         self.assertNotIn(group_id, self._get_groups_user_is_in(self.admin_user_tok))
@@ -122,15 +122,13 @@ def _check_group(self, group_id, expect_code):
             "GET", url.encode("ascii"), access_token=self.admin_user_tok
         )
 
-        self.assertEqual(
-            expect_code, int(channel.result["code"]), msg=channel.result["body"]
-        )
+        self.assertEqual(expect_code, channel.code, msg=channel.json_body)
 
     def _get_groups_user_is_in(self, access_token):
         """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)
 
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
         return channel.json_body["groups"]
 
@@ -210,10 +208,10 @@ def _ensure_quarantined(self, admin_user_tok, server_and_media_id):
 
         # Should be quarantined
         self.assertEqual(
-            404,
-            int(channel.code),
+            HTTPStatus.NOT_FOUND,
+            channel.code,
             msg=(
-                "Expected to receive a 404 on accessing quarantined media: %s"
+                "Expected to receive a HTTPStatus.NOT_FOUND on accessing quarantined media: %s"
                 % server_and_media_id
             ),
         )
@@ -232,8 +230,8 @@ def test_quarantine_media_requires_admin(self):
 
         # Expect a forbidden error
         self.assertEqual(
-            403,
-            int(channel.result["code"]),
+            HTTPStatus.FORBIDDEN,
+            channel.code,
             msg="Expected forbidden on quarantining media as a non-admin",
         )
 
@@ -247,8 +245,8 @@ def test_quarantine_media_requires_admin(self):
 
         # Expect a forbidden error
         self.assertEqual(
-            403,
-            int(channel.result["code"]),
+            HTTPStatus.FORBIDDEN,
+            channel.code,
             msg="Expected forbidden on quarantining media as a non-admin",
         )
 
@@ -279,7 +277,7 @@ def test_quarantine_media_by_id(self):
         )
 
         # Should be successful
-        self.assertEqual(200, int(channel.code), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code)
 
         # Quarantine the media
         url = "/_synapse/admin/v1/media/quarantine/%s/%s" % (
@@ -292,7 +290,7 @@ def test_quarantine_media_by_id(self):
             access_token=admin_user_tok,
         )
         self.pump(1.0)
-        self.assertEqual(200, int(channel.code), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
         # Attempt to access the media
         self._ensure_quarantined(admin_user_tok, server_name_and_media_id)
@@ -348,11 +346,9 @@ def test_quarantine_all_media_in_room(self, override_url_template=None):
             access_token=admin_user_tok,
         )
         self.pump(1.0)
-        self.assertEqual(200, int(channel.code), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
         self.assertEqual(
-            json.loads(channel.result["body"].decode("utf-8")),
-            {"num_quarantined": 2},
-            "Expected 2 quarantined items",
+            channel.json_body, {"num_quarantined": 2}, "Expected 2 quarantined items"
         )
 
         # Convert mxc URLs to server/media_id strings
@@ -396,11 +392,9 @@ def test_quarantine_all_media_by_user(self):
             access_token=admin_user_tok,
         )
         self.pump(1.0)
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
         self.assertEqual(
-            json.loads(channel.result["body"].decode("utf-8")),
-            {"num_quarantined": 2},
-            "Expected 2 quarantined items",
+            channel.json_body, {"num_quarantined": 2}, "Expected 2 quarantined items"
         )
 
         # Attempt to access each piece of media
@@ -432,7 +426,7 @@ def test_cannot_quarantine_safe_media(self):
         url = "/_synapse/admin/v1/media/protect/%s" % (urllib.parse.quote(media_id_2),)
         channel = self.make_request("POST", url, access_token=admin_user_tok)
         self.pump(1.0)
-        self.assertEqual(200, int(channel.code), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
 
         # Quarantine all media by this user
         url = "/_synapse/admin/v1/user/%s/media/quarantine" % urllib.parse.quote(
@@ -444,11 +438,9 @@ def test_cannot_quarantine_safe_media(self):
             access_token=admin_user_tok,
         )
         self.pump(1.0)
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
         self.assertEqual(
-            json.loads(channel.result["body"].decode("utf-8")),
-            {"num_quarantined": 1},
-            "Expected 1 quarantined item",
+            channel.json_body, {"num_quarantined": 1}, "Expected 1 quarantined item"
         )
 
         # Attempt to access each piece of media, the first should fail, the
@@ -467,10 +459,10 @@ def test_cannot_quarantine_safe_media(self):
 
         # Shouldn't be quarantined
         self.assertEqual(
-            200,
-            int(channel.code),
+            HTTPStatus.OK,
+            channel.code,
             msg=(
-                "Expected to receive a 200 on accessing not-quarantined media: %s"
+                "Expected to receive a HTTPStatus.OK on accessing not-quarantined media: %s"
                 % server_and_media_id_2
             ),
         )
@@ -499,7 +491,7 @@ def prepare(self, reactor, clock, hs):
     def test_purge_history(self):
         """
         Simple test of purge history API.
-        Test only that is is possible to call, get status 200 and purge_id.
+        Test only that is is possible to call, get status HTTPStatus.OK and purge_id.
         """
 
         channel = self.make_request(
@@ -509,7 +501,7 @@ def test_purge_history(self):
             access_token=self.admin_user_tok,
         )
 
-        self.assertEqual(200, channel.code, msg=channel.json_body)
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
         self.assertIn("purge_id", channel.json_body)
         purge_id = channel.json_body["purge_id"]
 
@@ -520,5 +512,5 @@ def test_purge_history(self):
             access_token=self.admin_user_tok,
         )
 
-        self.assertEqual(200, channel.code, msg=channel.json_body)
+        self.assertEqual(HTTPStatus.OK, channel.code, msg=channel.json_body)
         self.assertEqual("complete", channel.json_body["status"])

tests/rest/admin/test_background_updates.py

@@ -46,7 +46,7 @@ def prepare(self, reactor, clock, hs: HomeServer):
     )
     def test_requester_is_no_admin(self, method: str, url: str):
         """
-        If the user is not a server admin, an error 403 is returned.
+        If the user is not a server admin, an error HTTPStatus.FORBIDDEN is returned.
         """
 
         self.register_user("user", "pass", admin=False)