test_notifications.py 3.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  1. # Copyright 2022 The Matrix.org Foundation C.I.C.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from unittest.mock import AsyncMock, Mock
  15. from twisted.test.proto_helpers import MemoryReactor
  16. import synapse.rest.admin
  17. from synapse.rest.client import login, notifications, receipts, room
  18. from synapse.server import HomeServer
  19. from synapse.util import Clock
  20. from tests.unittest import HomeserverTestCase
  21. class HTTPPusherTests(HomeserverTestCase):
  22. servlets = [
  23. synapse.rest.admin.register_servlets_for_client_rest_resource,
  24. room.register_servlets,
  25. login.register_servlets,
  26. receipts.register_servlets,
  27. notifications.register_servlets,
  28. ]
  29. def prepare(
  30. self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
  31. ) -> None:
  32. self.store = homeserver.get_datastores().main
  33. self.module_api = homeserver.get_module_api()
  34. self.event_creation_handler = homeserver.get_event_creation_handler()
  35. self.sync_handler = homeserver.get_sync_handler()
  36. self.auth_handler = homeserver.get_auth_handler()
  37. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  38. # Mock out the calls over federation.
  39. fed_transport_client = Mock(spec=["send_transaction"])
  40. fed_transport_client.send_transaction = AsyncMock(return_value={})
  41. return self.setup_test_homeserver(
  42. federation_transport_client=fed_transport_client,
  43. )
  44. def test_notify_for_local_invites(self) -> None:
  45. """
  46. Local users will get notified for invites
  47. """
  48. user_id = self.register_user("user", "pass")
  49. access_token = self.login("user", "pass")
  50. other_user_id = self.register_user("otheruser", "pass")
  51. other_access_token = self.login("otheruser", "pass")
  52. # Create a room
  53. room = self.helper.create_room_as(user_id, tok=access_token)
  54. # Check we start with no pushes
  55. channel = self.make_request(
  56. "GET",
  57. "/notifications",
  58. access_token=other_access_token,
  59. )
  60. self.assertEqual(channel.code, 200, channel.result)
  61. self.assertEqual(len(channel.json_body["notifications"]), 0, channel.json_body)
  62. # Send an invite
  63. self.helper.invite(room=room, src=user_id, targ=other_user_id, tok=access_token)
  64. # We should have a notification now
  65. channel = self.make_request(
  66. "GET",
  67. "/notifications",
  68. access_token=other_access_token,
  69. )
  70. self.assertEqual(channel.code, 200)
  71. self.assertEqual(len(channel.json_body["notifications"]), 1, channel.json_body)
  72. self.assertEqual(
  73. channel.json_body["notifications"][0]["event"]["content"]["membership"],
  74. "invite",
  75. channel.json_body,
  76. )