test_appservice.py 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2015 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. from synapse.appservice import ApplicationService
  16. from mock import Mock, PropertyMock
  17. from tests import unittest
  18. def _regex(regex, exclusive=True):
  19. return {
  20. "regex": regex,
  21. "exclusive": exclusive
  22. }
  23. class ApplicationServiceTestCase(unittest.TestCase):
  24. def setUp(self):
  25. self.service = ApplicationService(
  26. url="some_url",
  27. token="some_token",
  28. namespaces={
  29. ApplicationService.NS_USERS: [],
  30. ApplicationService.NS_ROOMS: [],
  31. ApplicationService.NS_ALIASES: []
  32. }
  33. )
  34. self.event = Mock(
  35. type="m.something", room_id="!foo:bar", sender="@someone:somewhere"
  36. )
  37. def test_regex_user_id_prefix_match(self):
  38. self.service.namespaces[ApplicationService.NS_USERS].append(
  39. _regex("@irc_.*")
  40. )
  41. self.event.sender = "@irc_foobar:matrix.org"
  42. self.assertTrue(self.service.is_interested(self.event))
  43. def test_regex_user_id_prefix_no_match(self):
  44. self.service.namespaces[ApplicationService.NS_USERS].append(
  45. _regex("@irc_.*")
  46. )
  47. self.event.sender = "@someone_else:matrix.org"
  48. self.assertFalse(self.service.is_interested(self.event))
  49. def test_regex_room_member_is_checked(self):
  50. self.service.namespaces[ApplicationService.NS_USERS].append(
  51. _regex("@irc_.*")
  52. )
  53. self.event.sender = "@someone_else:matrix.org"
  54. self.event.type = "m.room.member"
  55. self.event.state_key = "@irc_foobar:matrix.org"
  56. self.assertTrue(self.service.is_interested(self.event))
  57. def test_regex_room_id_match(self):
  58. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  59. _regex("!some_prefix.*some_suffix:matrix.org")
  60. )
  61. self.event.room_id = "!some_prefixs0m3th1nGsome_suffix:matrix.org"
  62. self.assertTrue(self.service.is_interested(self.event))
  63. def test_regex_room_id_no_match(self):
  64. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  65. _regex("!some_prefix.*some_suffix:matrix.org")
  66. )
  67. self.event.room_id = "!XqBunHwQIXUiqCaoxq:matrix.org"
  68. self.assertFalse(self.service.is_interested(self.event))
  69. def test_regex_alias_match(self):
  70. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  71. _regex("#irc_.*:matrix.org")
  72. )
  73. self.assertTrue(self.service.is_interested(
  74. self.event,
  75. aliases_for_event=["#irc_foobar:matrix.org", "#athing:matrix.org"]
  76. ))
  77. def test_non_exclusive_alias(self):
  78. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  79. _regex("#irc_.*:matrix.org", exclusive=False)
  80. )
  81. self.assertFalse(self.service.is_exclusive_alias(
  82. "#irc_foobar:matrix.org"
  83. ))
  84. def test_non_exclusive_room(self):
  85. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  86. _regex("!irc_.*:matrix.org", exclusive=False)
  87. )
  88. self.assertFalse(self.service.is_exclusive_room(
  89. "!irc_foobar:matrix.org"
  90. ))
  91. def test_non_exclusive_user(self):
  92. self.service.namespaces[ApplicationService.NS_USERS].append(
  93. _regex("@irc_.*:matrix.org", exclusive=False)
  94. )
  95. self.assertFalse(self.service.is_exclusive_user(
  96. "@irc_foobar:matrix.org"
  97. ))
  98. def test_exclusive_alias(self):
  99. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  100. _regex("#irc_.*:matrix.org", exclusive=True)
  101. )
  102. self.assertTrue(self.service.is_exclusive_alias(
  103. "#irc_foobar:matrix.org"
  104. ))
  105. def test_exclusive_user(self):
  106. self.service.namespaces[ApplicationService.NS_USERS].append(
  107. _regex("@irc_.*:matrix.org", exclusive=True)
  108. )
  109. self.assertTrue(self.service.is_exclusive_user(
  110. "@irc_foobar:matrix.org"
  111. ))
  112. def test_exclusive_room(self):
  113. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  114. _regex("!irc_.*:matrix.org", exclusive=True)
  115. )
  116. self.assertTrue(self.service.is_exclusive_room(
  117. "!irc_foobar:matrix.org"
  118. ))
  119. def test_regex_alias_no_match(self):
  120. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  121. _regex("#irc_.*:matrix.org")
  122. )
  123. self.assertFalse(self.service.is_interested(
  124. self.event,
  125. aliases_for_event=["#xmpp_foobar:matrix.org", "#athing:matrix.org"]
  126. ))
  127. def test_regex_multiple_matches(self):
  128. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  129. _regex("#irc_.*:matrix.org")
  130. )
  131. self.service.namespaces[ApplicationService.NS_USERS].append(
  132. _regex("@irc_.*")
  133. )
  134. self.event.sender = "@irc_foobar:matrix.org"
  135. self.assertTrue(self.service.is_interested(
  136. self.event,
  137. aliases_for_event=["#irc_barfoo:matrix.org"]
  138. ))
  139. def test_restrict_to_rooms(self):
  140. self.service.namespaces[ApplicationService.NS_ROOMS].append(
  141. _regex("!flibble_.*:matrix.org")
  142. )
  143. self.service.namespaces[ApplicationService.NS_USERS].append(
  144. _regex("@irc_.*")
  145. )
  146. self.event.sender = "@irc_foobar:matrix.org"
  147. self.event.room_id = "!wibblewoo:matrix.org"
  148. self.assertFalse(self.service.is_interested(
  149. self.event,
  150. restrict_to=ApplicationService.NS_ROOMS
  151. ))
  152. def test_restrict_to_aliases(self):
  153. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  154. _regex("#xmpp_.*:matrix.org")
  155. )
  156. self.service.namespaces[ApplicationService.NS_USERS].append(
  157. _regex("@irc_.*")
  158. )
  159. self.event.sender = "@irc_foobar:matrix.org"
  160. self.assertFalse(self.service.is_interested(
  161. self.event,
  162. restrict_to=ApplicationService.NS_ALIASES,
  163. aliases_for_event=["#irc_barfoo:matrix.org"]
  164. ))
  165. def test_restrict_to_senders(self):
  166. self.service.namespaces[ApplicationService.NS_ALIASES].append(
  167. _regex("#xmpp_.*:matrix.org")
  168. )
  169. self.service.namespaces[ApplicationService.NS_USERS].append(
  170. _regex("@irc_.*")
  171. )
  172. self.event.sender = "@xmpp_foobar:matrix.org"
  173. self.assertFalse(self.service.is_interested(
  174. self.event,
  175. restrict_to=ApplicationService.NS_USERS,
  176. aliases_for_event=["#xmpp_barfoo:matrix.org"]
  177. ))
  178. def test_member_list_match(self):
  179. self.service.namespaces[ApplicationService.NS_USERS].append(
  180. _regex("@irc_.*")
  181. )
  182. join_list = [
  183. Mock(
  184. type="m.room.member", room_id="!foo:bar", sender="@alice:here",
  185. state_key="@alice:here"
  186. ),
  187. Mock(
  188. type="m.room.member", room_id="!foo:bar", sender="@irc_fo:here",
  189. state_key="@irc_fo:here" # AS user
  190. ),
  191. Mock(
  192. type="m.room.member", room_id="!foo:bar", sender="@bob:here",
  193. state_key="@bob:here"
  194. )
  195. ]
  196. self.event.sender = "@xmpp_foobar:matrix.org"
  197. self.assertTrue(self.service.is_interested(
  198. event=self.event,
  199. member_list=join_list
  200. ))