0
0
Fork 1
mirror of https://mau.dev/maunium/synapse.git synced 2024-06-14 16:48:18 +02:00

Begin to add unit tests for appservice glue and regex testing.

This commit is contained in:
Kegan Dougal 2015-02-04 12:24:20 +00:00
parent 17753f0c20
commit 525a218b2b
5 changed files with 158 additions and 4 deletions

View file

@ -14,8 +14,11 @@
# limitations under the License.
from synapse.api.constants import EventTypes
import logging
import re
logger = logging.getLogger(__name__)
class ApplicationService(object):
"""Defines an application service. This definition is mostly what is
@ -56,15 +59,22 @@ class ApplicationService(object):
return namespaces
def _matches_regex(self, test_string, namespace_key):
if not isinstance(test_string, basestring):
logger.warning(
"Expected a string to test regex against, but got %s",
test_string
)
return False
for regex in self.namespaces[namespace_key]:
if re.match(regex, test_string):
return True
return False
def _matches_user(self, event):
if (hasattr(event, "user_id") and
if (hasattr(event, "sender") and
self._matches_regex(
event.user_id, ApplicationService.NS_USERS)):
event.sender, ApplicationService.NS_USERS)):
return True
# also check m.room.member state key
if (hasattr(event, "type") and event.type == EventTypes.Member

View file

@ -26,10 +26,14 @@ import logging
logger = logging.getLogger(__name__)
class ApplicationServicesHandler(BaseHandler):
# NB: Purposefully not inheriting BaseHandler since that contains way too much
# setup code which this handler does not need or use. This makes testing a lot
# easier.
class ApplicationServicesHandler(object):
def __init__(self, hs):
super(ApplicationServicesHandler, self).__init__(hs)
self.store = hs.get_datastore()
self.hs = hs
self.appservice_api = ApplicationServiceApi(hs)
@defer.inlineCallbacks

View file

@ -0,0 +1,14 @@
# -*- coding: utf-8 -*-
# Copyright 2015 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

View file

@ -0,0 +1,58 @@
# -*- coding: utf-8 -*-
# Copyright 2015 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from synapse.appservice import ApplicationService
from mock import Mock, PropertyMock
from tests import unittest
class ApplicationServiceTestCase(unittest.TestCase):
def setUp(self):
self.service = ApplicationService(
url="some_url",
token="some_token",
namespaces={
ApplicationService.NS_USERS: [],
ApplicationService.NS_ROOMS: [],
ApplicationService.NS_ALIASES: []
}
)
self.event = Mock(
type="m.something", room_id="!foo:bar", sender="@someone:somewhere"
)
def test_regex_user_id_prefix_match(self):
self.service.namespaces[ApplicationService.NS_USERS].append(
"@irc_.*"
)
self.event.sender = "@irc_foobar:matrix.org"
self.assertTrue(self.service.is_interested(self.event))
def test_regex_user_id_prefix_no_match(self):
self.service.namespaces[ApplicationService.NS_USERS].append(
"@irc_.*"
)
self.event.sender = "@someone_else:matrix.org"
self.assertFalse(self.service.is_interested(self.event))
def test_regex_room_member_is_checked(self):
self.service.namespaces[ApplicationService.NS_USERS].append(
"@irc_.*"
)
self.event.sender = "@someone_else:matrix.org"
self.event.type = "m.room.member"
self.event.state_key = "@irc_foobar:matrix.org"
self.assertTrue(self.service.is_interested(self.event))

View file

@ -0,0 +1,68 @@
# -*- coding: utf-8 -*-
# Copyright 2015 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from twisted.internet import defer
from .. import unittest
from synapse.handlers.appservice import ApplicationServicesHandler
from collections import namedtuple
from mock import Mock
# TODO: Should this be a more general thing? tests/api/test_filtering.py uses it
MockEvent = namedtuple("MockEvent", "sender type room_id")
class AppServiceHandlerTestCase(unittest.TestCase):
""" Tests the ApplicationServicesHandler. """
def setUp(self):
self.mock_store = Mock()
self.mock_as_api = Mock()
hs = Mock()
hs.get_datastore = Mock(return_value=self.mock_store)
self.handler = ApplicationServicesHandler(hs) # thing being tested
# FIXME Would be nice to DI this rather than monkey patch:(
if not hasattr(self.handler, "appservice_api"):
# someone probably updated the handler but not the tests. Fail fast.
raise Exception("Test expected handler.appservice_api to exist.")
self.handler.appservice_api = self.mock_as_api
def test_notify_interested_services(self):
interested_service = self._mkservice(is_interested=True)
services = [
self._mkservice(is_interested=False),
interested_service,
self._mkservice(is_interested=False)
]
self.mock_store.get_app_services = Mock(return_value=services)
event = MockEvent(
sender="@someone:anywhere",
type="m.room.message",
room_id="!foo:bar"
)
self.mock_as_api.push = Mock()
self.handler.notify_interested_services(event)
self.mock_as_api.push.assert_called_once_with(interested_service, event)
def _mkservice(self, is_interested):
service = Mock()
service.is_interested = Mock(return_value=is_interested)
service.token = "mock_service_token"
service.url = "mock_service_url"
return service