paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-16 03:41:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

test_link_embed: Use responses module to mock HTTP responses.

Browse Source
This commit is contained in:
akshatdalton
2021-06-08 22:18:46 +00:00
committed by Tim Abbott
parent bef7120390
commit 3cc6ff57db
2 changed files with 121 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
zerver/lib/test_helpers.py
View File

@@ -319,25 +319,6 @@ class HostRequestMock:
return self.host return self.host
class MockPythonResponse:
def __init__(
self, text: str, status_code: int, headers: Optional[Dict[str, str]] = None
) -> None:
self.content = text.encode()
self.text = text
self.status_code = status_code
if headers is None:
headers = {"content-type": "text/html"}
self.headers = headers
@property
def ok(self) -> bool:
return self.status_code == 200
def iter_content(self, n: int) -> Generator[str, Any, None]:
yield self.text[:n]
INSTRUMENTING = os.environ.get("TEST_INSTRUMENT_URL_COVERAGE", "") == "TRUE" INSTRUMENTING = os.environ.get("TEST_INSTRUMENT_URL_COVERAGE", "") == "TRUE"
INSTRUMENTED_CALLS: List[Dict[str, Any]] = [] INSTRUMENTED_CALLS: List[Dict[str, Any]] = []

224
zerver/tests/test_link_embed.py
View File

