1
0

test_appservice.py 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2015, 2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import re
  16. from mock import Mock
  17. from twisted.internet import defer
  18. from synapse.appservice import ApplicationService
  19. from tests import unittest
  20. def _regex(regex, exclusive=True):
  21. return {"regex": re.compile(regex), "exclusive": exclusive}
  22. class ApplicationServiceTestCase(unittest.TestCase):
  23. def setUp(self):
  24. self.service = ApplicationService(
  25. id="unique_identifier",
  26. sender="@as:test",
  27. url="some_url",
  28. token="some_token",
  29. hostname="matrix.org", # only used by get_groups_for_user
  30. namespaces={
  31. ApplicationService.NS_USERS: [],
  32. ApplicationService.NS_ROOMS: [],
  33. ApplicationService.NS_ALIASES: [],
  34. },
  35. )
  36. self.event = Mock(
  37. type="m.something", room_id="!foo:bar", sender="@someone:somewhere"
  38. )
  39. self.store = Mock()
  40. @defer.inlineCallbacks
  41. def test_regex_user_id_prefix_match(self):
  42. self.service.namespaces[ApplicationService.NS_USERS].append(_regex("@irc_.*"))
  43. self.event.sender = "@irc_foobar:matrix.org"
  44. self.assertTrue(
  45. (yield defer.ensureDeferred(self.service.is_interested(self.event)))
  46. )
  47. @defer.inlineCallbacks
  48. def test_regex_user_id_prefix_no_match(self):
  49. self.service.namespaces[ApplicationService.NS_USERS].append(_regex("@irc_.*"))
  50. self.event.sender = "@someone_else:matrix.org"
  51. self.assertFalse(
  52. (yield defer.ensureDeferred(self.service.is_interested(self.event)))
  53. )
  54. @defer.inlineCallbacks
  55. def test_regex_room_member_is_checked(self):
  56. self.service.namespaces[ApplicationService.NS_USERS].append(_regex("@irc_.*"))
  57. self.event.sender = "@someone_else:matrix.org"
  58. self.event.type = "m.room.member"
  59. self.event.state_key = "@irc_foobar:matrix.org"
  60. self.assertTrue(
  61. (yield defer.ensureDeferred(self.service.is_interested(self.event)))
  62. )
  63. @defer.inlineCallbacks
  64. def test_regex_room_id_match(self):
  65. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  66. _regex("!some_prefix.*some_suffix:matrix.org")
  67. )
  68. self.event.room_id = "!some_prefixs0m3th1nGsome_suffix:matrix.org"
  69. self.assertTrue(
  70. (yield defer.ensureDeferred(self.service.is_interested(self.event)))
  71. )
  72. @defer.inlineCallbacks
  73. def test_regex_room_id_no_match(self):
  74. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  75. _regex("!some_prefix.*some_suffix:matrix.org")
  76. )
  77. self.event.room_id = "!XqBunHwQIXUiqCaoxq:matrix.org"
  78. self.assertFalse(
  79. (yield defer.ensureDeferred(self.service.is_interested(self.event)))
  80. )
  81. @defer.inlineCallbacks
  82. def test_regex_alias_match(self):
  83. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  84. _regex("#irc_.*:matrix.org")
  85. )
  86. self.store.get_aliases_for_room.return_value = defer.succeed(
  87. ["#irc_foobar:matrix.org", "#athing:matrix.org"]
  88. )
  89. self.store.get_users_in_room.return_value = defer.succeed([])
  90. self.assertTrue(
  91. (
  92. yield defer.ensureDeferred(
  93. self.service.is_interested(self.event, self.store)
  94. )
  95. )
  96. )
  97. def test_non_exclusive_alias(self):
  98. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  99. _regex("#irc_.*:matrix.org", exclusive=False)
  100. )
  101. self.assertFalse(self.service.is_exclusive_alias("#irc_foobar:matrix.org"))
  102. def test_non_exclusive_room(self):
  103. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  104. _regex("!irc_.*:matrix.org", exclusive=False)
  105. )
  106. self.assertFalse(self.service.is_exclusive_room("!irc_foobar:matrix.org"))
  107. def test_non_exclusive_user(self):
  108. self.service.namespaces[ApplicationService.NS_USERS].append(
  109. _regex("@irc_.*:matrix.org", exclusive=False)
  110. )
  111. self.assertFalse(self.service.is_exclusive_user("@irc_foobar:matrix.org"))
  112. def test_exclusive_alias(self):
  113. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  114. _regex("#irc_.*:matrix.org", exclusive=True)
  115. )
  116. self.assertTrue(self.service.is_exclusive_alias("#irc_foobar:matrix.org"))
  117. def test_exclusive_user(self):
  118. self.service.namespaces[ApplicationService.NS_USERS].append(
  119. _regex("@irc_.*:matrix.org", exclusive=True)
  120. )
  121. self.assertTrue(self.service.is_exclusive_user("@irc_foobar:matrix.org"))
  122. def test_exclusive_room(self):
  123. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  124. _regex("!irc_.*:matrix.org", exclusive=True)
  125. )
  126. self.assertTrue(self.service.is_exclusive_room("!irc_foobar:matrix.org"))
  127. @defer.inlineCallbacks
  128. def test_regex_alias_no_match(self):
  129. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  130. _regex("#irc_.*:matrix.org")
  131. )
  132. self.store.get_aliases_for_room.return_value = defer.succeed(
  133. ["#xmpp_foobar:matrix.org", "#athing:matrix.org"]
  134. )
  135. self.store.get_users_in_room.return_value = defer.succeed([])
  136. self.assertFalse(
  137. (
  138. yield defer.ensureDeferred(
  139. self.service.is_interested(self.event, self.store)
  140. )
  141. )
  142. )
  143. @defer.inlineCallbacks
  144. def test_regex_multiple_matches(self):
  145. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  146. _regex("#irc_.*:matrix.org")
  147. )
  148. self.service.namespaces[ApplicationService.NS_USERS].append(_regex("@irc_.*"))
  149. self.event.sender = "@irc_foobar:matrix.org"
  150. self.store.get_aliases_for_room.return_value = defer.succeed(
  151. ["#irc_barfoo:matrix.org"]
  152. )
  153. self.store.get_users_in_room.return_value = defer.succeed([])
  154. self.assertTrue(
  155. (
  156. yield defer.ensureDeferred(
  157. self.service.is_interested(self.event, self.store)
  158. )
  159. )
  160. )
  161. @defer.inlineCallbacks
  162. def test_interested_in_self(self):
  163. # make sure invites get through
  164. self.service.sender = "@appservice:name"
  165. self.service.namespaces[ApplicationService.NS_USERS].append(_regex("@irc_.*"))
  166. self.event.type = "m.room.member"
  167. self.event.content = {"membership": "invite"}
  168. self.event.state_key = self.service.sender
  169. self.assertTrue(
  170. (yield defer.ensureDeferred(self.service.is_interested(self.event)))
  171. )
  172. @defer.inlineCallbacks
  173. def test_member_list_match(self):
  174. self.service.namespaces[ApplicationService.NS_USERS].append(_regex("@irc_.*"))
  175. # Note that @irc_fo:here is the AS user.
  176. self.store.get_users_in_room.return_value = defer.succeed(
  177. ["@alice:here", "@irc_fo:here", "@bob:here"]
  178. )
  179. self.store.get_aliases_for_room.return_value = defer.succeed([])
  180. self.event.sender = "@xmpp_foobar:matrix.org"
  181. self.assertTrue(
  182. (
  183. yield defer.ensureDeferred(
  184. self.service.is_interested(event=self.event, store=self.store)
  185. )
  186. )
  187. )