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

Commits: af207e3d by Philipp Hörist at 2022-12-09T22:04:37+01:00 More http stuff

- - - - -

3 changed files:

- nbxmpp/const.py - + nbxmpp/http.py - + test/unit/test_http.py

Changes:

===================================== nbxmpp/const.py ===================================== @@ -453,6 +453,15 @@ class Mode(IntEnum): return self == Mode.ANONYMOUS_TEST

+class HTTPRequestError(IntEnum): + UNKNOWN = 0 + MISSING_CONTENT_LENGTH = 1 + INCOMPLETE = 2 + STATUS = 3 + CANCELLED = 4 + CONTENT_OVERFLOW = 5 + + MOODS = [ 'afraid', 'amazed',

===================================== nbxmpp/http.py ===================================== @@ -0,0 +1,394 @@ +# This file is part of Gajim. +# +# Gajim is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published +# by the Free Software Foundation; version 3 only. +# +# Gajim is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Gajim. If not, see http://www.gnu.org/licenses/. + +# XEP-0363: HTTP File Upload + +from __future__ import annotations +from pathlib import Path + +from typing import Any +from typing import Literal +from typing import Callable +from typing import Optional + +import logging +import threading +from threading import Thread + +from gi.repository import Soup +from gi.repository import Gio +from gi.repository import GLib + +import nbxmpp +from nbxmpp.const import HTTPRequestError +from nbxmpp.util import Observable + + +log = logging.getLogger('nbxmpp.http') + + +HTTP_METHODS_T = Literal['GET', 'POST'] +CHUNK_SIZE = 32768 +DEFAULT_USER_AGENT = f'nbxmpp/{nbxmpp.__version__}' + + +class HTTPLogAdapter(logging.LoggerAdapter): + def process(self, msg: str, kwargs: Any) -> tuple[str, Any]: + return f'{self.extra["request"]}: {msg}', kwargs + + +class HTTPSessionManager: + def __init__(self) -> None: + pass + + +class HTTPSession: + def __init__(self, user_agent: str = DEFAULT_USER_AGENT) -> None: + self._session = Soup.Session() + self._session.set_user_agent(user_agent) + + self._requests: list[HTTPRequest] = [] + + def get_soup_session(self) -> Soup.Session: + return self._session + + def set_proxy_resolver(self, + resolver: Optional[Gio.SimpleProxyResolver] + ) -> None: + + self._session.set_proxy_resolver(resolver) + + def create_request(self) -> HTTPRequest: + request = HTTPRequest(self) + request.subscribe('destroyed', self._on_request_destroyed) + self._requests.append(request) + return request + + def _on_request_destroyed(self, + request: HTTPRequest, + _signal_name: str + ) -> None: + + self._requests.remove(request) + + +class HTTPRequest(Observable): + + ''' + Signals + - started + - complete + - failed + - cancelled + - destroyed + ''' + + def __init__(self, session: HTTPSession) -> None: + Observable.__init__(self, HTTPLogAdapter(log, extra={'request': self})) + self._session = session + + self._data = b'' + self._received_size = 0 + self._uri: Optional[GLib.Uri] = None + + self._cancellable = Gio.Cancellable() + self._input_stream: Optional[Gio.InputStream] = None + self._output_stream: Optional[Gio.OutputStream] = None + self._file: Optional[Path] = None + self._is_finished = False + + self._content_length = 0 + self._content_type = '' + + self._message = None + self._progress_callback: Optional[Callable[..., None]] = None + + self._lock = threading.Lock() + self._log.info('Created') + + def get_content_length(self) -> int: + return self._content_length + + def get_content_type(self) -> str: + return self._content_type + + def get_data(self) -> bytes: + return self._data + + def get_uri(self) -> Optional[GLib.Uri]: + return self._uri + + def get_file_path(self) -> Optional[Gio.File]: + return self._file + + def get_status(self) -> Soup.Status: + return self._message.get_status() + + def get_progress(self) -> float: + if self._received_size == 0: + return 0 + return self._received_size / self._content_length + + def cancel(self) -> None: + if self._is_finished: + raise ValueError('Session already finished') + + self._cancellable.cancel() + + def set_progress_callback(self, func: Any) -> None: + self._progress_callback = func + + def send(self, + method: HTTP_METHODS_T, + uri_string: str, + file_path: Optional[Path] = None + ) -> None: + + if self._is_finished: + raise ValueError('Session already finished') + + if file_path is not None: + self._file = Gio.File.new_for_path(str(file_path)) + self._output_stream = self._file.replace(None, + False, + Gio.FileCreateFlags.REPLACE_DESTINATION, + self._cancellable) + + self._uri = GLib.Uri.parse(uri_string, GLib.UriFlags.HAS_PASSWORD) + self._message = Soup.Message.new_from_uri(method, self._uri) + self._message.connect('finished', self._on_finished) + + soup_session = self._session.get_soup_session() + soup_session.send_async(self._message, + GLib.PRIORITY_DEFAULT, + self._cancellable, + self._on_response) + + self._log.info('Request sent') + + def _on_response(self, + session: Soup.Session, + result: Gio.AsyncResult + ) -> None: + + self._log.info('Request response received') + try: + self._input_stream = session.send_finish(result) + except Exception as error: + self._log.error(error) + quark = GLib.quark_try_string('g-io-error-quark') + if error.matches(quark, Gio.IOErrorEnum.CANCELLED): + self._set_cancelled() + else: + self._set_failed(HTTPRequestError.UNKNOWN) + return + + response_headers = self._message.get_response_headers() + self._content_length = response_headers.get_content_length() + self._content_type = response_headers.get_content_type() + + if self._message.get_status() != Soup.Status.OK: + self._set_failed(HTTPRequestError.STATUS) + return + + if self._content_length == 0: + self._set_failed(HTTPRequestError.MISSING_CONTENT_LENGTH) + return + + self.notify('started') + + if self._file is not None: + thread = Thread(target=self._thread_write_to_file, + args=(self._content_length,)) + thread.start() + + else: + self._read_bytes_async() + + def _thread_write_to_file(self, content_length: int) -> None: + received_size = 0 + + while True: + try: + data = self._input_stream.read_bytes(CHUNK_SIZE, + self._cancellable) + except GLib.Error as error: + self._log.error(error) + quark = GLib.quark_try_string('g-io-error-quark') + if error.matches(quark, Gio.IOErrorEnum.CANCELLED): + result = HTTPRequestError.CANCELLED + else: + result = HTTPRequestError.UNKNOWN + + GLib.idle_add(self._set_thread_result, result) + return + + bytes_ = data.get_data() + if not bytes_: + GLib.idle_add(self._set_thread_result) + return + + length = len(bytes_) + received_size += length + if received_size > content_length: + GLib.idle_add(self._set_thread_result, + HTTPRequestError.CONTENT_OVERFLOW) + return + + try: + self._output_stream.write_all(bytes_, self._cancellable) + except Exception as error: + self._log.error(error) + quark = GLib.quark_try_string('g-io-error-quark') + if error.matches(quark, Gio.IOErrorEnum.CANCELLED): + result = HTTPRequestError.CANCELLED + else: + result = HTTPRequestError.UNKNOWN + + GLib.idle_add(self._set_thread_result, result) + return + + GLib.idle_add(self._set_thread_progress, received_size) + + def _set_thread_progress(self, received_size: int): + self._received_size = received_size + if self._progress_callback is not None: + self._progress_callback(self, self.get_progress()) + + def _set_thread_result(self, + error: Optional[HTTPRequestError] = None + ) -> None: + + if error is None: + self._close_all_streams() + return + + if error == HTTPRequestError.CANCELLED: + self._set_cancelled() + + else: + self._set_failed(error) + + def _read_bytes_async(self) -> None: + self._input_stream.read_bytes_async(CHUNK_SIZE, + GLib.PRIORITY_LOW, + self._cancellable, + self._on_read_bytes_async_result) + + def _on_read_bytes_async_result(self, + input_stream: Gio.InputStream, + result: Gio.AsyncResult + ) -> None: + + try: + chunk = input_stream.read_bytes_finish(result) + except GLib.Error as error: + self._log.error(error) + quark = GLib.quark_try_string('g-io-error-quark') + if error.matches(quark, Gio.IOErrorEnum.CANCELLED): + self._set_cancelled() + else: + self._set_failed(HTTPRequestError.UNKNOWN) + return + + data = chunk.get_data() + size = len(data) + if size == 0: + self._close_all_streams() + return + + self._received_size += size + self._data += data + + if self._received_size > self._content_length: + self._log.warning('More data received than the' + 'advertised content length') + self._set_failed(HTTPRequestError.CONTENT_OVERFLOW) + return + + if self._progress_callback is not None: + self._progress_callback(self, self.get_progress()) + + self._read_bytes_async() + + def _on_finished(self, message: Soup.Message) -> None: + self._set_finished() + + def _set_finished(self) -> None: + status = self._message.get_status() + if status == Soup.Status.NONE: + # Message has not been sent, can happen when we cancel the message + # before it is sent. The finished signal triggers before the + # cancelled exception. + return + + if self._is_finished: + return + + if self._received_size != self._content_length: + self._set_failed(HTTPRequestError.INCOMPLETE) + return + + if status != Soup.Status.OK: + self._set_failed(HTTPRequestError.STATUS) + return + + self._set_complete() + + def _set_failed(self, error: HTTPRequestError) -> None: + self._close_all_streams() + self.notify('failed', error) + self._cleanup() + + def _set_cancelled(self) -> None: + self._close_all_streams() + self.notify('cancelled') + self._cleanup() + + def _set_complete(self) -> None: + self._close_all_streams() + self.notify('complete') + self._cleanup() + + def _close_all_streams(self) -> None: + if self._input_stream is not None: + if not self._input_stream.is_closed(): + self._input_stream.close(None) + + if self._output_stream is not None: + if not self._output_stream.is_closed(): + self._output_stream.close(None) + + def _cleanup(self) -> None: + if self._is_finished: + raise ValueError('Request already finished') + + self._log.info('Run cleanup') + self._is_finished = True + + self._data = b'' + self._message.run_dispose() + + self._cancellable = None + self._session = None + self._progress_callback = None + self._input_stream = None + self._output_stream = None + self._file = None + + self.notify('destroyed') + self.remove_subscriptions() + + def __repr__(self) -> str: + return f'Request({id(self)})'

===================================== test/unit/test_http.py ===================================== @@ -0,0 +1,153 @@ +import unittest +from unittest.mock import call +from unittest.mock import Mock +from unittest.mock import patch +import os + +from gi.repository import GLib +from gi.repository import Soup + +from nbxmpp.const import HTTPRequestError +from nbxmpp.http import HTTPSession + + +TEST_DOWNLOAD_FILE = 'https://gajim.org/downloads/snap/win/build/pybonjour-1.1.1.zip' +FILE_DOES_NOT_EXIST = 'https://gajim.org/downloads/1.4/no-file.tar.gz' + + +import logging +consoleloghandler = logging.StreamHandler() +log = logging.getLogger('nbxmpp.http') +log.setLevel(logging.DEBUG) +log.addHandler(consoleloghandler) + + +@unittest.skipUnless(os.environ.get('NBXMPP_EXTERNAL_UNIT_TESTS'), + 'ENV var for external tests not set') +class HTTP(unittest.TestCase): + + @classmethod + def setUpClass(cls): + pass + + @classmethod + def tearDownClass(cls): + pass + + def test_in_memory_download_to_file(self): + + mainloop = GLib.MainLoop() + + session = HTTPSession() + request = session.create_request() + + with patch.object(request, 'notify', wraps=request.notify) as n_mock: + + progress_func = Mock() + request.set_progress_callback(progress_func) + request.subscribe('destroyed', lambda *args: mainloop.quit()) + request.send('GET', TEST_DOWNLOAD_FILE, '/home/lovetox/Downloads/test123') + + GLib.timeout_add_seconds(10, mainloop.quit) + mainloop.run() + + progress_func.assert_called() + n_mock.assert_has_calls([call('started'), + call('complete'), + call('destroyed')]) + + def test_in_memory_download(self): + + mainloop = GLib.MainLoop() + + session = HTTPSession() + request = session.create_request() + + with patch.object(request, 'notify', wraps=request.notify) as n_mock: + + progress_func = Mock() + request.set_progress_callback(progress_func) + request.subscribe('destroyed', lambda *args: mainloop.quit()) + request.send('GET', TEST_DOWNLOAD_FILE) + + GLib.timeout_add_seconds(10, mainloop.quit) + mainloop.run() + + progress_func.assert_called() + n_mock.assert_has_calls([call('started'), + call('complete'), + call('destroyed')]) + + def test_download_cancelled_before_start(self): + + mainloop = GLib.MainLoop() + + session = HTTPSession() + request = session.create_request() + + with patch.object(request, 'notify', wraps=request.notify) as n_mock: + + progress_func = Mock() + request.set_progress_callback(progress_func) + request.subscribe('destroyed', lambda *args: mainloop.quit()) + request.send('GET', TEST_DOWNLOAD_FILE) + + GLib.timeout_add(10, request.cancel) + GLib.timeout_add_seconds(2, mainloop.quit) + + mainloop.run() + + progress_func.assert_not_called() + n_mock.assert_has_calls([call('cancelled'), + call('destroyed')]) + + def test_download_cancelled_after_start(self): + + mainloop = GLib.MainLoop() + + session = HTTPSession() + request = session.create_request() + + def _on_progress(req, progress): + req.cancel() + + with patch.object(request, 'notify', wraps=request.notify) as n_mock: + + request.set_progress_callback(_on_progress) + request.subscribe('destroyed', lambda *args: mainloop.quit()) + request.send('GET', TEST_DOWNLOAD_FILE) + + GLib.timeout_add_seconds(10, mainloop.quit) + + mainloop.run() + + n_mock.assert_has_calls([call('started'), + call('cancelled'), + call('destroyed')]) + + def test_download_failed_404(self): + + mainloop = GLib.MainLoop() + + session = HTTPSession() + request = session.create_request() + + with patch.object(request, 'notify', wraps=request.notify) as n_mock: + + progress_func = Mock() + request.set_progress_callback(progress_func) + request.subscribe('destroyed', lambda *args: mainloop.quit()) + request.send('GET', FILE_DOES_NOT_EXIST) + + GLib.timeout_add_seconds(5, mainloop.quit) + + mainloop.run() + + progress_func.assert_not_called() + n_mock.assert_has_calls([call('failed', HTTPRequestError.STATUS), + call('destroyed')]) + self.assertEqual(request.get_status(), Soup.Status.NOT_FOUND) + + +if __name__ == '__main__': + unittest.main()

View it on GitLab: https://dev.gajim.org/gajim/python-nbxmpp/-/commit/af207e3deb812e666d1b88b2b...