diff --git a/tests/rest/client/test_login.py b/tests/rest/client/test_login.py
index 5b2243fe52..f5c195a075 100644
--- a/tests/rest/client/test_login.py
+++ b/tests/rest/client/test_login.py
@@ -445,26 +445,9 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
[f["type"] for f in channel.json_body["flows"]], expected_flow_types
)
- @override_config({"experimental_features": {"msc2858_enabled": True}})
- def test_get_msc2858_login_flows(self):
- """The SSO flow should include IdP info if MSC2858 is enabled"""
- channel = self.make_request("GET", "/_matrix/client/r0/login")
- self.assertEqual(channel.code, 200, channel.result)
-
- # stick the flows results in a dict by type
- flow_results: Dict[str, Any] = {}
- for f in channel.json_body["flows"]:
- flow_type = f["type"]
- self.assertNotIn(
- flow_type, flow_results, "duplicate flow type %s" % (flow_type,)
- )
- flow_results[flow_type] = f
-
- self.assertIn("m.login.sso", flow_results, "m.login.sso was not returned")
- sso_flow = flow_results.pop("m.login.sso")
- # we should have a set of IdPs
+ flows = {flow["type"]: flow for flow in channel.json_body["flows"]}
self.assertCountEqual(
- sso_flow["org.matrix.msc2858.identity_providers"],
+ flows["m.login.sso"]["identity_providers"],
[
{"id": "cas", "name": "CAS"},
{"id": "saml", "name": "SAML"},
@@ -473,19 +456,10 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
],
)
- # the rest of the flows are simple
- expected_flows = [
- {"type": "m.login.cas"},
- {"type": "m.login.token"},
- {"type": "m.login.password"},
- ] + ADDITIONAL_LOGIN_FLOWS
-
- self.assertCountEqual(flow_results.values(), expected_flows)
-
def test_multi_sso_redirect(self):
"""/login/sso/redirect should redirect to an identity picker"""
# first hit the redirect url, which should redirect to our idp picker
- channel = self._make_sso_redirect_request(False, None)
+ channel = self._make_sso_redirect_request(None)
self.assertEqual(channel.code, 302, channel.result)
uri = channel.headers.getRawHeaders("Location")[0]
@@ -637,24 +611,13 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
def test_client_idp_redirect_to_unknown(self):
"""If the client tries to pick an unknown IdP, return a 404"""
- channel = self._make_sso_redirect_request(False, "xxx")
+ channel = self._make_sso_redirect_request("xxx")
self.assertEqual(channel.code, 404, channel.result)
self.assertEqual(channel.json_body["errcode"], "M_NOT_FOUND")
def test_client_idp_redirect_to_oidc(self):
"""If the client pick a known IdP, redirect to it"""
- channel = self._make_sso_redirect_request(False, "oidc")
- self.assertEqual(channel.code, 302, channel.result)
- oidc_uri = channel.headers.getRawHeaders("Location")[0]
- oidc_uri_path, oidc_uri_query = oidc_uri.split("?", 1)
-
- # it should redirect us to the auth page of the OIDC server
- self.assertEqual(oidc_uri_path, TEST_OIDC_AUTH_ENDPOINT)
-
- @override_config({"experimental_features": {"msc2858_enabled": True}})
- def test_client_msc2858_redirect_to_oidc(self):
- """Test the unstable API"""
- channel = self._make_sso_redirect_request(True, "oidc")
+ channel = self._make_sso_redirect_request("oidc")
self.assertEqual(channel.code, 302, channel.result)
oidc_uri = channel.headers.getRawHeaders("Location")[0]
oidc_uri_path, oidc_uri_query = oidc_uri.split("?", 1)
@@ -662,26 +625,12 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
# it should redirect us to the auth page of the OIDC server
self.assertEqual(oidc_uri_path, TEST_OIDC_AUTH_ENDPOINT)
- def test_client_idp_redirect_msc2858_disabled(self):
- """If the client tries to use the MSC2858 endpoint but MSC2858 is disabled, return a 400"""
- channel = self._make_sso_redirect_request(True, "oidc")
- self.assertEqual(channel.code, 400, channel.result)
- self.assertEqual(channel.json_body["errcode"], "M_UNRECOGNIZED")
-
- def _make_sso_redirect_request(
- self, unstable_endpoint: bool = False, idp_prov: Optional[str] = None
- ):
+ def _make_sso_redirect_request(self, idp_prov: Optional[str] = None):
"""Send a request to /_matrix/client/r0/login/sso/redirect
- ... or the unstable equivalent
-
... possibly specifying an IDP provider
"""
- endpoint = (
- "/_matrix/client/unstable/org.matrix.msc2858/login/sso/redirect"
- if unstable_endpoint
- else "/_matrix/client/r0/login/sso/redirect"
- )
+ endpoint = "/_matrix/client/r0/login/sso/redirect"
if idp_prov is not None:
endpoint += "/" + idp_prov
endpoint += "?redirectUrl=" + urllib.parse.quote_plus(TEST_CLIENT_REDIRECT_URL)
diff --git a/tests/rest/media/v1/test_url_preview.py b/tests/rest/media/v1/test_url_preview.py
index d3ef7bb4c6..9f6fbfe6de 100644
--- a/tests/rest/media/v1/test_url_preview.py
+++ b/tests/rest/media/v1/test_url_preview.py
@@ -14,13 +14,14 @@
import json
import os
import re
-from unittest.mock import patch
from twisted.internet._resolver import HostResolution
from twisted.internet.address import IPv4Address, IPv6Address
from twisted.internet.error import DNSLookupError
from twisted.test.proto_helpers import AccumulatingProtocol
+from synapse.config.oembed import OEmbedEndpointConfig
+
from tests import unittest
from tests.server import FakeTransport
@@ -81,6 +82,27 @@ class URLPreviewTests(unittest.HomeserverTestCase):
hs = self.setup_test_homeserver(config=config)
+ # After the hs is created, modify the parsed oEmbed config (to avoid
+ # messing with files).
+ #
+ # Note that HTTP URLs are used to avoid having to deal with TLS in tests.
+ hs.config.oembed.oembed_patterns = [
+ OEmbedEndpointConfig(
+ api_endpoint="http://publish.twitter.com/oembed",
+ url_patterns=[
+ re.compile(r"http://twitter\.com/.+/status/.+"),
+ ],
+ formats=None,
+ ),
+ OEmbedEndpointConfig(
+ api_endpoint="http://www.hulu.com/api/oembed.{format}",
+ url_patterns=[
+ re.compile(r"http://www\.hulu\.com/watch/.+"),
+ ],
+ formats=["json"],
+ ),
+ ]
+
return hs
def prepare(self, reactor, clock, hs):
@@ -544,123 +566,146 @@ class URLPreviewTests(unittest.HomeserverTestCase):
def test_oembed_photo(self):
"""Test an oEmbed endpoint which returns a 'photo' type which redirects the preview to a new URL."""
- # Route the HTTP version to an HTTP endpoint so that the tests work.
- with patch.dict(
- "synapse.rest.media.v1.preview_url_resource._oembed_patterns",
- {
- re.compile(
- r"http://twitter\.com/.+/status/.+"
- ): "http://publish.twitter.com/oembed",
- },
- clear=True,
- ):
-
- self.lookups["publish.twitter.com"] = [(IPv4Address, "10.1.2.3")]
- self.lookups["cdn.twitter.com"] = [(IPv4Address, "10.1.2.3")]
-
- result = {
- "version": "1.0",
- "type": "photo",
- "url": "http://cdn.twitter.com/matrixdotorg",
- }
- oembed_content = json.dumps(result).encode("utf-8")
-
- end_content = (
- b"<html><head>"
- b"<title>Some Title</title>"
- b'<meta property="og:description" content="hi" />'
- b"</head></html>"
- )
+ self.lookups["publish.twitter.com"] = [(IPv4Address, "10.1.2.3")]
+ self.lookups["cdn.twitter.com"] = [(IPv4Address, "10.1.2.3")]
- channel = self.make_request(
- "GET",
- "preview_url?url=http://twitter.com/matrixdotorg/status/12345",
- shorthand=False,
- await_result=False,
- )
- self.pump()
-
- client = self.reactor.tcpClients[0][2].buildProtocol(None)
- server = AccumulatingProtocol()
- server.makeConnection(FakeTransport(client, self.reactor))
- client.makeConnection(FakeTransport(server, self.reactor))
- client.dataReceived(
- (
- b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
- b'Content-Type: application/json; charset="utf8"\r\n\r\n'
- )
- % (len(oembed_content),)
- + oembed_content
- )
+ result = {
+ "version": "1.0",
+ "type": "photo",
+ "url": "http://cdn.twitter.com/matrixdotorg",
+ }
+ oembed_content = json.dumps(result).encode("utf-8")
- self.pump()
-
- client = self.reactor.tcpClients[1][2].buildProtocol(None)
- server = AccumulatingProtocol()
- server.makeConnection(FakeTransport(client, self.reactor))
- client.makeConnection(FakeTransport(server, self.reactor))
- client.dataReceived(
- (
- b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
- b'Content-Type: text/html; charset="utf8"\r\n\r\n'
- )
- % (len(end_content),)
- + end_content
+ end_content = (
+ b"<html><head>"
+ b"<title>Some Title</title>"
+ b'<meta property="og:description" content="hi" />'
+ b"</head></html>"
+ )
+
+ channel = self.make_request(
+ "GET",
+ "preview_url?url=http://twitter.com/matrixdotorg/status/12345",
+ shorthand=False,
+ await_result=False,
+ )
+ self.pump()
+
+ client = self.reactor.tcpClients[0][2].buildProtocol(None)
+ server = AccumulatingProtocol()
+ server.makeConnection(FakeTransport(client, self.reactor))
+ client.makeConnection(FakeTransport(server, self.reactor))
+ client.dataReceived(
+ (
+ b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
+ b'Content-Type: application/json; charset="utf8"\r\n\r\n'
)
+ % (len(oembed_content),)
+ + oembed_content
+ )
- self.pump()
+ self.pump()
- self.assertEqual(channel.code, 200)
- self.assertEqual(
- channel.json_body, {"og:title": "Some Title", "og:description": "hi"}
+ client = self.reactor.tcpClients[1][2].buildProtocol(None)
+ server = AccumulatingProtocol()
+ server.makeConnection(FakeTransport(client, self.reactor))
+ client.makeConnection(FakeTransport(server, self.reactor))
+ client.dataReceived(
+ (
+ b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
+ b'Content-Type: text/html; charset="utf8"\r\n\r\n'
)
+ % (len(end_content),)
+ + end_content
+ )
+
+ self.pump()
+
+ self.assertEqual(channel.code, 200)
+ self.assertEqual(
+ channel.json_body, {"og:title": "Some Title", "og:description": "hi"}
+ )
def test_oembed_rich(self):
"""Test an oEmbed endpoint which returns HTML content via the 'rich' type."""
- # Route the HTTP version to an HTTP endpoint so that the tests work.
- with patch.dict(
- "synapse.rest.media.v1.preview_url_resource._oembed_patterns",
- {
- re.compile(
- r"http://twitter\.com/.+/status/.+"
- ): "http://publish.twitter.com/oembed",
- },
- clear=True,
- ):
-
- self.lookups["publish.twitter.com"] = [(IPv4Address, "10.1.2.3")]
-
- result = {
- "version": "1.0",
- "type": "rich",
- "html": "<div>Content Preview</div>",
- }
- end_content = json.dumps(result).encode("utf-8")
-
- channel = self.make_request(
- "GET",
- "preview_url?url=http://twitter.com/matrixdotorg/status/12345",
- shorthand=False,
- await_result=False,
- )
- self.pump()
-
- client = self.reactor.tcpClients[0][2].buildProtocol(None)
- server = AccumulatingProtocol()
- server.makeConnection(FakeTransport(client, self.reactor))
- client.makeConnection(FakeTransport(server, self.reactor))
- client.dataReceived(
- (
- b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
- b'Content-Type: application/json; charset="utf8"\r\n\r\n'
- )
- % (len(end_content),)
- + end_content
+ self.lookups["publish.twitter.com"] = [(IPv4Address, "10.1.2.3")]
+
+ result = {
+ "version": "1.0",
+ "type": "rich",
+ "html": "<div>Content Preview</div>",
+ }
+ end_content = json.dumps(result).encode("utf-8")
+
+ channel = self.make_request(
+ "GET",
+ "preview_url?url=http://twitter.com/matrixdotorg/status/12345",
+ shorthand=False,
+ await_result=False,
+ )
+ self.pump()
+
+ client = self.reactor.tcpClients[0][2].buildProtocol(None)
+ server = AccumulatingProtocol()
+ server.makeConnection(FakeTransport(client, self.reactor))
+ client.makeConnection(FakeTransport(server, self.reactor))
+ client.dataReceived(
+ (
+ b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
+ b'Content-Type: application/json; charset="utf8"\r\n\r\n'
)
+ % (len(end_content),)
+ + end_content
+ )
- self.pump()
- self.assertEqual(channel.code, 200)
- self.assertEqual(
- channel.json_body,
- {"og:title": None, "og:description": "Content Preview"},
+ self.pump()
+ self.assertEqual(channel.code, 200)
+ self.assertEqual(
+ channel.json_body,
+ {"og:title": None, "og:description": "Content Preview"},
+ )
+
+ def test_oembed_format(self):
+ """Test an oEmbed endpoint which requires the format in the URL."""
+ self.lookups["www.hulu.com"] = [(IPv4Address, "10.1.2.3")]
+
+ result = {
+ "version": "1.0",
+ "type": "rich",
+ "html": "<div>Content Preview</div>",
+ }
+ end_content = json.dumps(result).encode("utf-8")
+
+ channel = self.make_request(
+ "GET",
+ "preview_url?url=http://www.hulu.com/watch/12345",
+ shorthand=False,
+ await_result=False,
+ )
+ self.pump()
+
+ client = self.reactor.tcpClients[0][2].buildProtocol(None)
+ server = AccumulatingProtocol()
+ server.makeConnection(FakeTransport(client, self.reactor))
+ client.makeConnection(FakeTransport(server, self.reactor))
+ client.dataReceived(
+ (
+ b"HTTP/1.0 200 OK\r\nContent-Length: %d\r\n"
+ b'Content-Type: application/json; charset="utf8"\r\n\r\n'
)
+ % (len(end_content),)
+ + end_content
+ )
+
+ self.pump()
+
+ # The {format} should have been turned into json.
+ self.assertIn(b"/api/oembed.json", server.data)
+ # A URL parameter of format=json should be provided.
+ self.assertIn(b"format=json", server.data)
+
+ self.assertEqual(channel.code, 200)
+ self.assertEqual(
+ channel.json_body,
+ {"og:title": None, "og:description": "Content Preview"},
+ )
|
