Philipp Hörist pushed to branch soup3 at gajim / python-nbxmpp

Commits:

3 changed files:

Changes:

  • nbxmpp/const.py
    ... ... @@ -453,6 +453,15 @@ class Mode(IntEnum):
    453 453 
             return self == Mode.ANONYMOUS_TEST
    454 454
    455 455
    456 
    +class HTTPRequestError(IntEnum):
    457 
    +    UNKNOWN = 0
    458 
    +    MISSING_CONTENT_LENGTH = 1
    459 
    +    INCOMPLETE = 2
    460 
    +    STATUS = 3
    461 
    +    CANCELLED = 4
    462 
    +    CONTENT_OVERFLOW = 5
    463 
    +
    464 
    +
    456 465 
     MOODS = [
    457 466 
         'afraid',
    458 467 
         'amazed',

  • nbxmpp/http.py
    1 
    +# This file is part of Gajim.
    2 
    +#
    3 
    +# Gajim is free software; you can redistribute it and/or modify
    4 
    +# it under the terms of the GNU General Public License as published
    5 
    +# by the Free Software Foundation; version 3 only.
    6 
    +#
    7 
    +# Gajim is distributed in the hope that it will be useful,
    8 
    +# but WITHOUT ANY WARRANTY; without even the implied warranty of
    9 
    +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    10 
    +# GNU General Public License for more details.
    11 
    +#
    12 
    +# You should have received a copy of the GNU General Public License
    13 
    +# along with Gajim.  If not, see <http://www.gnu.org/licenses/>.
    14 
    +
    15 
    +# XEP-0363: HTTP File Upload
    16 
    +
    17 
    +from __future__ import annotations
    18 
    +from pathlib import Path
    19 
    +
    20 
    +from typing import Any
    21 
    +from typing import Literal
    22 
    +from typing import Callable
    23 
    +from typing import Optional
    24 
    +
    25 
    +import logging
    26 
    +import threading
    27 
    +from threading import Thread
    28 
    +
    29 
    +from gi.repository import Soup
    30 
    +from gi.repository import Gio
    31 
    +from gi.repository import GLib
    32 
    +
    33 
    +import nbxmpp
    34 
    +from nbxmpp.const import HTTPRequestError
    35 
    +from nbxmpp.util import Observable
    36 
    +
    37 
    +
    38 
    +log = logging.getLogger('nbxmpp.http')
    39 
    +
    40 
    +
    41 
    +HTTP_METHODS_T = Literal['GET', 'POST']
    42 
    +CHUNK_SIZE = 32768
    43 
    +DEFAULT_USER_AGENT = f'nbxmpp/{nbxmpp.__version__}'
    44 
    +
    45 
    +
    46 
    +class HTTPLogAdapter(logging.LoggerAdapter):
    47 
    +    def process(self, msg: str, kwargs: Any) -> tuple[str, Any]:
    48 
    +        return f'{self.extra["request"]}: {msg}', kwargs
    49 
    +
    50 
    +
    51 
    +class HTTPSessionManager:
    52 
    +    def __init__(self) -> None:
    53 
    +        pass
    54 
    +
    55 
    +
    56 
    +class HTTPSession:
    57 
    +    def __init__(self, user_agent: str = DEFAULT_USER_AGENT) -> None:
    58 
    +        self._session = Soup.Session()
    59 
    +        self._session.set_user_agent(user_agent)
    60 
    +
    61 
    +        self._requests: list[HTTPRequest] = []
    62 
    +
    63 
    +    def get_soup_session(self) -> Soup.Session:
    64 
    +        return self._session
    65 
    +
    66 
    +    def set_proxy_resolver(self,
    67 
    +                           resolver: Optional[Gio.SimpleProxyResolver]
    68 
    +                           ) -> None:
    69 
    +
    70 
    +        self._session.set_proxy_resolver(resolver)
    71 
    +
    72 
    +    def create_request(self) -> HTTPRequest:
    73 
    +        request = HTTPRequest(self)
    74 
    +        request.subscribe('destroyed', self._on_request_destroyed)
    75 
    +        self._requests.append(request)
    76 
    +        return request
    77 
    +
    78 
    +    def _on_request_destroyed(self,
    79 
    +                              request: HTTPRequest,
    80 
    +                              _signal_name: str
    81 
    +                              ) -> None:
    82 
    +
    83 
    +        self._requests.remove(request)
    84 
    +
    85 
    +
    86 
    +class HTTPRequest(Observable):
    87 
    +
    88 
    +    '''
    89 
    +    Signals
    90 
    +        - started
    91 
    +        - complete
    92 
    +        - failed
    93 
    +        - cancelled
    94 
    +        - destroyed
    95 
    +    '''
    96 
    +
    97 
    +    def __init__(self, session: HTTPSession) -> None:
    98 
    +        Observable.__init__(self, HTTPLogAdapter(log, extra={'request': self}))
    99 
    +        self._session = session
    100 
    +
    101 
    +        self._data = b''
    102 
    +        self._received_size = 0
    103 
    +        self._uri: Optional[GLib.Uri] = None
    104 
    +
    105 
    +        self._cancellable = Gio.Cancellable()
    106 
    +        self._input_stream: Optional[Gio.InputStream] = None
    107 
    +        self._output_stream: Optional[Gio.OutputStream] = None
    108 
    +        self._file: Optional[Path] = None
    109 
    +        self._is_finished = False
    110 
    +
    111 
    +        self._content_length = 0
    112 
    +        self._content_type = ''
    113 
    +
    114 
    +        self._message = None
    115 
    +        self._progress_callback: Optional[Callable[..., None]] = None
    116 
    +
    117 
    +        self._lock = threading.Lock()
    118 
    +        self._log.info('Created')
    119 
    +
    120 
    +    def get_content_length(self) -> int:
    121 
    +        return self._content_length
    122 
    +
    123 
    +    def get_content_type(self) -> str:
    124 
    +        return self._content_type
    125 
    +
    126 
    +    def get_data(self) -> bytes:
    127 
    +        return self._data
    128 
    +
    129 
    +    def get_uri(self) -> Optional[GLib.Uri]:
    130 
    +        return self._uri
    131 
    +
    132 
    +    def get_file_path(self) -> Optional[Gio.File]:
    133 
    +        return self._file
    134 
    +
    135 
    +    def get_status(self) -> Soup.Status:
    136 
    +        return self._message.get_status()
    137 
    +
    138 
    +    def get_progress(self) -> float:
    139 
    +        if self._received_size == 0:
    140 
    +            return 0
    141 
    +        return self._received_size / self._content_length
    142 
    +
    143 
    +    def cancel(self) -> None:
    144 
    +        if self._is_finished:
    145 
    +            raise ValueError('Session already finished')
    146 
    +
    147 
    +        self._cancellable.cancel()
    148 
    +
    149 
    +    def set_progress_callback(self, func: Any) -> None:
    150 
    +        self._progress_callback = func
    151 
    +
    152 
    +    def send(self,
    153 
    +             method: HTTP_METHODS_T,
    154 
    +             uri_string: str,
    155 
    +             file_path: Optional[Path] = None
    156 
    +             ) -> None:
    157 
    +
    158 
    +        if self._is_finished:
    159 
    +            raise ValueError('Session already finished')
    160 
    +
    161 
    +        if file_path is not None:
    162 
    +            self._file = Gio.File.new_for_path(str(file_path))
    163 
    +            self._output_stream = self._file.replace(None,
    164 
    +                                                     False,
    165 
    +                                                     Gio.FileCreateFlags.REPLACE_DESTINATION,
    166 
    +                                                     self._cancellable)
    167 
    +
    168 
    +        self._uri = GLib.Uri.parse(uri_string, GLib.UriFlags.HAS_PASSWORD)
    169 
    +        self._message = Soup.Message.new_from_uri(method, self._uri)
    170 
    +        self._message.connect('finished', self._on_finished)
    171 
    +
    172 
    +        soup_session = self._session.get_soup_session()
    173 
    +        soup_session.send_async(self._message,
    174 
    +                                GLib.PRIORITY_DEFAULT,
    175 
    +                                self._cancellable,
    176 
    +                                self._on_response)
    177 
    +
    178 
    +        self._log.info('Request sent')
    179 
    +
    180 
    +    def _on_response(self,
    181 
    +                     session: Soup.Session,
    182 
    +                     result: Gio.AsyncResult
    183 
    +                     ) -> None:
    184 
    +
    185 
    +        self._log.info('Request response received')
    186 
    +        try:
    187 
    +            self._input_stream = session.send_finish(result)
    188 
    +        except Exception as error:
    189 
    +            self._log.error(error)
    190 
    +            quark = GLib.quark_try_string('g-io-error-quark')
    191 
    +            if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
    192 
    +                self._set_cancelled()
    193 
    +            else:
    194 
    +                self._set_failed(HTTPRequestError.UNKNOWN)
    195 
    +            return
    196 
    +
    197 
    +        response_headers = self._message.get_response_headers()
    198 
    +        self._content_length = response_headers.get_content_length()
    199 
    +        self._content_type = response_headers.get_content_type()
    200 
    +
    201 
    +        if self._message.get_status() != Soup.Status.OK:
    202 
    +            self._set_failed(HTTPRequestError.STATUS)
    203 
    +            return
    204 
    +
    205 
    +        if self._content_length == 0:
    206 
    +            self._set_failed(HTTPRequestError.MISSING_CONTENT_LENGTH)
    207 
    +            return
    208 
    +
    209 
    +        self.notify('started')
    210 
    +
    211 
    +        if self._file is not None:
    212 
    +            thread = Thread(target=self._thread_write_to_file,
    213 
    +                            args=(self._content_length,))
    214 
    +            thread.start()
    215 
    +
    216 
    +        else:
    217 
    +            self._read_bytes_async()
    218 
    +
    219 
    +    def _thread_write_to_file(self, content_length: int) -> None:
    220 
    +        received_size = 0
    221 
    +
    222 
    +        while True:
    223 
    +            try:
    224 
    +                data = self._input_stream.read_bytes(CHUNK_SIZE,
    225 
    +                                                     self._cancellable)
    226 
    +            except GLib.Error as error:
    227 
    +                self._log.error(error)
    228 
    +                quark = GLib.quark_try_string('g-io-error-quark')
    229 
    +                if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
    230 
    +                    result = HTTPRequestError.CANCELLED
    231 
    +                else:
    232 
    +                    result = HTTPRequestError.UNKNOWN
    233 
    +
    234 
    +                GLib.idle_add(self._set_thread_result, result)
    235 
    +                return
    236 
    +
    237 
    +            bytes_ = data.get_data()
    238 
    +            if not bytes_:
    239 
    +                GLib.idle_add(self._set_thread_result)
    240 
    +                return
    241 
    +
    242 
    +            length = len(bytes_)
    243 
    +            received_size += length
    244 
    +            if received_size > content_length:
    245 
    +                GLib.idle_add(self._set_thread_result,
    246 
    +                              HTTPRequestError.CONTENT_OVERFLOW)
    247 
    +                return
    248 
    +
    249 
    +            try:
    250 
    +                self._output_stream.write_all(bytes_, self._cancellable)
    251 
    +            except Exception as error:
    252 
    +                self._log.error(error)
    253 
    +                quark = GLib.quark_try_string('g-io-error-quark')
    254 
    +                if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
    255 
    +                    result = HTTPRequestError.CANCELLED
    256 
    +                else:
    257 
    +                    result = HTTPRequestError.UNKNOWN
    258 
    +
    259 
    +                GLib.idle_add(self._set_thread_result, result)
    260 
    +                return
    261 
    +
    262 
    +            GLib.idle_add(self._set_thread_progress, received_size)
    263 
    +
    264 
    +    def _set_thread_progress(self, received_size: int):
    265 
    +        self._received_size = received_size
    266 
    +        if self._progress_callback is not None:
    267 
    +            self._progress_callback(self, self.get_progress())
    268 
    +
    269 
    +    def _set_thread_result(self,
    270 
    +                           error: Optional[HTTPRequestError] = None
    271 
    +                           ) -> None:
    272 
    +
    273 
    +        if error is None:
    274 
    +            self._close_all_streams()
    275 
    +            return
    276 
    +
    277 
    +        if error == HTTPRequestError.CANCELLED:
    278 
    +            self._set_cancelled()
    279 
    +
    280 
    +        else:
    281 
    +            self._set_failed(error)
    282 
    +
    283 
    +    def _read_bytes_async(self) -> None:
    284 
    +        self._input_stream.read_bytes_async(CHUNK_SIZE,
    285 
    +                                            GLib.PRIORITY_LOW,
    286 
    +                                            self._cancellable,
    287 
    +                                            self._on_read_bytes_async_result)
    288 
    +
    289 
    +    def _on_read_bytes_async_result(self,
    290 
    +                                    input_stream: Gio.InputStream,
    291 
    +                                    result: Gio.AsyncResult
    292 
    +                                    ) -> None:
    293 
    +
    294 
    +        try:
    295 
    +            chunk = input_stream.read_bytes_finish(result)
    296 
    +        except GLib.Error as error:
    297 
    +            self._log.error(error)
    298 
    +            quark = GLib.quark_try_string('g-io-error-quark')
    299 
    +            if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
    300 
    +                self._set_cancelled()
    301 
    +            else:
    302 
    +                self._set_failed(HTTPRequestError.UNKNOWN)
    303 
    +            return
    304 
    +
    305 
    +        data = chunk.get_data()
    306 
    +        size = len(data)
    307 
    +        if size == 0:
    308 
    +            self._close_all_streams()
    309 
    +            return
    310 
    +
    311 
    +        self._received_size += size
    312 
    +        self._data += data
    313 
    +
    314 
    +        if self._received_size > self._content_length:
    315 
    +            self._log.warning('More data received than the'
    316 
    +                              'advertised content length')
    317 
    +            self._set_failed(HTTPRequestError.CONTENT_OVERFLOW)
    318 
    +            return
    319 
    +
    320 
    +        if self._progress_callback is not None:
    321 
    +            self._progress_callback(self, self.get_progress())
    322 
    +
    323 
    +        self._read_bytes_async()
    324 
    +
    325 
    +    def _on_finished(self, message: Soup.Message) -> None:
    326 
    +        self._set_finished()
    327 
    +
    328 
    +    def _set_finished(self) -> None:
    329 
    +        status = self._message.get_status()
    330 
    +        if status == Soup.Status.NONE:
    331 
    +            # Message has not been sent, can happen when we cancel the message
    332 
    +            # before it is sent. The finished signal triggers before the
    333 
    +            # cancelled exception.
    334 
    +            return
    335 
    +
    336 
    +        if self._is_finished:
    337 
    +            return
    338 
    +
    339 
    +        if self._received_size != self._content_length:
    340 
    +            self._set_failed(HTTPRequestError.INCOMPLETE)
    341 
    +            return
    342 
    +
    343 
    +        if status != Soup.Status.OK:
    344 
    +            self._set_failed(HTTPRequestError.STATUS)
    345 
    +            return
    346 
    +
    347 
    +        self._set_complete()
    348 
    +
    349 
    +    def _set_failed(self, error: HTTPRequestError) -> None:
    350 
    +        self._close_all_streams()
    351 
    +        self.notify('failed', error)
    352 
    +        self._cleanup()
    353 
    +
    354 
    +    def _set_cancelled(self) -> None:
    355 
    +        self._close_all_streams()
    356 
    +        self.notify('cancelled')
    357 
    +        self._cleanup()
    358 
    +
    359 
    +    def _set_complete(self) -> None:
    360 
    +        self._close_all_streams()
    361 
    +        self.notify('complete')
    362 
    +        self._cleanup()
    363 
    +
    364 
    +    def _close_all_streams(self) -> None:
    365 
    +        if self._input_stream is not None:
    366 
    +            if not self._input_stream.is_closed():
    367 
    +                self._input_stream.close(None)
    368 
    +
    369 
    +        if self._output_stream is not None:
    370 
    +            if not self._output_stream.is_closed():
    371 
    +                self._output_stream.close(None)
    372 
    +
    373 
    +    def _cleanup(self) -> None:
    374 
    +        if self._is_finished:
    375 
    +            raise ValueError('Request already finished')
    376 
    +
    377 
    +        self._log.info('Run cleanup')
    378 
    +        self._is_finished = True
    379 
    +
    380 
    +        self._data = b''
    381 
    +        self._message.run_dispose()
    382 
    +
    383 
    +        self._cancellable = None
    384 
    +        self._session = None
    385 
    +        self._progress_callback = None
    386 
    +        self._input_stream = None
    387 
    +        self._output_stream = None
    388 
    +        self._file = None
    389 
    +
    390 
    +        self.notify('destroyed')
    391 
    +        self.remove_subscriptions()
    392 
    +
    393 
    +    def __repr__(self) -> str:
    394 
    +        return f'Request({id(self)})'

  • test/unit/test_http.py
    1 
    +import unittest
    2 
    +from unittest.mock import call
    3 
    +from unittest.mock import Mock
    4 
    +from unittest.mock import patch
    5 
    +import os
    6 
    +
    7 
    +from gi.repository import GLib
    8 
    +from gi.repository import Soup
    9 
    +
    10 
    +from nbxmpp.const import HTTPRequestError
    11 
    +from nbxmpp.http import HTTPSession
    12 
    +
    13 
    +
    14 
    +TEST_DOWNLOAD_FILE = 'https://gajim.org/downloads/snap/win/build/pybonjour-1.1.1.zip'
    15 
    +FILE_DOES_NOT_EXIST = 'https://gajim.org/downloads/1.4/no-file.tar.gz'
    16 
    +
    17 
    +
    18 
    +import logging
    19 
    +consoleloghandler = logging.StreamHandler()
    20 
    +log = logging.getLogger('nbxmpp.http')
    21 
    +log.setLevel(logging.DEBUG)
    22 
    +log.addHandler(consoleloghandler)
    23 
    +
    24 
    +
    25 
    +@unittest.skipUnless(os.environ.get('NBXMPP_EXTERNAL_UNIT_TESTS'),
    26 
    +                     'ENV var for external tests not set')
    27 
    +class HTTP(unittest.TestCase):
    28 
    +
    29 
    +    @classmethod
    30 
    +    def setUpClass(cls):
    31 
    +        pass
    32 
    +
    33 
    +    @classmethod
    34 
    +    def tearDownClass(cls):
    35 
    +        pass
    36 
    +
    37 
    +    def test_in_memory_download_to_file(self):
    38 
    +
    39 
    +        mainloop = GLib.MainLoop()
    40 
    +
    41 
    +        session = HTTPSession()
    42 
    +        request = session.create_request()
    43 
    +
    44 
    +        with patch.object(request, 'notify', wraps=request.notify) as n_mock:
    45 
    +
    46 
    +            progress_func = Mock()
    47 
    +            request.set_progress_callback(progress_func)
    48 
    +            request.subscribe('destroyed', lambda *args: mainloop.quit())
    49 
    +            request.send('GET', TEST_DOWNLOAD_FILE, '/home/lovetox/Downloads/test123')
    50 
    +
    51 
    +            GLib.timeout_add_seconds(10, mainloop.quit)
    52 
    +            mainloop.run()
    53 
    +
    54 
    +            progress_func.assert_called()
    55 
    +            n_mock.assert_has_calls([call('started'),
    56 
    +                                     call('complete'),
    57 
    +                                     call('destroyed')])
    58 
    +
    59 
    +    def test_in_memory_download(self):
    60 
    +
    61 
    +        mainloop = GLib.MainLoop()
    62 
    +
    63 
    +        session = HTTPSession()
    64 
    +        request = session.create_request()
    65 
    +
    66 
    +        with patch.object(request, 'notify', wraps=request.notify) as n_mock:
    67 
    +
    68 
    +            progress_func = Mock()
    69 
    +            request.set_progress_callback(progress_func)
    70 
    +            request.subscribe('destroyed', lambda *args: mainloop.quit())
    71 
    +            request.send('GET', TEST_DOWNLOAD_FILE)
    72 
    +
    73 
    +            GLib.timeout_add_seconds(10, mainloop.quit)
    74 
    +            mainloop.run()
    75 
    +
    76 
    +            progress_func.assert_called()
    77 
    +            n_mock.assert_has_calls([call('started'),
    78 
    +                                     call('complete'),
    79 
    +                                     call('destroyed')])
    80 
    +
    81 
    +    def test_download_cancelled_before_start(self):
    82 
    +
    83 
    +        mainloop = GLib.MainLoop()
    84 
    +
    85 
    +        session = HTTPSession()
    86 
    +        request = session.create_request()
    87 
    +
    88 
    +        with patch.object(request, 'notify', wraps=request.notify) as n_mock:
    89 
    +
    90 
    +            progress_func = Mock()
    91 
    +            request.set_progress_callback(progress_func)
    92 
    +            request.subscribe('destroyed', lambda *args: mainloop.quit())
    93 
    +            request.send('GET', TEST_DOWNLOAD_FILE)
    94 
    +
    95 
    +            GLib.timeout_add(10, request.cancel)
    96 
    +            GLib.timeout_add_seconds(2, mainloop.quit)
    97 
    +
    98 
    +            mainloop.run()
    99 
    +
    100 
    +            progress_func.assert_not_called()
    101 
    +            n_mock.assert_has_calls([call('cancelled'),
    102 
    +                                     call('destroyed')])
    103 
    +
    104 
    +    def test_download_cancelled_after_start(self):
    105 
    +
    106 
    +        mainloop = GLib.MainLoop()
    107 
    +
    108 
    +        session = HTTPSession()
    109 
    +        request = session.create_request()
    110 
    +
    111 
    +        def _on_progress(req, progress):
    112 
    +            req.cancel()
    113 
    +
    114 
    +        with patch.object(request, 'notify', wraps=request.notify) as n_mock:
    115 
    +
    116 
    +            request.set_progress_callback(_on_progress)
    117 
    +            request.subscribe('destroyed', lambda *args: mainloop.quit())
    118 
    +            request.send('GET', TEST_DOWNLOAD_FILE)
    119 
    +
    120 
    +            GLib.timeout_add_seconds(10, mainloop.quit)
    121 
    +
    122 
    +            mainloop.run()
    123 
    +
    124 
    +            n_mock.assert_has_calls([call('started'),
    125 
    +                                     call('cancelled'),
    126 
    +                                     call('destroyed')])
    127 
    +
    128 
    +    def test_download_failed_404(self):
    129 
    +
    130 
    +        mainloop = GLib.MainLoop()
    131 
    +
    132 
    +        session = HTTPSession()
    133 
    +        request = session.create_request()
    134 
    +
    135 
    +        with patch.object(request, 'notify', wraps=request.notify) as n_mock:
    136 
    +
    137 
    +            progress_func = Mock()
    138 
    +            request.set_progress_callback(progress_func)
    139 
    +            request.subscribe('destroyed', lambda *args: mainloop.quit())
    140 
    +            request.send('GET', FILE_DOES_NOT_EXIST)
    141 
    +
    142 
    +            GLib.timeout_add_seconds(5, mainloop.quit)
    143 
    +
    144 
    +            mainloop.run()
    145 
    +
    146 
    +            progress_func.assert_not_called()
    147 
    +            n_mock.assert_has_calls([call('failed', HTTPRequestError.STATUS),
    148 
    +                                     call('destroyed')])
    149 
    +            self.assertEqual(request.get_status(), Soup.Status.NOT_FOUND)
    150 
    +
    151 
    +
    152 
    +if __name__ == '__main__':
    153 
    +    unittest.main()

    • —
    View it on GitLab.
    You're receiving this email because of your account on dev.gajim.org. Manage all notifications · Help