@@ -1,16 +1,19 @@
from typing import Any, Callable, Dict, Optional from collections import OrderedDict
from typing import Any, Optional, Union
from unittest import mock from unittest import mock
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
import orjson import responses
from django.test import override_settings from django.test import override_settings
from django.utils.html import escape from django.utils.html import escape
from pyoembed.providers import get_provider
from requests.exceptions import ConnectionError from requests.exceptions import ConnectionError
from zerver.lib.actions import queue_json_publish from zerver.lib.actions import queue_json_publish
from zerver.lib.cache import NotFoundInCache, cache_set, preview_url_cache_key from zerver.lib.cache import NotFoundInCache, cache_set, preview_url_cache_key
from zerver.lib.camo import get_camo_url from zerver.lib.camo import get_camo_url
from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import MockPythonResponse, mock_queue_publish from zerver.lib.test_helpers import mock_queue_publish
from zerver.lib.url_preview.oembed import get_oembed_data, strip_cdata from zerver.lib.url_preview.oembed import get_oembed_data, strip_cdata
from zerver.lib.url_preview.parsers import GenericParser, OpenGraphParser from zerver.lib.url_preview.parsers import GenericParser, OpenGraphParser
from zerver.lib.url_preview.preview import get_link_embed_data, link_embed_data_from_cache from zerver.lib.url_preview.preview import get_link_embed_data, link_embed_data_from_cache
@@ -33,13 +36,25 @@ TEST_CACHES = {
} }
def reconstruct_url(url: str, maxwidth: int = 640, maxheight: int = 480) -> str:
# The following code is taken from
# https://github.com/rafaelmartins/pyoembed/blob/master/pyoembed/__init__.py.
# This is a helper function which will be indirectly use to mock the HTTP responses.
provider = get_provider(str(url))
oembed_url = provider.oembed_url(url)
scheme, netloc, path, query_string, fragment = urlsplit(oembed_url)
query_params = OrderedDict(parse_qsl(query_string))
query_params["maxwidth"] = str(maxwidth)
query_params["maxheight"] = str(maxheight)
final_url = urlunsplit((scheme, netloc, path, urlencode(query_params, True), fragment))
return final_url
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
class OembedTestCase(ZulipTestCase): class OembedTestCase(ZulipTestCase):
@mock.patch("pyoembed.requests.get") @responses.activate
def test_present_provider(self, get: Any) -> None: def test_present_provider(self) -> None:
get.return_value = response = mock.Mock()
response.headers = {"content-type": "application/json"}
response.ok = True
response_data = { response_data = {
"type": "rich", "type": "rich",
"thumbnail_url": "https://scontent.cdninstagram.com/t51.2885-15/n.jpg", "thumbnail_url": "https://scontent.cdninstagram.com/t51.2885-15/n.jpg",
@@ -51,19 +66,23 @@ class OembedTestCase(ZulipTestCase):
"width": 658, "width": 658,
"height": 400, "height": 400,
} }
response.text = orjson.dumps(response_data).decode()
url = "http://instagram.com/p/BLtI2WdAymy" url = "http://instagram.com/p/BLtI2WdAymy"
reconstructed_url = reconstruct_url(url)
responses.add(
responses.GET,
reconstructed_url,
json=response_data,
status=200,
)
data = get_oembed_data(url) data = get_oembed_data(url)
self.assertIsInstance(data, dict) self.assertIsInstance(data, dict)
self.assertIn("title", data) self.assertIn("title", data)
assert data is not None # allow mypy to infer data is indexable assert data is not None # allow mypy to infer data is indexable
self.assertEqual(data["title"], response_data["title"]) self.assertEqual(data["title"], response_data["title"])
@mock.patch("pyoembed.requests.get") @responses.activate
def test_photo_provider(self, get: Any) -> None: def test_photo_provider(self) -> None:
get.return_value = response = mock.Mock()
response.headers = {"content-type": "application/json"}
response.ok = True
response_data = { response_data = {
"type": "photo", "type": "photo",
"thumbnail_url": "https://scontent.cdninstagram.com/t51.2885-15/n.jpg", "thumbnail_url": "https://scontent.cdninstagram.com/t51.2885-15/n.jpg",
@@ -76,8 +95,15 @@ class OembedTestCase(ZulipTestCase):
"width": 658, "width": 658,
"height": 400, "height": 400,
} }
response.text = orjson.dumps(response_data).decode()
url = "http://imgur.com/photo/158727223" url = "http://imgur.com/photo/158727223"
reconstructed_url = reconstruct_url(url)
responses.add(
responses.GET,
reconstructed_url,
json=response_data,
status=200,
)
data = get_oembed_data(url) data = get_oembed_data(url)
self.assertIsInstance(data, dict) self.assertIsInstance(data, dict)
self.assertIn("title", data) self.assertIn("title", data)
@@ -85,11 +111,8 @@ class OembedTestCase(ZulipTestCase):
self.assertEqual(data["title"], response_data["title"]) self.assertEqual(data["title"], response_data["title"])
self.assertTrue(data["oembed"]) self.assertTrue(data["oembed"])
@mock.patch("pyoembed.requests.get") @responses.activate
def test_video_provider(self, get: Any) -> None: def test_video_provider(self) -> None:
get.return_value = response = mock.Mock()
response.headers = {"content-type": "application/json"}
response.ok = True
response_data = { response_data = {
"type": "video", "type": "video",
"thumbnail_url": "https://scontent.cdninstagram.com/t51.2885-15/n.jpg", "thumbnail_url": "https://scontent.cdninstagram.com/t51.2885-15/n.jpg",
@@ -101,29 +124,39 @@ class OembedTestCase(ZulipTestCase):
"width": 658, "width": 658,
"height": 400, "height": 400,
} }
response.text = orjson.dumps(response_data).decode()
url = "http://blip.tv/video/158727223" url = "http://blip.tv/video/158727223"
reconstructed_url = reconstruct_url(url)
responses.add(
responses.GET,
reconstructed_url,
json=response_data,
status=200,
)
data = get_oembed_data(url) data = get_oembed_data(url)
self.assertIsInstance(data, dict) self.assertIsInstance(data, dict)
self.assertIn("title", data) self.assertIn("title", data)
assert data is not None # allow mypy to infer data is indexable assert data is not None # allow mypy to infer data is indexable
self.assertEqual(data["title"], response_data["title"]) self.assertEqual(data["title"], response_data["title"])
@mock.patch("pyoembed.requests.get") @responses.activate
def test_error_request(self, get: Any) -> None: def test_error_request(self) -> None:
get.return_value = response = mock.Mock()
response.ok = False
url = "http://instagram.com/p/BLtI2WdAymy" url = "http://instagram.com/p/BLtI2WdAymy"
reconstructed_url = reconstruct_url(url)
responses.add(responses.GET, reconstructed_url, status=400)
data = get_oembed_data(url) data = get_oembed_data(url)
self.assertIsNone(data) self.assertIsNone(data)
@mock.patch("pyoembed.requests.get") @responses.activate
def test_invalid_json_in_response(self, get: Any) -> None: def test_invalid_json_in_response(self) -> None:
get.return_value = response = mock.Mock()
response.headers = {"content-type": "application/json"}
response.ok = True
response.text = "{invalid json}"
url = "http://instagram.com/p/BLtI2WdAymy" url = "http://instagram.com/p/BLtI2WdAymy"
reconstructed_url = reconstruct_url(url)
responses.add(
responses.GET,
reconstructed_url,
json="{invalid json}",
status=200,
)
data = get_oembed_data(url) data = get_oembed_data(url)
self.assertIsNone(data) self.assertIsNone(data)
@@ -298,17 +331,18 @@ class PreviewTestCase(ZulipTestCase):
def create_mock_response( def create_mock_response(
cls, cls,
url: str, url: str,
status: int = 200,
relative_url: bool = False, relative_url: bool = False,
headers: Optional[Dict[str, str]] = None, content_type: str = "text/html",
html: Optional[str] = None, body: Optional[Union[str, ConnectionError]] = None,
) -> Callable[..., MockPythonResponse]: ) -> None:
if html is None: if body is None:
html = cls.open_graph_html body = cls.open_graph_html
if relative_url is True: if relative_url is True and isinstance(body, str):
html = html.replace("http://ia.media-imdb.com", "") body = body.replace("http://ia.media-imdb.com", "")
response = MockPythonResponse(html, 200, headers) responses.add(responses.GET, url, body=body, status=status, content_type=content_type)
return lambda k, **kwargs: {url: response}.get(k, MockPythonResponse("", 404, headers))
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_edit_message_history(self) -> None: def test_edit_message_history(self) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@@ -318,7 +352,7 @@ class PreviewTestCase(ZulipTestCase):
) )
url = "http://test.org/" url = "http://test.org/"
mocked_response = mock.Mock(side_effect=self.create_mock_response(url)) self.create_mock_response(url)
with mock_queue_publish("zerver.lib.actions.queue_json_publish") as patched: with mock_queue_publish("zerver.lib.actions.queue_json_publish") as patched:
result = self.client_patch( result = self.client_patch(
@@ -335,9 +369,7 @@ class PreviewTestCase(ZulipTestCase):
event = patched.call_args[0][1] event = patched.call_args[0][1]
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
self.assertTrue( self.assertTrue(
"INFO:root:Time spent on get_link_embed_data for http://test.org/: " "INFO:root:Time spent on get_link_embed_data for http://test.org/: "
@@ -348,6 +380,7 @@ class PreviewTestCase(ZulipTestCase):
msg = Message.objects.select_related("sender").get(id=msg_id) msg = Message.objects.select_related("sender").get(id=msg_id)
self.assertIn(embedded_link, msg.rendered_content) self.assertIn(embedded_link, msg.rendered_content)
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def _send_message_with_test_org_url( def _send_message_with_test_org_url(
self, sender: UserProfile, queue_should_run: bool = True, relative_url: bool = False self, sender: UserProfile, queue_should_run: bool = True, relative_url: bool = False
@@ -374,16 +407,11 @@ class PreviewTestCase(ZulipTestCase):
msg = Message.objects.select_related("sender").get(id=msg_id) msg = Message.objects.select_related("sender").get(id=msg_id)
self.assertNotIn(f'<a href="{url}" title="The Rock">The Rock</a>', msg.rendered_content) self.assertNotIn(f'<a href="{url}" title="The Rock">The Rock</a>', msg.rendered_content)
# Mock the network request result so the test can be fast without Internet self.create_mock_response(url, relative_url=relative_url)
mocked_response = mock.Mock(
side_effect=self.create_mock_response(url, relative_url=relative_url)
)
# Run the queue processor to potentially rerender things # Run the queue processor to potentially rerender things
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
self.assertTrue( self.assertTrue(
"INFO:root:Time spent on get_link_embed_data for http://test.org/: " "INFO:root:Time spent on get_link_embed_data for http://test.org/: "
@@ -393,6 +421,7 @@ class PreviewTestCase(ZulipTestCase):
msg = Message.objects.select_related("sender").get(id=msg_id) msg = Message.objects.select_related("sender").get(id=msg_id)
return msg return msg
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_message_update_race_condition(self) -> None: def test_message_update_race_condition(self) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@@ -409,16 +438,11 @@ class PreviewTestCase(ZulipTestCase):
event = patched.call_args[0][1] event = patched.call_args[0][1]
def wrapped_queue_json_publish(*args: Any, **kwargs: Any) -> None: def wrapped_queue_json_publish(*args: Any, **kwargs: Any) -> None:
# Mock the network request result so the test can be fast without Internet self.create_mock_response(original_url)
mocked_response_original = mock.Mock( self.create_mock_response(edited_url)
side_effect=self.create_mock_response(original_url)
)
mocked_response_edited = mock.Mock(side_effect=self.create_mock_response(edited_url))
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response_original), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
# Run the queue processor. This will simulate the event for original_url being # Run the queue processor. This will simulate the event for original_url being
# processed after the message has been edited. # processed after the message has been edited.
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
@@ -432,12 +456,11 @@ class PreviewTestCase(ZulipTestCase):
self.assertNotIn( self.assertNotIn(
f'<a href="{original_url}" title="The Rock">The Rock</a>', msg.rendered_content f'<a href="{original_url}" title="The Rock">The Rock</a>', msg.rendered_content
) )
mocked_response_edited.assert_not_called()
self.assertTrue(responses.assert_call_count(edited_url, 0))
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response_edited), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
# Now proceed with the original queue_json_publish and call the # Now proceed with the original queue_json_publish and call the
# up-to-date event for edited_url. # up-to-date event for edited_url.
queue_json_publish(*args, **kwargs) queue_json_publish(*args, **kwargs)
@@ -516,6 +539,7 @@ class PreviewTestCase(ZulipTestCase):
) )
self.assertEqual(msg.rendered_content, with_preview_relative) self.assertEqual(msg.rendered_content, with_preview_relative)
@responses.activate
def test_http_error_get_data(self) -> None: def test_http_error_get_data(self) -> None:
url = "http://test.org/" url = "http://test.org/"
msg_id = self.send_personal_message( msg_id = self.send_personal_message(
@@ -530,10 +554,11 @@ class PreviewTestCase(ZulipTestCase):
"message_realm_id": msg.sender.realm_id, "message_realm_id": msg.sender.realm_id,
"message_content": url, "message_content": url,
} }
self.create_mock_response(url, body=ConnectionError())
with self.settings(INLINE_URL_EMBED_PREVIEW=True, TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(INLINE_URL_EMBED_PREVIEW=True, TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch( with self.assertLogs(level="INFO") as info_logs:
"requests.get", mock.Mock(side_effect=ConnectionError())
), self.assertLogs(level="INFO") as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
self.assertTrue( self.assertTrue(
"INFO:root:Time spent on get_link_embed_data for http://test.org/: " "INFO:root:Time spent on get_link_embed_data for http://test.org/: "
@@ -562,6 +587,7 @@ class PreviewTestCase(ZulipTestCase):
cache_set(key, link_embed_data, "database") cache_set(key, link_embed_data, "database")
self.assertEqual(link_embed_data, link_embed_data_from_cache(url)) self.assertEqual(link_embed_data, link_embed_data_from_cache(url))
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_link_preview_non_html_data(self) -> None: def test_link_preview_non_html_data(self) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@@ -574,13 +600,11 @@ class PreviewTestCase(ZulipTestCase):
self.assertEqual(queue, "embed_links") self.assertEqual(queue, "embed_links")
event = patched.call_args[0][1] event = patched.call_args[0][1]
headers = {"content-type": "application/octet-stream"} content_type = "application/octet-stream"
mocked_response = mock.Mock(side_effect=self.create_mock_response(url, headers=headers)) self.create_mock_response(url, content_type=content_type)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
cached_data = link_embed_data_from_cache(url) cached_data = link_embed_data_from_cache(url)
self.assertTrue( self.assertTrue(
@@ -595,6 +619,7 @@ class PreviewTestCase(ZulipTestCase):
msg.rendered_content, msg.rendered_content,
) )
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_link_preview_no_open_graph_image(self) -> None: def test_link_preview_no_open_graph_image(self) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@@ -611,11 +636,9 @@ class PreviewTestCase(ZulipTestCase):
html = "\n".join( html = "\n".join(
line for line in self.open_graph_html.splitlines() if "og:image" not in line line for line in self.open_graph_html.splitlines() if "og:image" not in line
) )
mocked_response = mock.Mock(side_effect=self.create_mock_response(url, html=html)) self.create_mock_response(url, body=html)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
cached_data = link_embed_data_from_cache(url) cached_data = link_embed_data_from_cache(url)
self.assertTrue( self.assertTrue(
@@ -631,6 +654,7 @@ class PreviewTestCase(ZulipTestCase):
msg.rendered_content, msg.rendered_content,
) )
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_link_preview_open_graph_image_missing_content(self) -> None: def test_link_preview_open_graph_image_missing_content(self) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@@ -648,11 +672,9 @@ class PreviewTestCase(ZulipTestCase):
line if "og:image" not in line else '<meta property="og:image"/>' line if "og:image" not in line else '<meta property="og:image"/>'
for line in self.open_graph_html.splitlines() for line in self.open_graph_html.splitlines()
) )
mocked_response = mock.Mock(side_effect=self.create_mock_response(url, html=html)) self.create_mock_response(url, body=html)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
cached_data = link_embed_data_from_cache(url) cached_data = link_embed_data_from_cache(url)
self.assertTrue( self.assertTrue(
@@ -668,6 +690,7 @@ class PreviewTestCase(ZulipTestCase):
msg.rendered_content, msg.rendered_content,
) )
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_link_preview_no_content_type_header(self) -> None: def test_link_preview_no_content_type_header(self) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@@ -680,12 +703,9 @@ class PreviewTestCase(ZulipTestCase):
self.assertEqual(queue, "embed_links") self.assertEqual(queue, "embed_links")
event = patched.call_args[0][1] event = patched.call_args[0][1]
headers = {"content-type": ""} # No content type header self.create_mock_response(url)
mocked_response = mock.Mock(side_effect=self.create_mock_response(url, headers=headers))
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
data = link_embed_data_from_cache(url) data = link_embed_data_from_cache(url)
self.assertTrue( self.assertTrue(
@@ -700,6 +720,7 @@ class PreviewTestCase(ZulipTestCase):
self.assertIn(data["title"], msg.rendered_content) self.assertIn(data["title"], msg.rendered_content)
self.assertIn(data["image"], msg.rendered_content) self.assertIn(data["image"], msg.rendered_content)
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_valid_content_type_error_get_data(self) -> None: def test_valid_content_type_error_get_data(self) -> None:
url = "http://test.org/" url = "http://test.org/"
@@ -717,6 +738,8 @@ class PreviewTestCase(ZulipTestCase):
"message_content": url, "message_content": url,
} }
self.create_mock_response(url, body=ConnectionError())
with mock.patch( with mock.patch(
"zerver.lib.url_preview.preview.get_oembed_data", "zerver.lib.url_preview.preview.get_oembed_data",
side_effect=lambda *args, **kwargs: None, side_effect=lambda *args, **kwargs: None,
@@ -725,9 +748,7 @@ class PreviewTestCase(ZulipTestCase):
"zerver.lib.url_preview.preview.valid_content_type", side_effect=lambda k: True "zerver.lib.url_preview.preview.valid_content_type", side_effect=lambda k: True
): ):
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch( with self.assertLogs(level="INFO") as info_logs:
"requests.get", mock.Mock(side_effect=ConnectionError())
), self.assertLogs(level="INFO") as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
self.assertTrue( self.assertTrue(
"INFO:root:Time spent on get_link_embed_data for http://test.org/: " "INFO:root:Time spent on get_link_embed_data for http://test.org/: "
@@ -742,6 +763,7 @@ class PreviewTestCase(ZulipTestCase):
'<p><a href="http://test.org/">http://test.org/</a></p>', msg.rendered_content '<p><a href="http://test.org/">http://test.org/</a></p>', msg.rendered_content
) )
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_invalid_url(self) -> None: def test_invalid_url(self) -> None:
url = "http://test.org/" url = "http://test.org/"
@@ -760,11 +782,9 @@ class PreviewTestCase(ZulipTestCase):
"message_content": error_url, "message_content": error_url,
} }
mocked_response = mock.Mock(side_effect=self.create_mock_response(url)) self.create_mock_response(error_url, status=404)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
FetchLinksEmbedData().consume(event) FetchLinksEmbedData().consume(event)
self.assertTrue( self.assertTrue(
"INFO:root:Time spent on get_link_embed_data for http://test.org/x: " "INFO:root:Time spent on get_link_embed_data for http://test.org/x: "
@@ -778,7 +798,9 @@ class PreviewTestCase(ZulipTestCase):
self.assertEqual( self.assertEqual(
'<p><a href="http://test.org/x">http://test.org/x</a></p>', msg.rendered_content '<p><a href="http://test.org/x">http://test.org/x</a></p>', msg.rendered_content
) )
self.assertTrue(responses.assert_call_count(url, 0))
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_safe_oembed_html_url(self) -> None: def test_safe_oembed_html_url(self) -> None:
url = "http://test.org/" url = "http://test.org/"
@@ -802,11 +824,9 @@ class PreviewTestCase(ZulipTestCase):
"type": "video", "type": "video",
"image": f"{url}/image.png", "image": f"{url}/image.png",
} }
mocked_response = mock.Mock(side_effect=self.create_mock_response(url)) self.create_mock_response(url)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
with mock.patch( with mock.patch(
"zerver.lib.url_preview.preview.get_oembed_data", "zerver.lib.url_preview.preview.get_oembed_data",
lambda *args, **kwargs: mocked_data, lambda *args, **kwargs: mocked_data,
@@ -822,6 +842,7 @@ class PreviewTestCase(ZulipTestCase):
msg.refresh_from_db() msg.refresh_from_db()
self.assertIn('a data-id="{}"'.format(escape(mocked_data["html"])), msg.rendered_content) self.assertIn('a data-id="{}"'.format(escape(mocked_data["html"])), msg.rendered_content)
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_youtube_url_title_replaces_url(self) -> None: def test_youtube_url_title_replaces_url(self) -> None:
url = "https://www.youtube.com/watch?v=eSJTXC7Ixgg" url = "https://www.youtube.com/watch?v=eSJTXC7Ixgg"
@@ -840,11 +861,9 @@ class PreviewTestCase(ZulipTestCase):
} }
mocked_data = {"title": "Clearer Code at Scale - Static Types at Zulip and Dropbox"} mocked_data = {"title": "Clearer Code at Scale - Static Types at Zulip and Dropbox"}
mocked_response = mock.Mock(side_effect=self.create_mock_response(url)) self.create_mock_response(url)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
with mock.patch( with mock.patch(
"zerver.lib.markdown.link_preview.link_embed_data_from_cache", "zerver.lib.markdown.link_preview.link_embed_data_from_cache",
lambda *args, **kwargs: mocked_data, lambda *args, **kwargs: mocked_data,
@@ -859,6 +878,7 @@ class PreviewTestCase(ZulipTestCase):
expected_content = f"""<p><a href="https://www.youtube.com/watch?v=eSJTXC7Ixgg">YouTube - Clearer Code at Scale - Static Types at Zulip and Dropbox</a></p>\n<div class="youtube-video message_inline_image"><a data-id="eSJTXC7Ixgg" href="https://www.youtube.com/watch?v=eSJTXC7Ixgg"><img src="{get_camo_url("https://i.ytimg.com/vi/eSJTXC7Ixgg/default.jpg")}"></a></div>""" expected_content = f"""<p><a href="https://www.youtube.com/watch?v=eSJTXC7Ixgg">YouTube - Clearer Code at Scale - Static Types at Zulip and Dropbox</a></p>\n<div class="youtube-video message_inline_image"><a data-id="eSJTXC7Ixgg" href="https://www.youtube.com/watch?v=eSJTXC7Ixgg"><img src="{get_camo_url("https://i.ytimg.com/vi/eSJTXC7Ixgg/default.jpg")}"></a></div>"""
self.assertEqual(expected_content, msg.rendered_content) self.assertEqual(expected_content, msg.rendered_content)
@responses.activate
@override_settings(INLINE_URL_EMBED_PREVIEW=True) @override_settings(INLINE_URL_EMBED_PREVIEW=True)
def test_custom_title_replaces_youtube_url_title(self) -> None: def test_custom_title_replaces_youtube_url_title(self) -> None:
url = "[YouTube link](https://www.youtube.com/watch?v=eSJTXC7Ixgg)" url = "[YouTube link](https://www.youtube.com/watch?v=eSJTXC7Ixgg)"
@@ -877,11 +897,9 @@ class PreviewTestCase(ZulipTestCase):
} }
mocked_data = {"title": "Clearer Code at Scale - Static Types at Zulip and Dropbox"} mocked_data = {"title": "Clearer Code at Scale - Static Types at Zulip and Dropbox"}
mocked_response = mock.Mock(side_effect=self.create_mock_response(url)) self.create_mock_response(url)
with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES): with self.settings(TEST_SUITE=False, CACHES=TEST_CACHES):
with mock.patch("requests.get", mocked_response), self.assertLogs( with self.assertLogs(level="INFO") as info_logs:
level="INFO"
) as info_logs:
with mock.patch( with mock.patch(
"zerver.lib.markdown.link_preview.link_embed_data_from_cache", "zerver.lib.markdown.link_preview.link_embed_data_from_cache",
lambda *args, **kwargs: mocked_data, lambda *args, **kwargs: mocked_data,