mcvsoftware/updater
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove dependency on requests

Browse Source
This commit is contained in:
Manuel Cortez 2022-02-23 11:09:51 -06:00
parent d7f7528a1d
commit 01b81bd8d7
No known key found for this signature in database
GPG Key ID: 9E0735CA15EFE790
6 changed files with 65 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitlab-ci.yml
View File

@ -9,7 +9,7 @@ test:mipy:
image: python:3.7
interruptible: true
script:
- 'pip install --upgrade mypy pytest types-requests'
- 'pip install --upgrade mypy pytest'
- 'pip install --upgrade -r requirements.txt'
- 'mypy updater'
only:
@ -25,7 +25,7 @@ test:pytest:
image: python:3.8
interruptible: true
script:
- 'pip install --upgrade pytest coverage requests_mock'
- 'pip install --upgrade pytest coverage'
- 'pip install --upgrade -r requirements.txt'
- 'coverage run --source . --omit conf.py -m pytest --junitxml=report.xml'
- 'coverage report'

3
requirements.txt
View File

@ -1,2 +1 @@
pypubsub
requests
pypubsub

51
test/test_core.py
View File

@ -1,10 +1,9 @@
import sys
import os
import pytest
import requests_mock
from unittest import mock
from json.decoder import JSONDecodeError
from requests.exceptions import HTTPError
from urllib.error import HTTPError
from updater import core
app_name: str = "a simple app"
@ -17,45 +16,36 @@ win32api = mock.Mock(name="win32api")
win32api.__name__ = "win32api"
sys.modules["win32api"] = win32api
def test_requests_session():
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
assert hasattr(updater, "session")
assert app_name in updater.session.headers.get("User-Agent")
assert current_version in updater.session.headers.get("User-Agent")
def test_get_update_information_valid_json(file_data, json_data):
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
with requests_mock.Mocker(session=updater.session) as mocked:
mocked.get(endpoint, json=json_data)
response = mock.Mock()
response.getcode.return_value = 200
response.read.return_value = file_data
with mock.patch("urllib.request.urlopen", return_value=response):
contents = updater.get_update_information()
assert contents == json_data
def test_get_update_information_invalid_json(file_data, json_data):
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
with requests_mock.Mocker(session=updater.session) as mocked:
mocked.get(endpoint, text="thisisnotjson")
response = mock.Mock()
response.getcode.return_value = 200
response.read.return_value = "invalid json"
with mock.patch("urllib.request.urlopen", return_value=response):
with pytest.raises(JSONDecodeError):
contents = updater.get_update_information()
def test_get_update_information_not_found():
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
with requests_mock.Mocker(session=updater.session) as mocked:
mocked.get(endpoint, status_code=404)
with mock.patch("urllib.request.urlopen", side_effect=HTTPError(updater.endpoint, 404, "not found", None, None)):
with pytest.raises(HTTPError):
contents = updater.get_update_information()
def test_version_data_no_update(json_data):
global app_name, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=json_data.get("current_version"))
updater.create_session()
results = updater.get_version_data(json_data)
assert results == (False, False, False)
@ -66,7 +56,6 @@ def test_version_data_no_update(json_data):
def test_version_data_update_available(json_data, platform, architecture):
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
with mock.patch("platform.system", return_value=platform):
with mock.patch("platform.architecture", return_value=architecture):
results = updater.get_version_data(json_data)
@ -76,25 +65,23 @@ def test_version_data_update_available(json_data, platform, architecture):
def test_version_data_architecture_not_found(json_data):
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
with mock.patch("platform.system", return_value="nonos"):
with mock.patch("platform.architecture", return_value=("31bits", "")):
with pytest.raises(KeyError):
results = updater.get_version_data(json_data)
def fake_download_function(url, destination, callback):
callback(0, 1024, 2048)
return ("file", "headers")
def test_download_update():
global app_name, current_version, endpoint
updater = core.UpdaterCore(endpoint=endpoint, app_name=app_name, current_version=current_version)
updater.create_session()
with mock.patch("pubsub.pub.sendMessage") as sendMessage_mock:
with mock.patch("io.open") as open_mock:
with requests_mock.Mocker(session=updater.session) as mocker:
updatefile = os.path.join(os.path.dirname(__file__), "helloworld_v2_win64.zip")
mocker.get("http://downloads.update.org/update.zip", body=open(updatefile, "rb"))
result = updater.download_update(update_url="http://downloads.update.org/update.zip", update_destination="update.zip")
open_mock.assert_called_once_with("update.zip", "w+b")
assert result == "update.zip"
assert sendMessage_mock.call_count > 0
with mock.patch("pubsub.pub.sendMessage") as pub_sendMessage:
with mock.patch("urllib.request.urlretrieve", side_effect=fake_download_function):
result = updater.download_update(update_url="http://downloads.update.org/update.zip", update_destination="update.zip")
assert result == "update.zip"
pub_sendMessage.assert_called_once()
def test_extract_archive():
# This only tests if archive extraction methods were called successfully and with the right parameters.

