[WIP] Handle media links

tests/helper/test_helper.py

@@ -3,10 +3,14 @@
 import zulipterminal.helper
 from zulipterminal.helper import (
     canonicalize_color, classify_unread_counts, display_error_if_present,
-    hash_util_decode, index_messages, notify, powerset,
+    hash_util_decode, index_messages, notify, open_media, powerset,
+    process_media,
 )
 
 
+SERVER_URL = 'https://chat.zulip.zulip'
+
+
 def test_index_messages_narrow_all_messages(mocker,
                                             messages_successful_response,
                                             index_all_messages,
@@ -314,3 +318,52 @@ def test_hash_util_decode(quoted_string, expected_unquoted_string):
     return_value = hash_util_decode(quoted_string)
 
     assert return_value == expected_unquoted_string
+
+
+def test_process_media(mocker, media_path='/tmp/zt-somerandomtext-image.png',
+                       media_link=SERVER_URL + '/user_uploads/path/image.png'):
+    mocker.patch('zulipterminal.helper.requests')
+    mocker.patch('zulipterminal.helper.open')
+    (mocker.patch('zulipterminal.helper.NamedTemporaryFile').return_value
+     .__enter__.return_value.name) = media_path
+    # The command, which is platform dependent, does not matter for the test.
+    mocker.patch('zulipterminal.helper.LINUX', True)
+    controller = mocker.Mock()
+
+    process_media(controller, media_link)
+
+    controller.show_media_confirmation_popup.assert_called_once_with(
+        open_media, 'xdg-open', media_path
+    )
+
+
+@pytest.mark.parametrize('returncode, error', [
+    (0, []),
+    (1, [' The command ', ('bold', 'some_command'), ' did not run successfully'
+         '. Exited with ', ('bold', '1')]),
+])
+def test_open_media(mocker, returncode, error, command='some_command',
+                    media_path='/tmp/zt-somerandomtext-image.png'):
+    mocked_run = mocker.patch('zulipterminal.helper.subprocess.run')
+    mocked_run.return_value.returncode = returncode
+    controller = mocker.Mock()
+
+    open_media(controller, command, media_path)
+
+    assert mocked_run.called
+    if error:
+        controller.view.set_footer_text.assert_called_once_with(error,
+                                                                duration=3)
+
+
+def test_open_media_exception(mocker, command='some_command',
+                              media_path='/tmp/zt-somerandomtext-image.png',
+                              error=[' The command ', ('bold', 'some_command'),
+                                     ' could not be found']):
+    mocker.patch('zulipterminal.helper.subprocess.run',
+                 side_effect=FileNotFoundError())
+    controller = mocker.Mock()
+
+    open_media(controller, command, media_path)
+
+    controller.view.set_footer_text.assert_called_once_with(error, duration=3)

tests/ui_tools/test_buttons.py

@@ -59,28 +59,33 @@ def test_update_widget(self, mocker, caption, expected_cursor_position,
     @pytest.mark.parametrize([
             'link',
             'handle_narrow_link_called',
+            'process_media_called',
         ],
         [
-            (SERVER_URL + '/#narrow/stream/1-Stream-1', True),
-            (SERVER_URL + '/user_uploads/some/path/image.png', False),
-            ('https://foo.com', False),
+            (SERVER_URL + '/#narrow/stream/1-Stream-1', True, False),
+            (SERVER_URL + '/user_uploads/some/path/image.png', False, True),
+            ('https://foo.com', False, False),
         ],
         ids=[
             'internal_narrow_link',
             'internal_media_link',
             'external_link',
         ]
     )
-    def test_handle_link(self, mocker, link, handle_narrow_link_called):
+    def test_handle_link(self, mocker, link, handle_narrow_link_called,
+                         process_media_called):
         self.controller.model.server_url = SERVER_URL
+        self.controller.loop.widget = mocker.Mock(spec=Overlay)
         self.handle_narrow_link = (
             mocker.patch(BUTTONS + '.MessageLinkButton.handle_narrow_link')
         )
+        self.process_media = mocker.patch(BUTTONS + '.process_media')
         mocked_button = self.message_link_button(link=link)
 
         mocked_button.handle_link()
 
         assert self.handle_narrow_link.called == handle_narrow_link_called
+        assert self.process_media.called == process_media_called
 
     @pytest.mark.parametrize('stream_data, expected_response', [
             ('206-zulip-terminal', dict(stream_id=206, stream_name=None)),

zulipterminal/core.py

@@ -173,6 +173,17 @@ def show_edit_history(
                             'Edit History (up/down scrolls)')
         )
 
+    def show_media_confirmation_popup(self, func: Any, command: str,
+                                      media_path: str) -> None:
+        callback = partial(func, self, command, media_path)
+        question = urwid.Text([
+            'Your requested media has been downloaded at ',
+            ('bold', media_path),
+            '. Do you want to view it?'
+        ])
+        self.loop.widget = PopUpConfirmationView(self, question, callback,
+                                                 location='center')
+
     def search_messages(self, text: str) -> None:
         # Search for a text in messages
         self.model.index['search'].clear()

zulipterminal/helper.py

@@ -6,6 +6,7 @@
 from functools import wraps
 from itertools import chain, combinations
 from re import ASCII, match
+from tempfile import NamedTemporaryFile
 from threading import Thread
 from typing import (
     Any, Callable, DefaultDict, Dict, FrozenSet, Iterable, List, Set, Tuple,
@@ -14,6 +15,7 @@
 from urllib.parse import unquote
 
 import lxml.html
+import requests
 from typing_extensions import TypedDict
 
 
@@ -647,3 +649,53 @@ def hash_util_decode(string: str) -> str:
     # Acknowledge custom string replacements in zulip/zulip's
     # zerver/lib/url_encoding.py before unquote.
     return unquote(string.replace('.', '%'))
+
+
+@asynch
+def process_media(controller: Any, media_link: str) -> None:
+    media_name = media_link.split('/')[-1]
+    client = controller.client
+    auth = requests.auth.HTTPBasicAuth(client.email, client.api_key)
+
+    with requests.get(media_link, auth=auth, stream=True) as r:
+        r.raise_for_status()
+        with NamedTemporaryFile(mode='wb', delete=False, prefix='zt-',
+                                suffix='-{}'.format(media_name)) as f:
+            media_path = f.name
+            for chunk in r.iter_content(chunk_size=8192):
+                if chunk:  # Filter out keep-alive new chunks.
+                    f.write(chunk)
+                    controller.view.set_footer_text([
+                        ' Downloading ', ('bold', media_name),
+                    ])
+        controller.view.set_footer_text([' Downloaded ', ('bold', media_name)],
+                                        duration=3)
+
+    if LINUX:
+        command = 'xdg-open'
+    elif WSL:
+        command = 'explorer'  # Needs testing.
+    elif MACOS:
+        command = 'open'
+
+    controller.show_media_confirmation_popup(open_media, command, media_path)
+
+
+@asynch
+def open_media(controller: Any, command: str, media_path: str) -> None:
+    error = []
+    try:
+        process = subprocess.run([command, media_path],
+                                 stdout=subprocess.DEVNULL,
+                                 stderr=subprocess.DEVNULL)
+        exit_status = process.returncode
+        if exit_status != 0:
+            error = [
+                ' The command ', ('bold', command), ' did not run successfully'
+                '. Exited with ', ('bold', str(exit_status)),
+            ]
+    except FileNotFoundError:
+        error = [' The command ', ('bold', command), ' could not be found']
+
+    if error:
+        controller.view.set_footer_text(error, duration=3)

zulipterminal/ui_tools/buttons.py

@@ -7,7 +7,7 @@
 
 from zulipterminal.config.keys import is_command_key, keys_for_command
 from zulipterminal.helper import (
-    StreamData, edit_mode_captions, hash_util_decode,
+    StreamData, edit_mode_captions, hash_util_decode, process_media,
 )
 from zulipterminal.urwid_types import urwid_Size
 
@@ -300,6 +300,11 @@ def handle_link(self, *_: Any) -> None:
         server_url = self.model.server_url
         if self.link.startswith(urljoin(server_url, '/#narrow/')):
             self.handle_narrow_link()
+        elif self.link.startswith(urljoin(server_url, '/user_uploads/')):
+            # Exit pop-up promptly, let the media download in the background.
+            if isinstance(self.controller.loop.widget, urwid.Overlay):
+                self.controller.exit_popup()
+            process_media(self.controller, self.link)
 
     @staticmethod
     def _decode_stream_data(encoded_stream_data: str) -> DecodedStream:

zulipterminal/ui_tools/views.py

@@ -1006,9 +1006,13 @@ def __init__(self, controller: Any, title: str) -> None:
         super().__init__(controller, widgets, 'HELP', popup_width, title)
 
 
+PopUpConfirmationViewLocation = Literal['top-left', 'center']
+
+
 class PopUpConfirmationView(urwid.Overlay):
     def __init__(self, controller: Any, question: Any,
-                 success_callback: Callable[[], None]):
+                 success_callback: Callable[[], None],
+                 location: PopUpConfirmationViewLocation='top-left'):
         self.controller = controller
         self.success_callback = success_callback
         yes = urwid.Button('Yes', self.exit_popup_yes)
@@ -1019,15 +1023,31 @@ def __init__(self, controller: Any, question: Any,
             'No', 4), None, 'selected')
         display_widget = urwid.GridFlow([yes, no], 3, 5, 1, 'center')
         wrapped_widget = urwid.WidgetWrap(display_widget)
+        widgets = [question, urwid.Divider(), wrapped_widget]
         prompt = urwid.LineBox(
             urwid.ListBox(
                 urwid.SimpleFocusListWalker(
-                    [question, urwid.Divider(), wrapped_widget]
+                    widgets
                 )))
-        urwid.Overlay.__init__(self, prompt, self.controller.view,
-                               align="left", valign="top",
-                               width=self.controller.view.LEFT_WIDTH + 1,
-                               height=8)
+
+        if location == 'top-left':
+            align = 'left'
+            valign = 'top'
+            width = self.controller.view.LEFT_WIDTH + 1
+            height = 8
+        else:
+            align = 'center'
+            valign = 'middle'
+
+            max_cols, max_rows = controller.maximum_popup_dimensions()
+            # +2 to compensate for the LineBox characters.
+            width = min(max_cols,
+                        max(question.pack()[0], len('Yes'), len('No'))) + 2
+            height = min(max_rows,
+                         sum(widget.rows((width,)) for widget in widgets)) + 2
+
+        urwid.Overlay.__init__(self, prompt, self.controller.view, align=align,
+                               valign=valign, width=width, height=height)
 
     def exit_popup_yes(self, args: Any) -> None:
         self.success_callback()