synapse/tests/unittest.py

378 lines
12 KiB
Python
Raw Normal View History

# -*- coding: utf-8 -*-
2016-01-07 05:26:29 +01:00
# Copyright 2014-2016 OpenMarket Ltd
# Copyright 2018 New Vector
#
# 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.
2018-10-01 16:11:58 +02:00
import hashlib
import hmac
import logging
from mock import Mock
from canonicaljson import json
import twisted
import twisted.logger
2018-09-03 18:21:48 +02:00
from twisted.internet.defer import Deferred
from twisted.trial import unittest
from synapse.http.server import JsonResource
from synapse.http.site import SynapseRequest
from synapse.server import HomeServer
from synapse.types import UserID, create_requester
from synapse.util.logcontext import LoggingContextFilter
from tests.server import get_clock, make_request, render, setup_test_homeserver
2018-10-01 16:11:58 +02:00
from tests.utils import default_config
# Set up putting Synapse's logs into Trial's.
rootLogger = logging.getLogger()
log_format = (
"%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(request)s - %(message)s"
)
class ToTwistedHandler(logging.Handler):
tx_log = twisted.logger.Logger()
def emit(self, record):
log_entry = self.format(record)
log_level = record.levelname.lower().replace('warning', 'warn')
2018-06-27 12:38:03 +02:00
self.tx_log.emit(
twisted.logger.LogLevel.levelWithName(log_level),
log_entry.replace("{", r"(").replace("}", r")"),
)
handler = ToTwistedHandler()
formatter = logging.Formatter(log_format)
handler.setFormatter(formatter)
handler.addFilter(LoggingContextFilter(request=""))
rootLogger.addHandler(handler)
def around(target):
"""A CLOS-style 'around' modifier, which wraps the original method of the
given instance with another piece of code.
@around(self)
def method_name(orig, *args, **kwargs):
return orig(*args, **kwargs)
"""
2018-08-10 15:54:09 +02:00
def _around(code):
name = code.__name__
orig = getattr(target, name)
2016-02-19 16:34:38 +01:00
def new(*args, **kwargs):
return code(orig, *args, **kwargs)
2016-02-19 16:34:38 +01:00
setattr(target, name, new)
2016-02-19 16:34:38 +01:00
return _around
class TestCase(unittest.TestCase):
2014-09-12 19:45:48 +02:00
"""A subclass of twisted.trial's TestCase which looks for 'loglevel'
attributes on both itself and its individual test methods, to override the
root logger's logging level while that test (case|method) runs."""
def __init__(self, methodName, *args, **kwargs):
super(TestCase, self).__init__(methodName, *args, **kwargs)
method = getattr(self, methodName)
level = getattr(method, "loglevel", getattr(self, "loglevel", logging.ERROR))
@around(self)
def setUp(orig):
# enable debugging of delayed calls - this means that we get a
# traceback when a unit test exits leaving things on the reactor.
twisted.internet.base.DelayedCall.debug = True
old_level = logging.getLogger().level
if old_level != level:
2018-08-10 15:54:09 +02:00
@around(self)
def tearDown(orig):
ret = orig()
logging.getLogger().setLevel(old_level)
return ret
logging.getLogger().setLevel(level)
return orig()
def assertObjectHasAttributes(self, attrs, obj):
"""Asserts that the given object has each of the attributes given, and
that the value of each matches according to assertEquals."""
for (key, value) in attrs.items():
if not hasattr(obj, key):
raise AssertionError("Expected obj to have a '.%s'" % key)
try:
self.assertEquals(attrs[key], getattr(obj, key))
except AssertionError as e:
2018-10-01 16:11:58 +02:00
raise (type(e))(e.message + " for '.%s'" % key)
def assert_dict(self, required, actual):
"""Does a partial assert of a dict.
Args:
required (dict): The keys and value which MUST be in 'actual'.
actual (dict): The test result. Extra keys will not be checked.
"""
for key in required:
2018-08-10 15:54:09 +02:00
self.assertEquals(
required[key], actual[key], msg="%s mismatch. %s" % (key, actual)
)
def DEBUG(target):
2014-09-12 19:45:48 +02:00
"""A decorator to set the .loglevel attribute to logging.DEBUG.
Can apply to either a TestCase or an individual test method."""
target.loglevel = logging.DEBUG
return target
class HomeserverTestCase(TestCase):
"""
A base TestCase that reduces boilerplate for HomeServer-using test cases.
Attributes:
servlets (list[function]): List of servlet registration function.
user_id (str): The user ID to assume if auth is hijacked.
hijack_auth (bool): Whether to hijack auth to return the user specified
in user_id.
"""
2018-08-30 16:19:58 +02:00
servlets = []
hijack_auth = True
def setUp(self):
"""
Set up the TestCase by calling the homeserver constructor, optionally
hijacking the authentication system to return a fixed user, and then
calling the prepare function.
"""
self.reactor, self.clock = get_clock()
self._hs_args = {"clock": self.clock, "reactor": self.reactor}
self.hs = self.make_homeserver(self.reactor, self.clock)
if self.hs is None:
raise Exception("No homeserver returned from make_homeserver.")
if not isinstance(self.hs, HomeServer):
raise Exception("A homeserver wasn't returned, but %r" % (self.hs,))
# Register the resources
self.resource = JsonResource(self.hs)
for servlet in self.servlets:
servlet(self.hs, self.resource)
if hasattr(self, "user_id"):
from tests.rest.client.v1.utils import RestHelper
self.helper = RestHelper(self.hs, self.resource, self.user_id)
if self.hijack_auth:
def get_user_by_access_token(token=None, allow_guest=False):
return {
"user": UserID.from_string(self.helper.auth_user_id),
"token_id": 1,
"is_guest": False,
}
def get_user_by_req(request, allow_guest=False, rights="access"):
return create_requester(
UserID.from_string(self.helper.auth_user_id), 1, False, None
)
self.hs.get_auth().get_user_by_req = get_user_by_req
self.hs.get_auth().get_user_by_access_token = get_user_by_access_token
self.hs.get_auth().get_access_token_from_request = Mock(
return_value="1234"
)
if hasattr(self, "prepare"):
self.prepare(self.reactor, self.clock, self.hs)
def make_homeserver(self, reactor, clock):
"""
Make and return a homeserver.
Args:
reactor: A Twisted Reactor, or something that pretends to be one.
clock (synapse.util.Clock): The Clock, associated with the reactor.
Returns:
A homeserver (synapse.server.HomeServer) suitable for testing.
Function to be overridden in subclasses.
"""
hs = self.setup_test_homeserver()
return hs
2018-10-01 16:11:58 +02:00
def default_config(self, name="test"):
"""
Get a default HomeServer config object.
Args:
name (str): The homeserver name/domain.
"""
return default_config(name)
def prepare(self, reactor, clock, homeserver):
"""
Prepare for the test. This involves things like mocking out parts of
the homeserver, or building test data common across the whole test
suite.
Args:
reactor: A Twisted Reactor, or something that pretends to be one.
clock (synapse.util.Clock): The Clock, associated with the reactor.
homeserver (synapse.server.HomeServer): The HomeServer to test
against.
Function to optionally be overridden in subclasses.
"""
def make_request(
self, method, path, content=b"", access_token=None, request=SynapseRequest
):
"""
Create a SynapseRequest at the path using the method and containing the
given content.
Args:
method (bytes/unicode): The HTTP request method ("verb").
path (bytes/unicode): The HTTP path, suitably URL encoded (e.g.
escaped UTF-8 & spaces and such).
content (bytes or dict): The body of the request. JSON-encoded, if
a dict.
Returns:
A synapse.http.site.SynapseRequest.
"""
if isinstance(content, dict):
content = json.dumps(content).encode('utf8')
return make_request(method, path, content, access_token, request)
def render(self, request):
"""
Render a request against the resources registered by the test class's
servlets.
Args:
request (synapse.http.site.SynapseRequest): The request to render.
"""
render(request, self.resource, self.reactor)
def setup_test_homeserver(self, *args, **kwargs):
"""
Set up the test homeserver, meant to be called by the overridable
make_homeserver. It automatically passes through the test class's
clock & reactor.
Args:
See tests.utils.setup_test_homeserver.
Returns:
synapse.server.HomeServer
"""
kwargs = dict(kwargs)
kwargs.update(self._hs_args)
return setup_test_homeserver(self.addCleanup, *args, **kwargs)
2018-08-30 16:19:58 +02:00
2018-09-03 18:21:48 +02:00
def pump(self, by=0.0):
2018-08-30 16:19:58 +02:00
"""
Pump the reactor enough that Deferreds will fire.
"""
2018-09-03 18:21:48 +02:00
self.reactor.pump([by] * 100)
2018-08-30 16:19:58 +02:00
def get_success(self, d):
2018-09-03 18:21:48 +02:00
if not isinstance(d, Deferred):
return d
2018-08-30 16:19:58 +02:00
self.pump()
return self.successResultOf(d)
2018-10-01 16:11:58 +02:00
def register_user(self, username, password, admin=False):
"""
Register a user. Requires the Admin API be registered.
Args:
username (bytes/unicode): The user part of the new user.
password (bytes/unicode): The password of the new user.
admin (bool): Whether the user should be created as an admin
or not.
Returns:
The MXID of the new user (unicode).
"""
self.hs.config.registration_shared_secret = u"shared"
# Create the user
request, channel = self.make_request("GET", "/_matrix/client/r0/admin/register")
self.render(request)
nonce = channel.json_body["nonce"]
want_mac = hmac.new(key=b"shared", digestmod=hashlib.sha1)
nonce_str = b"\x00".join([username.encode('utf8'), password.encode('utf8')])
if admin:
nonce_str += b"\x00admin"
else:
nonce_str += b"\x00notadmin"
want_mac.update(nonce.encode('ascii') + b"\x00" + nonce_str)
want_mac = want_mac.hexdigest()
body = json.dumps(
{
"nonce": nonce,
"username": username,
"password": password,
"admin": admin,
"mac": want_mac,
}
)
request, channel = self.make_request(
"POST", "/_matrix/client/r0/admin/register", body.encode('utf8')
)
self.render(request)
self.assertEqual(channel.code, 200)
user_id = channel.json_body["user_id"]
return user_id
def login(self, username, password, device_id=None):
"""
Log in a user, and get an access token. Requires the Login API be
registered.
"""
body = {"type": "m.login.password", "user": username, "password": password}
if device_id:
body["device_id"] = device_id
request, channel = self.make_request(
"POST", "/_matrix/client/r0/login", json.dumps(body).encode('utf8')
)
self.render(request)
self.assertEqual(channel.code, 200)
access_token = channel.json_body["access_token"].encode('ascii')
return access_token