diff --git a/src/UDSClient.py b/src/UDSClient.py index ecee70e..91f5eee 100755 --- a/src/UDSClient.py +++ b/src/UDSClient.py @@ -34,6 +34,7 @@ Author: Adolfo Gómez, dkmaster at dkmon dot com ''' import contextlib +import logging import sys import os import platform @@ -96,7 +97,7 @@ def close_window(self) -> None: self.close() def show_error(self, error: Exception) -> None: - logger.error('got error: %s', error) + logger.error('Error: %s', error) self.stop_animation() # In fact, main window is hidden, so this is not visible... :) self.ui.info.setText('UDS Plugin Error') @@ -132,16 +133,16 @@ def stop_animation(self) -> None: def fetch_version(self) -> None: try: self.api.get_version() - except exceptions.InvalidVersion as e: + except exceptions.InvalidVersionException as e: UDSClient.error_message( 'Upgrade required', - 'A newer connector version is required.\nA browser will be opened to download it.', + f'UDS Client version {e.required_version} is required.\nA browser will be opened to download it.', ) - webbrowser.open(e.downloadUrl) + webbrowser.open(e.link) self.close_window() return - except Exception as e: # pylint: disable=broad-exception-caught - if logger.getEffectiveLevel() == 10: + except Exception as e: + if logger.getEffectiveLevel() == logging.DEBUG: logger.exception('Get Version') self.show_error(e) self.close_window() @@ -169,8 +170,9 @@ def fetch_transport_data(self) -> None: self.ui.info.setText(str(e) + ', retrying access...') # Retry operation in ten seconds QtCore.QTimer.singleShot(10000, self.fetch_transport_data) - except Exception as e: # pylint: disable=broad-exception-caught - if logger.getEffectiveLevel() == 10: + except Exception as e: + # If debug is enabled, show exception + if logger.getEffectiveLevel() == logging.DEBUG: logger.exception('Get Transport Data') self.show_error(e) @@ -192,7 +194,7 @@ def start(self) -> None: Starts proccess by requesting version info """ self.ui.info.setText('Initializing...') - QtCore.QTimer.singleShot(100, self.fetch_version) + QtCore.QTimer.singleShot(100, self.fetch_version) # Will make it async, not blocking the gui @staticmethod def warning_message(title: str, message: str, *, yes_no: bool = False) -> bool: @@ -306,12 +308,12 @@ def minimal(api: RestApi, ticket: str, scrambler: str) -> int: logger.debug('Getting version') try: api.get_version() - except exceptions.InvalidVersion as e: + except exceptions.InvalidVersionException as e: UDSClient.error_message( 'Upgrade required', 'A newer connector version is required.\nA browser will be opened to download it.', ) - webbrowser.open(e.downloadUrl) + webbrowser.open(e.link) return 0 logger.debug('Transport data') script, params = api.get_script_and_parameters(ticket, scrambler) diff --git a/src/uds/exceptions.py b/src/uds/exceptions.py index ef369fb..4ca5b1a 100644 --- a/src/uds/exceptions.py +++ b/src/uds/exceptions.py @@ -14,9 +14,10 @@ class RetryException(UDSException): pass -class InvalidVersion(UDSException): - downloadUrl: str +class InvalidVersionException(UDSException): + link: str - def __init__(self, downloadUrl: str) -> None: - super().__init__(downloadUrl) - self.downloadUrl = downloadUrl + def __init__(self, client_link: str, required_version: str) -> None: + super().__init__(client_link) + self.link = client_link + self.required_version = required_version diff --git a/src/uds/rest.py b/src/uds/rest.py index bd935a7..691bfdd 100644 --- a/src/uds/rest.py +++ b/src/uds/rest.py @@ -56,7 +56,7 @@ class RestApi: _rest_api_endpoint: str # base Rest API URL _on_invalid_certificate: typing.Optional[CertCallbackType] - _server_version: str + _required_version: str def __init__( self, @@ -67,7 +67,7 @@ def __init__( self._rest_api_endpoint = rest_api_endpoint self._on_invalid_certificate = on_invalid_certificate - self._server_version = '' + self._required_version = '' def get(self, path: str, params: typing.Optional[typing.Mapping[str, str]] = None) -> typing.Any: if params: @@ -79,7 +79,8 @@ def get(self, path: str, params: typing.Optional[typing.Mapping[str, str]] = Non def process_error(self, data: typing.Any) -> None: if 'error' in data: - if data.get('retryable', '0') == '1': + # Get retrayable from data, if not present, use old key + if data.get('is_retrayable', data.get('retryable', '0')) == '1': raise exceptions.RetryException(data['error']) raise exceptions.UDSException(data['error']) @@ -89,19 +90,25 @@ def get_version(self) -> str: Also checks that the version is valid for us. If not, will raise an "InvalidVersion' exception''' - downloadUrl = '' - if not self._server_version: - data = self.get('') + client_link = '' + if not self._required_version: + data = self.get('') # Version is returned on 'main' path self.process_error(data) - self._server_version = data['result']['requiredVersion'] - downloadUrl = data['result']['downloadUrl'] + # get server version, using new key but, if not present, use old one + # Note: old version will be removed on 5.0.0 (As all 4.0 brokers will already return the new keys) + if 'requiredVersion' in data['result']: + self._required_version = data['result']['requiredVersion'] + client_link = data['result']['downloadUrl'] + else: + self._required_version = data['result']['required_version'] + client_link = data['result']['client_link'] try: - if self._server_version > consts.VERSION: - raise exceptions.InvalidVersion(downloadUrl) + if self._required_version > consts.VERSION: + raise exceptions.InvalidVersionException(client_link, self._required_version) - return self._server_version - except exceptions.InvalidVersion: + return self._required_version + except exceptions.InvalidVersionException: raise except Exception as e: raise exceptions.UDSException(e) from e diff --git a/src/uds/tools.py b/src/uds/tools.py index 775e9c9..f627f1f 100644 --- a/src/uds/tools.py +++ b/src/uds/tools.py @@ -48,7 +48,16 @@ from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding -import psutil +try: + import psutil + + def process_iter(*args: typing.Any, **kwargs: typing.Any) -> typing.Any: + return psutil.process_iter(*args, **kwargs) + +except ImportError: + def process_iter(*args: typing.Any, **kwargs: typing.Any) -> typing.Any: + return [] + from . import consts from .log import logger @@ -164,17 +173,11 @@ def wait_for_tasks() -> None: elif hasattr(task, 'wait'): task.wait() # If wait for spanwed process (look for process with task pid) and we can look for them... - logger.debug( - 'Psutil: %s, waitForSubp: %s, hasattr: %s', - psutil, - wait_subprocesses, - hasattr(task, 'pid'), - ) if psutil and wait_subprocesses and hasattr(task, 'pid'): subprocesses: list['psutil.Process'] = list( filter( lambda x: x.ppid() == task.pid, # type x: psutil.Process - psutil.process_iter(attrs=('ppid',)), + process_iter(attrs=('ppid',)), ) ) logger.debug('Waiting for subprocesses... %s, %s', task.pid, subprocesses) diff --git a/tests/test_main.py b/tests/test_main.py index fffd970..264bc00 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -30,6 +30,8 @@ ''' import logging import typing +import sys +import os from unittest import TestCase, mock import UDSClient @@ -41,6 +43,11 @@ class TestClient(TestCase): + def setUp(self) -> None: + # If linux, and do not have X11, we will skip the tests + if sys.platform == 'linux' and 'DISPLAY' not in os.environ: + self.skipTest('Skipping test on linux without X11') + def test_commandline(self): def _check_url(url: str, minimal: typing.Optional[str] = None, with_minimal: bool = False) -> None: host, ticket, scrambler, use_minimal = UDSClient.parse_arguments( @@ -83,20 +90,70 @@ def _check_url(url: str, minimal: typing.Optional[str] = None, with_minimal: boo def test_rest(self): # This is a simple test, we will test the rest api is mocked correctly with fixtures.patch_rest_api() as api: - self.assertEqual(api.get_version(), fixtures.SERVER_VERSION) - self.assertEqual(api.get_script_and_parameters('ticket', 'scrambler'), (fixtures.SCRIPT, fixtures.PARAMETERS)) - + self.assertEqual(api.get_version(), fixtures.REQUIRED_VERSION) + self.assertEqual( + api.get_script_and_parameters('ticket', 'scrambler'), (fixtures.SCRIPT, fixtures.PARAMETERS) + ) + from_api = rest.RestApi.api('host', lambda x, y: True) # Repeat tests, should return same results - self.assertEqual(from_api.get_version(), fixtures.SERVER_VERSION) - self.assertEqual(from_api.get_script_and_parameters('ticket', 'scrambler'), (fixtures.SCRIPT, fixtures.PARAMETERS)) + self.assertEqual(from_api.get_version(), fixtures.REQUIRED_VERSION) + self.assertEqual( + from_api.get_script_and_parameters('ticket', 'scrambler'), + (fixtures.SCRIPT, fixtures.PARAMETERS), + ) # And also, the api is the same self.assertEqual(from_api, api) def test_udsclient(self): - # This is a simple test, we will test the rest api is mocked correctly with fixtures.patched_uds_client() as client: # patch UDSClient module waiting_tasks_processor to avoid waiting for tasks with mock.patch('UDSClient.waiting_tasks_processor'): + # Patch builting "exec" + with mock.patch('builtins.exec') as builtins_exec: + # Desencadenate the full process + client.fetch_version() + + # These are in fact mocks, but type checker does not know that + client.api.get_version.assert_called_with() # type: ignore + client.api.get_script_and_parameters.assert_called_with(client.ticket, client.scrambler) # type: ignore + + # Builtin exec should be called with: + # - The script + # - The globals, because the globals in scripts may be different, we use mock.ANY + # - The locals -> {'parent': self, 'sp': params}, where self is the client and params is the parameters + builtins_exec.assert_called_with( + fixtures.SCRIPT, mock.ANY, {'parent': client, 'sp': fixtures.PARAMETERS} + ) + + # And also, process_waiting_tasks should be called, to process remaining tasks + client.process_waiting_tasks.assert_called_with() # type: ignore + + logger.debug('Testing fetch_script') + + def test_udsclient_invalid_version(self): + with fixtures.patched_uds_client() as client: + with mock.patch('webbrowser.open') as webbrowser_open: + fixtures.REQUIRED_VERSION = '.'.join( + str(int(x) + 1) for x in consts.VERSION.split('.') + ) # This will make the version greater than the required + client.fetch_version() + + # error message should be called to show the required new version + # but we do not check message content, just that it was called + # It's an static method, so we can check it directly + UDSClient.UDSClient.error_message.assert_called() # type: ignore + webbrowser_open.assert_called_with(fixtures.CLIENT_LINK) + + def test_udsclient_error_version(self): + with fixtures.patched_uds_client() as client: + with mock.patch('webbrowser.open') as webbrowser_open: + fixtures.REQUIRED_VERSION = 'fail' client.fetch_version() - + + # error message should be called to show problem checking version + UDSClient.UDSClient.error_message.assert_called() # type: ignore + # webrowser should not be called + webbrowser_open.assert_not_called() + + self.assertTrue(client.has_error) diff --git a/tests/utils/fixtures.py b/tests/utils/fixtures.py index cdb0b6c..c3e78d6 100644 --- a/tests/utils/fixtures.py +++ b/tests/utils/fixtures.py @@ -34,11 +34,13 @@ from unittest import mock import UDSClient +from uds import consts, exceptions from uds import rest, ui from . import autospec -SERVER_VERSION: str = '4.0.0' +REQUIRED_VERSION: str = '4.0.0' +CLIENT_LINK: str = 'https://sample.client.link/udsclient.downloadable' SCRIPT: str = ''' # TODO: add testing script here ''' @@ -46,8 +48,17 @@ # TODO: add parameters here } + +def check_version() -> str: + if REQUIRED_VERSION == 'fail': + raise Exception('Version check failed miserably! :) (just for testing)') + if consts.VERSION < REQUIRED_VERSION: + raise exceptions.InvalidVersionException(CLIENT_LINK, REQUIRED_VERSION) + return REQUIRED_VERSION + + REST_METHODS_INFO: typing.List[autospec.AutoSpecMethodInfo] = [ - autospec.AutoSpecMethodInfo(rest.RestApi.get_version, return_value=SERVER_VERSION), + autospec.AutoSpecMethodInfo(rest.RestApi.get_version, method=check_version), autospec.AutoSpecMethodInfo(rest.RestApi.get_script_and_parameters, return_value=(SCRIPT, PARAMETERS)), ] @@ -78,5 +89,14 @@ def patch_rest_api( def patched_uds_client() -> typing.Generator['UDSClient.UDSClient', None, None]: app = ui.QtWidgets.QApplication([]) with patch_rest_api() as client: - yield UDSClient.UDSClient(client, 'ticket', 'scrambler') - app.quit() \ No newline at end of file + uds_client = UDSClient.UDSClient(client, 'ticket', 'scrambler') + # Now, patch object: + # - process_waiting_tasks so we do not launch any task + # - error_message so we do not show any error message + # - warning_message so we do not show any warning message + # error_message and warning_message are static methods, so we need to patch them on the class + with mock.patch.object(uds_client, 'process_waiting_tasks'), mock.patch( + 'UDSClient.UDSClient.error_message' + ), mock.patch('UDSClient.UDSClient.warning_message'): + yield uds_client + app.quit()