diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/config/test_server.py | 61 | ||||
| -rw-r--r-- | tests/push/test_email.py | 51 | ||||
| -rw-r--r-- | tests/rest/client/v1/test_login.py | 16 |
3 files changed, 121 insertions, 7 deletions
diff --git a/tests/config/test_server.py b/tests/config/test_server.py index a10d017120..98af7aa675 100644 --- a/tests/config/test_server.py +++ b/tests/config/test_server.py @@ -15,7 +15,8 @@ import yaml -from synapse.config.server import ServerConfig, is_threepid_reserved +from synapse.config._base import ConfigError +from synapse.config.server import ServerConfig, generate_ip_set, is_threepid_reserved from tests import unittest @@ -128,3 +129,61 @@ class ServerConfigTestCase(unittest.TestCase): ) self.assertEqual(conf["listeners"], expected_listeners) + + +class GenerateIpSetTestCase(unittest.TestCase): + def test_empty(self): + ip_set = generate_ip_set(()) + self.assertFalse(ip_set) + + ip_set = generate_ip_set((), ()) + self.assertFalse(ip_set) + + def test_generate(self): + """Check adding IPv4 and IPv6 addresses.""" + # IPv4 address + ip_set = generate_ip_set(("1.2.3.4",)) + self.assertEqual(len(ip_set.iter_cidrs()), 4) + + # IPv4 CIDR + ip_set = generate_ip_set(("1.2.3.4/24",)) + self.assertEqual(len(ip_set.iter_cidrs()), 4) + + # IPv6 address + ip_set = generate_ip_set(("2001:db8::8a2e:370:7334",)) + self.assertEqual(len(ip_set.iter_cidrs()), 1) + + # IPv6 CIDR + ip_set = generate_ip_set(("2001:db8::/104",)) + self.assertEqual(len(ip_set.iter_cidrs()), 1) + + # The addresses can overlap OK. + ip_set = generate_ip_set(("1.2.3.4", "::1.2.3.4")) + self.assertEqual(len(ip_set.iter_cidrs()), 4) + + def test_extra(self): + """Extra IP addresses are treated the same.""" + ip_set = generate_ip_set((), ("1.2.3.4",)) + self.assertEqual(len(ip_set.iter_cidrs()), 4) + + ip_set = generate_ip_set(("1.1.1.1",), ("1.2.3.4",)) + self.assertEqual(len(ip_set.iter_cidrs()), 8) + + # They can duplicate without error. + ip_set = generate_ip_set(("1.2.3.4",), ("1.2.3.4",)) + self.assertEqual(len(ip_set.iter_cidrs()), 4) + + def test_bad_value(self): + """An error should be raised if a bad value is passed in.""" + with self.assertRaises(ConfigError): + generate_ip_set(("not-an-ip",)) + + with self.assertRaises(ConfigError): + generate_ip_set(("1.2.3.4/128",)) + + with self.assertRaises(ConfigError): + generate_ip_set((":::",)) + + # The following get treated as empty data. + self.assertFalse(generate_ip_set(None)) + self.assertFalse(generate_ip_set({})) diff --git a/tests/push/test_email.py b/tests/push/test_email.py index c4e1e7ed85..22f452ec24 100644 --- a/tests/push/test_email.py +++ b/tests/push/test_email.py @@ -124,13 +124,18 @@ class EmailPusherTests(HomeserverTestCase): ) self.helper.join(room=room, user=self.others[0].id, tok=self.others[0].token) - # The other user sends some messages + # The other user sends a single message. self.helper.send(room, body="Hi!", tok=self.others[0].token) - self.helper.send(room, body="There!", tok=self.others[0].token) # We should get emailed about that message self._check_for_mail() + # The other user sends multiple messages. + self.helper.send(room, body="Hi!", tok=self.others[0].token) + self.helper.send(room, body="There!", tok=self.others[0].token) + + self._check_for_mail() + def test_invite_sends_email(self): # Create a room and invite the user to it room = self.helper.create_room_as(self.others[0].id, tok=self.others[0].token) @@ -217,6 +222,45 @@ class EmailPusherTests(HomeserverTestCase): # We should get emailed about those messages self._check_for_mail() + def test_empty_room(self): + """All users leaving a room shouldn't cause the pusher to break.""" + # Create a simple room with two users + room = self.helper.create_room_as(self.user_id, tok=self.access_token) + self.helper.invite( + room=room, src=self.user_id, tok=self.access_token, targ=self.others[0].id + ) + self.helper.join(room=room, user=self.others[0].id, tok=self.others[0].token) + + # The other user sends a single message. + self.helper.send(room, body="Hi!", tok=self.others[0].token) + + # Leave the room before the message is processed. + self.helper.leave(room, self.user_id, tok=self.access_token) + self.helper.leave(room, self.others[0].id, tok=self.others[0].token) + + # We should get emailed about that message + self._check_for_mail() + + def test_empty_room_multiple_messages(self): + """All users leaving a room shouldn't cause the pusher to break.""" + # Create a simple room with two users + room = self.helper.create_room_as(self.user_id, tok=self.access_token) + self.helper.invite( + room=room, src=self.user_id, tok=self.access_token, targ=self.others[0].id + ) + self.helper.join(room=room, user=self.others[0].id, tok=self.others[0].token) + + # The other user sends a single message. + self.helper.send(room, body="Hi!", tok=self.others[0].token) + self.helper.send(room, body="There!", tok=self.others[0].token) + + # Leave the room before the message is processed. + self.helper.leave(room, self.user_id, tok=self.access_token) + self.helper.leave(room, self.others[0].id, tok=self.others[0].token) + + # We should get emailed about that message + self._check_for_mail() + def test_encrypted_message(self): room = self.helper.create_room_as(self.user_id, tok=self.access_token) self.helper.invite( @@ -269,3 +313,6 @@ class EmailPusherTests(HomeserverTestCase): pushers = list(pushers) self.assertEqual(len(pushers), 1) self.assertTrue(pushers[0].last_stream_ordering > last_stream_ordering) + + # Reset the attempts. + self.email_attempts = [] diff --git a/tests/rest/client/v1/test_login.py b/tests/rest/client/v1/test_login.py index 66dfdaffbc..ceb4ad2366 100644 --- a/tests/rest/client/v1/test_login.py +++ b/tests/rest/client/v1/test_login.py @@ -15,7 +15,7 @@ import time import urllib.parse -from typing import Any, Dict, Union +from typing import Any, Dict, List, Union from urllib.parse import urlencode from mock import Mock @@ -493,13 +493,21 @@ class MultiSSOTestCase(unittest.HomeserverTestCase): self.assertEqual(channel.code, 200, channel.result) # parse the form to check it has fields assumed elsewhere in this class + html = channel.result["body"].decode("utf-8") p = TestHtmlParser() - p.feed(channel.result["body"].decode("utf-8")) + p.feed(html) p.close() - self.assertCountEqual(p.radios["idp"], ["cas", "oidc", "oidc-idp1", "saml"]) + # there should be a link for each href + returned_idps = [] # type: List[str] + for link in p.links: + path, query = link.split("?", 1) + self.assertEqual(path, "pick_idp") + params = urllib.parse.parse_qs(query) + self.assertEqual(params["redirectUrl"], [TEST_CLIENT_REDIRECT_URL]) + returned_idps.append(params["idp"][0]) - self.assertEqual(p.hiddens["redirectUrl"], TEST_CLIENT_REDIRECT_URL) + self.assertCountEqual(returned_idps, ["cas", "oidc", "oidc-idp1", "saml"]) def test_multi_sso_redirect_to_cas(self): """If CAS is chosen, should redirect to the CAS server""" |