68
test/test_wxupdater.py
View File

@ -100,47 +100,43 @@ def test_on_update_almost_complete():
@mock.patch("tempfile.mkdtemp", return_value="tmp")
def test_check_for_updates_update_available(tempfile):
updater = wxupdater.WXUpdater(endpoint="https://example.com/update.zip", app_name="My awesome application", current_version="0.1")
with mock.patch.object(updater, "create_session") as create_session:
with mock.patch.object(updater, "initialize") as initialize:
with mock.patch.object(updater, "get_update_information") as get_update_information:
with mock.patch.object(updater, "get_version_data") as get_version_data:
with mock.patch.object(updater, "on_new_update_available") as on_new_update_available:
with mock.patch.object(updater, "download_update") as download_update:
with mock.patch.object(updater, "extract_update") as extract_update:
with mock.patch.object(updater, "move_bootstrap") as move_bootstrap:
with mock.patch.object(updater, "on_update_almost_complete") as on_update_almost_complete:
with mock.patch.object(updater, "execute_bootstrap") as execute_bootstrap:
updater.check_for_updates()
execute_bootstrap.assert_called_once()
on_update_almost_complete.assert_called_once()
move_bootstrap.assert_called_once()
extract_update.assert_called_once()
download_update.assert_called_once()
on_new_update_available.assert_called_once()
get_version_data.assert_called_once()
get_update_information.assert_called_once()
initialize.assert_called_once()
create_session.assert_called_once()
with mock.patch.object(updater, "initialize") as initialize:
with mock.patch.object(updater, "get_update_information") as get_update_information:
with mock.patch.object(updater, "get_version_data") as get_version_data:
with mock.patch.object(updater, "on_new_update_available") as on_new_update_available:
with mock.patch.object(updater, "download_update") as download_update:
with mock.patch.object(updater, "extract_update") as extract_update:
with mock.patch.object(updater, "move_bootstrap") as move_bootstrap:
with mock.patch.object(updater, "on_update_almost_complete") as on_update_almost_complete:
with mock.patch.object(updater, "execute_bootstrap") as execute_bootstrap:
updater.check_for_updates()
execute_bootstrap.assert_called_once()
on_update_almost_complete.assert_called_once()
move_bootstrap.assert_called_once()
extract_update.assert_called_once()
download_update.assert_called_once()
on_new_update_available.assert_called_once()
get_version_data.assert_called_once()
get_update_information.assert_called_once()
initialize.assert_called_once()
@mock.patch("tempfile.mkdtemp", return_value="tmp")
def test_check_for_updates_no_update_available(tempfile):
updater = wxupdater.WXUpdater(endpoint="https://example.com/update.zip", app_name="My awesome application", current_version="0.1")
with mock.patch.object(updater, "create_session") as create_session:
with mock.patch.object(updater, "initialize") as initialize:
with mock.patch.object(updater, "get_update_information") as update_information:
with mock.patch.object(updater, "get_version_data", return_value=(False, False, False)) as get_version_data:
result = updater.check_for_updates()
assert result == None
get_version_data.assert_called_once()
with mock.patch.object(updater, "initialize") as initialize:
with mock.patch.object(updater, "get_update_information") as update_information:
with mock.patch.object(updater, "get_version_data", return_value=(False, False, False)) as get_version_data:
result = updater.check_for_updates()
assert result == None
get_version_data.assert_called_once()
@mock.patch("tempfile.mkdtemp", return_value="tmp")
def test_check_for_updates_user_cancelled_update(tempfile):
updater = wxupdater.WXUpdater(endpoint="https://example.com/update.zip", app_name="My awesome application", current_version="0.1")
with mock.patch.object(updater, "create_session") as create_session:
with mock.patch.object(updater, "initialize") as initialize:
with mock.patch.object(updater, "get_update_information") as update_information:
with mock.patch.object(updater, "get_version_data") as get_version_data:
with mock.patch.object(updater, "on_new_update_available", return_value=False) as on_new_update_available:
result = updater.check_for_updates()
assert result == None
on_new_update_available.assert_called_once()
with mock.patch.object(updater, "initialize") as initialize:
with mock.patch.object(updater, "get_update_information") as update_information:
with mock.patch.object(updater, "get_version_data") as get_version_data:
with mock.patch.object(updater, "on_new_update_available", return_value=False) as on_new_update_available:
result = updater.check_for_updates()
assert result == None
on_new_update_available.assert_called_once()

