diff options
Diffstat (limited to 'tests/rest/admin/test_media.py')
| -rw-r--r-- | tests/rest/admin/test_media.py | 106 |
1 files changed, 102 insertions, 4 deletions
diff --git a/tests/rest/admin/test_media.py b/tests/rest/admin/test_media.py index ce30a19213..db0e78c039 100644 --- a/tests/rest/admin/test_media.py +++ b/tests/rest/admin/test_media.py @@ -27,6 +27,9 @@ from tests import unittest from tests.server import FakeSite, make_request from tests.test_utils import SMALL_PNG +VALID_TIMESTAMP = 1609459200000 # 2021-01-01 in milliseconds +INVALID_TIMESTAMP_IN_S = 1893456000 # 2030-01-01 in seconds + class DeleteMediaByIDTestCase(unittest.HomeserverTestCase): @@ -203,6 +206,9 @@ class DeleteMediaByDateSizeTestCase(unittest.HomeserverTestCase): self.filepaths = MediaFilePaths(hs.config.media.media_store_path) self.url = "/_synapse/admin/v1/media/%s/delete" % self.server_name + # Move clock up to somewhat realistic time + self.reactor.advance(1000000000) + def test_no_auth(self): """ Try to delete media without authentication. @@ -237,7 +243,7 @@ class DeleteMediaByDateSizeTestCase(unittest.HomeserverTestCase): channel = self.make_request( "POST", - url + "?before_ts=1234", + url + f"?before_ts={VALID_TIMESTAMP}", access_token=self.admin_user_tok, ) @@ -273,13 +279,27 @@ class DeleteMediaByDateSizeTestCase(unittest.HomeserverTestCase): self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"]) self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) self.assertEqual( - "Query parameter before_ts must be a string representing a positive integer.", + "Query parameter before_ts must be a positive integer.", channel.json_body["error"], ) channel = self.make_request( "POST", - self.url + "?before_ts=1234&size_gt=-1234", + self.url + f"?before_ts={INVALID_TIMESTAMP_IN_S}", + access_token=self.admin_user_tok, + ) + + self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"]) + self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) + self.assertEqual( + "Query parameter before_ts you provided is from the year 1970. " + + "Double check that you are providing a timestamp in milliseconds.", + channel.json_body["error"], + ) + + channel = self.make_request( + "POST", + self.url + f"?before_ts={VALID_TIMESTAMP}&size_gt=-1234", access_token=self.admin_user_tok, ) @@ -292,7 +312,7 @@ class DeleteMediaByDateSizeTestCase(unittest.HomeserverTestCase): channel = self.make_request( "POST", - self.url + "?before_ts=1234&keep_profiles=not_bool", + self.url + f"?before_ts={VALID_TIMESTAMP}&keep_profiles=not_bool", access_token=self.admin_user_tok, ) @@ -767,3 +787,81 @@ class ProtectMediaByIDTestCase(unittest.HomeserverTestCase): media_info = self.get_success(self.store.get_local_media(self.media_id)) self.assertFalse(media_info["safe_from_quarantine"]) + + +class PurgeMediaCacheTestCase(unittest.HomeserverTestCase): + + servlets = [ + synapse.rest.admin.register_servlets, + synapse.rest.admin.register_servlets_for_media_repo, + login.register_servlets, + profile.register_servlets, + room.register_servlets, + ] + + def prepare(self, reactor, clock, hs): + self.media_repo = hs.get_media_repository_resource() + self.server_name = hs.hostname + + self.admin_user = self.register_user("admin", "pass", admin=True) + self.admin_user_tok = self.login("admin", "pass") + + self.filepaths = MediaFilePaths(hs.config.media.media_store_path) + self.url = "/_synapse/admin/v1/purge_media_cache" + + def test_no_auth(self): + """ + Try to delete media without authentication. + """ + + channel = self.make_request("POST", self.url, b"{}") + + self.assertEqual(401, int(channel.result["code"]), msg=channel.result["body"]) + self.assertEqual(Codes.MISSING_TOKEN, channel.json_body["errcode"]) + + def test_requester_is_not_admin(self): + """ + If the user is not a server admin, an error is returned. + """ + self.other_user = self.register_user("user", "pass") + self.other_user_token = self.login("user", "pass") + + channel = self.make_request( + "POST", + self.url, + access_token=self.other_user_token, + ) + + self.assertEqual(403, int(channel.result["code"]), msg=channel.result["body"]) + self.assertEqual(Codes.FORBIDDEN, channel.json_body["errcode"]) + + def test_invalid_parameter(self): + """ + If parameters are invalid, an error is returned. + """ + channel = self.make_request( + "POST", + self.url + "?before_ts=-1234", + access_token=self.admin_user_tok, + ) + + self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"]) + self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) + self.assertEqual( + "Query parameter before_ts must be a positive integer.", + channel.json_body["error"], + ) + + channel = self.make_request( + "POST", + self.url + f"?before_ts={INVALID_TIMESTAMP_IN_S}", + access_token=self.admin_user_tok, + ) + + self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"]) + self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"]) + self.assertEqual( + "Query parameter before_ts you provided is from the year 1970. " + + "Double check that you are providing a timestamp in milliseconds.", + channel.json_body["error"], + ) |