33
updater/core.py
View File

@ -9,7 +9,8 @@ import os
import platform
import zipfile
import logging
import requests
import json
import urllib.request
from pubsub import pub # type: ignore
from typing import Optional, Dict, Tuple, Union, Any
from . import paths
@ -38,22 +39,16 @@ class UpdaterCore(object):
self.app_name = app_name
self.password = password
def create_session(self) -> None:
""" Creates a requests session for calling update server. The session will add an user agent based in parameters passed to :py:class:`updater.core.updaterCore`'s constructor. """
user_agent: str = "%s/%s" % (self.app_name, self.current_version)
self.session: requests.Session = requests.Session()
self.session.headers['User-Agent'] = self.session.headers['User-Agent'] + user_agent
def get_update_information(self) -> Dict[str, Any]:
""" Calls the provided URL endpoint and returns information about the available update sent by the server. The format should adhere to the json specifications for updates.
If the server returns a status code different to 200 or the json file is not valid, this will raise either a :external:py:exc:`requests.HTTPError`, :external:py:exc:`requests.RequestException` or a :external:py:exc:`json.JSONDecodeError`.
If the server returns a status code different to 200 or the json file is not valid, this will raise either a :py:exc:`urllib.error.HTTPError` or a :external:py:exc:`json.JSONDecodeError`.
:rtype: dict
"""
response: requests.Response = self.session.get(self.endpoint)
response.raise_for_status()
content: Dict[str, Any] = response.json()
response = urllib.request.urlopen(self.endpoint)
data: str = response.read()
content: Dict[str, Any] = json.loads(data)
return content
def get_version_data(self, content: Dict[str, Any]) -> Tuple[Union[bool, str], Union[bool, str], Union[bool, str]]:
@ -99,17 +94,11 @@ class UpdaterCore(object):
:returns: The update file path in the system.
:rtype: str
"""
total_downloaded: int = 0
total_size: int = 0
with io.open(update_destination, 'w+b') as outfile:
download: requests.Response = self.session.get(update_url, stream=True)
total_size = int(download.headers.get('content-length', 0))
log.debug("Total update size: %d" % total_size)
download.raise_for_status()
for chunk in download.iter_content(chunk_size):
outfile.write(chunk)
total_downloaded += len(chunk)
pub.sendMessage("updater.update-progress", total_downloaded=total_downloaded, total_size=total_size)
def _download_callback(transferred_blocks, block_size, total_size):
total_downloaded = transferred_blocks*block_size
pub.sendMessage("updater.update-progress", total_downloaded=total_downloaded, total_size=total_size)
filename, headers = urllib.request.urlretrieve(update_url, update_destination, _download_callback)
log.debug("Update downloaded")
return update_destination

1
updater/wxupdater.py
View File

@ -151,7 +151,6 @@ class WXUpdater(core.UpdaterCore):
If there are updates available, displays a dialog to confirm the download of update. If the update downloads successfully, it also extracts and installs it.
"""
self.create_session()
self.initialize()
update_info = self.get_update_information()
version_data = self.get_version_data(update_info)