diff options
Diffstat (limited to 'src/leap/util')
-rw-r--r-- | src/leap/util/__init__.py | 105 | ||||
-rw-r--r-- | src/leap/util/constants.py | 19 | ||||
-rw-r--r-- | src/leap/util/keyring_helpers.py | 37 | ||||
-rw-r--r-- | src/leap/util/leap_argparse.py | 72 | ||||
-rw-r--r-- | src/leap/util/leap_log_handler.py | 134 | ||||
-rw-r--r-- | src/leap/util/privilege_policies.py | 169 | ||||
-rw-r--r-- | src/leap/util/pyside_tests_helper.py | 136 | ||||
-rw-r--r-- | src/leap/util/request_helpers.py | 58 | ||||
-rw-r--r-- | src/leap/util/requirement_checker.py | 101 | ||||
-rw-r--r-- | src/leap/util/streamtologger.py | 59 | ||||
-rw-r--r-- | src/leap/util/tests/__init__.py | 0 | ||||
-rw-r--r-- | src/leap/util/tests/test_is_release_version.py | 57 | ||||
-rw-r--r-- | src/leap/util/tests/test_leap_log_handler.py | 118 | ||||
-rw-r--r-- | src/leap/util/tests/test_streamtologger.py | 122 |
14 files changed, 0 insertions, 1187 deletions
diff --git a/src/leap/util/__init__.py b/src/leap/util/__init__.py deleted file mode 100644 index f39b52e9..00000000 --- a/src/leap/util/__init__.py +++ /dev/null @@ -1,105 +0,0 @@ -# -*- coding: utf-8 -*- -# __init__.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -Initializes version and app info, plus some small and handy functions. -""" -import datetime -import os - -from pkg_resources import parse_version - - -def _is_release_version(version): - """ - Helper to determine whether a version is a final release or not. - The release needs to be of the form: w.x.y.z containing only numbers - and dots. - - :param version: the version string - :type version: str - :returns: if the version is a release version or not. - :rtype: bool - """ - parsed_version = parse_version(version) - not_number = 0 - for x in parsed_version: - try: - int(x) - except: - not_number += 1 - - return not_number == 1 - - -__version__ = "unknown" -IS_RELEASE_VERSION = False - -try: - from leap._version import get_versions - __version__ = get_versions()['version'] - IS_RELEASE_VERSION = _is_release_version(__version__) - del get_versions -except ImportError: - #running on a tree that has not run - #the setup.py setver - pass - -__appname__ = "unknown" -try: - from leap._appname import __appname__ -except ImportError: - #running on a tree that has not run - #the setup.py setver - pass - -__full_version__ = __appname__ + '/' + str(__version__) - - -def first(things): - """ - Return the head of a collection. - """ - try: - return things[0] - except TypeError: - return None - - -def get_modification_ts(path): - """ - Gets modification time of a file. - - :param path: the path to get ts from - :type path: str - :returns: modification time - :rtype: datetime object - """ - ts = os.path.getmtime(path) - return datetime.datetime.fromtimestamp(ts) - - -def update_modification_ts(path): - """ - Sets modification time of a file to current time. - - :param path: the path to set ts to. - :type path: str - :returns: modification time - :rtype: datetime object - """ - os.utime(path, None) - return get_modification_ts(path) diff --git a/src/leap/util/constants.py b/src/leap/util/constants.py deleted file mode 100644 index 63f6b1f7..00000000 --- a/src/leap/util/constants.py +++ /dev/null @@ -1,19 +0,0 @@ -# -*- coding: utf-8 -*- -# constants.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. - -SIGNUP_TIMEOUT = 5 -REQUEST_TIMEOUT = 10 diff --git a/src/leap/util/keyring_helpers.py b/src/leap/util/keyring_helpers.py deleted file mode 100644 index 8f354f28..00000000 --- a/src/leap/util/keyring_helpers.py +++ /dev/null @@ -1,37 +0,0 @@ -# -*- coding: utf-8 -*- -# keyring_helpers.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. - -""" -Keyring helpers. -""" - -import keyring - -OBSOLETE_KEYRINGS = [ - keyring.backends.file.EncryptedKeyring, - keyring.backends.file.PlaintextKeyring -] - - -def has_keyring(): - """ - Returns whether we have an useful keyring to use. - - :rtype: bool - """ - kr = keyring.get_keyring() - return kr is not None and kr.__class__ not in OBSOLETE_KEYRINGS diff --git a/src/leap/util/leap_argparse.py b/src/leap/util/leap_argparse.py deleted file mode 100644 index f60c4e10..00000000 --- a/src/leap/util/leap_argparse.py +++ /dev/null @@ -1,72 +0,0 @@ -# -*- coding: utf-8 -*- -# leap_argparse.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. - -import argparse - -from leap.util import IS_RELEASE_VERSION - - -def build_parser(): - """ - All the options for the leap arg parser - Some of these could be switched on only if debug flag is present! - """ - epilog = "Copyright 2012 The LEAP Encryption Access Project" - parser = argparse.ArgumentParser(description=""" -Launches Bitmask""", epilog=epilog) - parser.add_argument('-d', '--debug', action="store_true", - help=("Launches Bitmask in debug mode, writing debug" - "info to stdout")) - # TODO: when we are ready to disable the --danger flag remove 'True or ' - if True or not IS_RELEASE_VERSION: - help_text = "Bypasses the certificate check for bootstrap" - parser.add_argument('--danger', action="store_true", help=help_text) - - parser.add_argument('-l', '--logfile', metavar="LOG FILE", nargs='?', - action="store", dest="log_file", - #type=argparse.FileType('w'), - help='optional log file') - parser.add_argument('--openvpn-verbosity', nargs='?', - type=int, - action="store", dest="openvpn_verb", - help='verbosity level for openvpn logs [1-6]') - parser.add_argument('-s', '--standalone', action="store_true", - help='Makes Bitmask use standalone' - 'directories for configuration and binary' - 'searching') - - # Not in use, we might want to reintroduce them. - #parser.add_argument('-i', '--no-provider-checks', - #action="store_true", default=False, - #help="skips download of provider config files. gets " - #"config from local files only. Will fail if cannot " - #"find any") - #parser.add_argument('-k', '--no-ca-verify', - #action="store_true", default=False, - #help="(insecure). Skips verification of the server " - #"certificate used in TLS handshake.") - #parser.add_argument('-c', '--config', metavar="CONFIG FILE", nargs='?', - #action="store", dest="config_file", - #type=argparse.FileType('r'), - #help='optional config file') - return parser - - -def init_leapc_args(): - parser = build_parser() - opts, unknown = parser.parse_known_args() - return parser, opts diff --git a/src/leap/util/leap_log_handler.py b/src/leap/util/leap_log_handler.py deleted file mode 100644 index 9adb21a5..00000000 --- a/src/leap/util/leap_log_handler.py +++ /dev/null @@ -1,134 +0,0 @@ -# -*- coding: utf-8 -*- -# leap_log_handler.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -Custom handler for the logger window. -""" -import logging - -from PySide import QtCore - - -class LogHandler(logging.Handler): - """ - This is the custom handler that implements our desired formatting - and also keeps a history of all the logged events. - """ - - MESSAGE_KEY = 'message' - RECORD_KEY = 'record' - - def __init__(self, qtsignal): - """ - LogHander initialization. - Calls parent method and keeps a reference to the qtsignal - that will be used to fire the gui update. - """ - # TODO This is going to eat lots of memory after some time. - # Should be pruned at some moment. - self._log_history = [] - - logging.Handler.__init__(self) - self._qtsignal = qtsignal - - def _get_format(self, logging_level): - """ - Sets the log format depending on the parameter. - It uses html and css to set the colors for the logs. - - :param logging_level: the debug level to define the color. - :type logging_level: str. - """ - log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' - formatter = logging.Formatter(log_format) - - return formatter - - def emit(self, logRecord): - """ - This method is fired every time that a record is logged by the - logging module. - This method reimplements logging.Handler.emit that is fired - in every logged message. - - :param logRecord: the record emitted by the logging module. - :type logRecord: logging.LogRecord. - """ - self.setFormatter(self._get_format(logRecord.levelname)) - log = self.format(logRecord) - log_item = {self.RECORD_KEY: logRecord, self.MESSAGE_KEY: log} - self._log_history.append(log_item) - self._qtsignal(log_item) - - -class HandlerAdapter(object): - """ - New style class that accesses all attributes from the LogHandler. - - Used as a workaround for a problem with multiple inheritance with Pyside - that surfaced under OSX with pyside 1.1.0. - """ - MESSAGE_KEY = 'message' - RECORD_KEY = 'record' - - def __init__(self, qtsignal): - self._handler = LogHandler(qtsignal=qtsignal) - - def setLevel(self, *args, **kwargs): - return self._handler.setLevel(*args, **kwargs) - - def handle(self, *args, **kwargs): - return self._handler.handle(*args, **kwargs) - - @property - def level(self): - return self._handler.level - - -class LeapLogHandler(QtCore.QObject, HandlerAdapter): - """ - Custom logging handler. It emits Qt signals so it can be plugged to a gui. - - Its inner handler also stores an history of logs that can be fetched after - having been connected to a gui. - """ - # All dicts returned are of the form - # {'record': LogRecord, 'message': str} - new_log = QtCore.Signal(dict) - - def __init__(self): - """ - LeapLogHandler initialization. - Initializes parent classes. - """ - QtCore.QObject.__init__(self) - HandlerAdapter.__init__(self, qtsignal=self.qtsignal) - - def qtsignal(self, log_item): - # WARNING: the new-style connection does NOT work because PySide - # translates the emit method to self.emit, and that collides with - # the emit method for logging.Handler - # self.new_log.emit(log_item) - QtCore.QObject.emit( - self, - QtCore.SIGNAL('new_log(PyObject)'), log_item) - - @property - def log_history(self): - """ - Returns the history of the logged messages. - """ - return self._handler._log_history diff --git a/src/leap/util/privilege_policies.py b/src/leap/util/privilege_policies.py deleted file mode 100644 index 72442553..00000000 --- a/src/leap/util/privilege_policies.py +++ /dev/null @@ -1,169 +0,0 @@ -# -*- coding: utf-8 -*- -# privilege_policies.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -Helpers to determine if the needed policies for privilege escalation -are operative under this client run. -""" -import logging -import os -import platform - -from abc import ABCMeta, abstractmethod - -logger = logging.getLogger(__name__) - - -POLICY_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?> -<!DOCTYPE policyconfig PUBLIC - "-//freedesktop//DTD PolicyKit Policy Configuration 1.0//EN" - "http://www.freedesktop.org/standards/PolicyKit/1/policyconfig.dtd"> -<policyconfig> - - <vendor>LEAP Project</vendor> - <vendor_url>https://leap.se/</vendor_url> - - <action id="net.openvpn.gui.leap.run-openvpn"> - <description>Runs the openvpn binary</description> - <description xml:lang="es">Ejecuta el binario openvpn</description> - <message>OpenVPN needs that you authenticate to start</message> - <message xml:lang="es"> - OpenVPN necesita autorizacion para comenzar - </message> - <icon_name>package-x-generic</icon_name> - <defaults> - <allow_any>yes</allow_any> - <allow_inactive>yes</allow_inactive> - <allow_active>yes</allow_active> - </defaults> - <annotate key="org.freedesktop.policykit.exec.path">{path}</annotate> - <annotate key="org.freedesktop.policykit.exec.allow_gui">true</annotate> - </action> -</policyconfig> -""" - - -def is_missing_policy_permissions(): - """ - Returns True if we do not have implemented a policy checker for this - platform, or if the policy checker exists but it cannot find the - appropriate policy mechanisms in place. - - :rtype: bool - """ - _system = platform.system() - platform_checker = _system + "PolicyChecker" - policy_checker = globals().get(platform_checker, None) - if not policy_checker: - # it is true that we miss permission to escalate - # privileges without asking for password each time. - logger.debug("we could not find a policy checker implementation " - "for %s" % (_system,)) - return True - return policy_checker().is_missing_policy_permissions() - - -def get_policy_contents(openvpn_path): - """ - Returns the contents that the policy file should have. - - :param openvpn_path: the openvpn path to use in the polkit file - :type openvpn_path: str - :rtype: str - """ - return POLICY_TEMPLATE.format(path=openvpn_path) - - -def is_policy_outdated(path): - """ - Returns if the existing polkit file is outdated, comparing if the path - is correct. - - :param path: the path that should have the polkit file. - :type path: str. - :rtype: bool - """ - _system = platform.system() - platform_checker = _system + "PolicyChecker" - policy_checker = globals().get(platform_checker, None) - if policy_checker is None: - logger.debug("we could not find a policy checker implementation " - "for %s" % (_system,)) - return False - return policy_checker().is_outdated(path) - - -class PolicyChecker: - """ - Abstract PolicyChecker class - """ - - __metaclass__ = ABCMeta - - @abstractmethod - def is_missing_policy_permissions(self): - """ - Returns True if we could not find any policy mechanisms that - are defined to be in used for this particular platform. - - :rtype: bool - """ - return True - - -class LinuxPolicyChecker(PolicyChecker): - """ - PolicyChecker for Linux - """ - LINUX_POLKIT_FILE = ("/usr/share/polkit-1/actions/" - "net.openvpn.gui.leap.policy") - - @classmethod - def get_polkit_path(self): - """ - Returns the polkit file path. - - :rtype: str - """ - return self.LINUX_POLKIT_FILE - - def is_missing_policy_permissions(self): - """ - Returns True if we could not find the appropriate policykit file - in place - - :rtype: bool - """ - return not os.path.isfile(self.LINUX_POLKIT_FILE) - - def is_outdated(self, path): - """ - Returns if the existing polkit file is outdated, comparing if the path - is correct. - - :param path: the path that should have the polkit file. - :type path: str. - :rtype: bool - """ - polkit = None - try: - with open(self.LINUX_POLKIT_FILE) as f: - polkit = f.read() - except IOError, e: - logger.error("Error reading polkit file(%s): %r" % ( - self.LINUX_POLKIT_FILE, e)) - - return get_policy_contents(path) != polkit diff --git a/src/leap/util/pyside_tests_helper.py b/src/leap/util/pyside_tests_helper.py deleted file mode 100644 index 5c0eb8d6..00000000 --- a/src/leap/util/pyside_tests_helper.py +++ /dev/null @@ -1,136 +0,0 @@ - -'''Helper classes and functions''' - -import os -import unittest - -from random import randint - -from PySide.QtCore import QCoreApplication, QTimer - -try: - from PySide.QtGui import QApplication -except ImportError: - has_gui = False -else: - has_gui = True - - -def adjust_filename(filename, orig_mod_filename): - dirpath = os.path.dirname(os.path.abspath(orig_mod_filename)) - return os.path.join(dirpath, filename) - - -class NoQtGuiError(Exception): - def __init__(self): - Exception.__init__(self, 'No QtGui found') - - -class BasicPySlotCase(object): - '''Base class that tests python slots and signal emissions. - - Python slots are defined as any callable passed to QObject.connect(). - ''' - def setUp(self): - self.called = False - - def tearDown(self): - try: - del self.args - except: - pass - - def cb(self, *args): - '''Simple callback with arbitrary arguments. - - The test function must setup the 'args' attribute with a sequence - containing the arguments expected to be received by this slot. - Currently only a single connection is supported. - ''' - if tuple(self.args) == args: - self.called = True - else: - raise ValueError('Invalid arguments for callback') - - -_instance = None -_timed_instance = None - -if has_gui: - class UsesQApplication(unittest.TestCase): - '''Helper class to provide QApplication instances''' - - qapplication = True - - def setUp(self): - '''Creates the QApplication instance''' - - # Simple way of making instance a singleton - super(UsesQApplication, self).setUp() - global _instance - if _instance is None: - _instance = QApplication([]) - - self.app = _instance - - def tearDown(self): - '''Deletes the reference owned by self''' - del self.app - super(UsesQApplication, self).tearDown() - - class TimedQApplication(unittest.TestCase): - '''Helper class with timed QApplication exec loop''' - - def setUp(self, timeout=100): - '''Setups this Application. - - timeout - timeout in milisseconds''' - global _timed_instance - if _timed_instance is None: - _timed_instance = QApplication([]) - - self.app = _timed_instance - QTimer.singleShot(timeout, self.app.quit) - - def tearDown(self): - '''Delete resources''' - del self.app -else: - class UsesQApplication(unittest.TestCase): - def setUp(self): - raise NoQtGuiError() - - class TimedQapplication(unittest.TestCase): - def setUp(self): - raise NoQtGuiError() - -_core_instance = None - - -class UsesQCoreApplication(unittest.TestCase): - '''Helper class for test cases that require an QCoreApplication - Just connect or call self.exit_app_cb. When called, will ask - self.app to exit. - ''' - - def setUp(self): - '''Set up resources''' - - global _core_instance - if _core_instance is None: - _core_instance = QCoreApplication([]) - - self.app = _core_instance - - def tearDown(self): - '''Release resources''' - del self.app - - def exit_app_cb(self): - '''Quits the application''' - self.app.exit(0) - - -def random_string(size=5): - '''Generate random string with the given size''' - return ''.join(map(chr, [randint(33, 126) for x in range(size)])) diff --git a/src/leap/util/request_helpers.py b/src/leap/util/request_helpers.py deleted file mode 100644 index 74aaa06b..00000000 --- a/src/leap/util/request_helpers.py +++ /dev/null @@ -1,58 +0,0 @@ -# -*- coding: utf-8 -*- -# request_helpers.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. - -""" -Request helpers for backward compatible "parsing" of requests -""" - -import time - -import json - -from dateutil import parser as dateparser - - -def get_content(request): - """ - Returns the content by trying to get it from the json - property/function or from content, in that order. - Also returns the mtime for that content if available - - :param request: request as it is given by requests - :type request: Response - - :rtype: tuple (contents, mtime) - """ - - contents = "" - mtime = None - - if request and request.content and request.json: - if callable(request.json): - contents = json.dumps(request.json()) - else: - contents = json.dumps(request.json) - else: - contents = request.content - - mtime = None - last_modified = request.headers.get('last-modified', None) - if last_modified: - dt = dateparser.parse(unicode(last_modified)) - mtime = int(time.mktime(dt.timetuple()) + dt.microsecond / 1000000.0) - - return contents, mtime diff --git a/src/leap/util/requirement_checker.py b/src/leap/util/requirement_checker.py deleted file mode 100644 index 1d9b9923..00000000 --- a/src/leap/util/requirement_checker.py +++ /dev/null @@ -1,101 +0,0 @@ -# -*- coding: utf-8 -*- -# requirement_checker.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. - -""" -Utility to check the needed requirements. -""" - -import os -import logging - -from pkg_resources import (DistributionNotFound, - get_distribution, - Requirement, - resource_stream, - VersionConflict) - -logger = logging.getLogger(__name__) - - -def get_requirements(): - """ - This function returns a list with requirements. - It checks either if its running from the source or if its installed. - - :returns: a list with packages names, required for the app. - :return type: list of str. - """ - develop = True - requirements = [] - - try: - # if we are running from the source - from pkg import util - requirements = util.parse_requirements() - except ImportError: - develop = False - - # if we are running from the package - if not develop: - requires_file_name = os.path.join('leap', 'util', 'reqs.txt') - dist_name = Requirement.parse('leap-client') - - try: - with resource_stream(dist_name, requires_file_name) as stream: - requirements = [line.strip() for line in stream] - except Exception, e: - logger.error("Requirements file not found. %r" % (e, )) - - return requirements - - -def check_requirements(): - """ - This function check the dependencies declared in the - requirement(s) file(s) and logs the results. - """ - logger.debug("Checking requirements...") - requirements = get_requirements() - - for package in requirements: - try: - get_distribution(package) - except VersionConflict: - required_package = Requirement.parse(package) - required_version = required_package.specs[0] - required_name = required_package.key - - installed_package = get_distribution(required_name) - installed_version = installed_package.version - installed_location = installed_package.location - - msg = "Error: version not satisfied. " - msg += "Expected %s, installed %s (path: %s)." % ( - required_version, installed_version, installed_location) - - result = "%s ... %s" % (package, msg) - logger.error(result) - except DistributionNotFound: - msg = "Error: package not found!" - result = "%s ... %s" % (package, msg) - logger.error(result) - else: - msg = "OK" - result = "%s ... %s" % (package, msg) - logger.debug(result) - - logger.debug('Done') diff --git a/src/leap/util/streamtologger.py b/src/leap/util/streamtologger.py deleted file mode 100644 index 25a06718..00000000 --- a/src/leap/util/streamtologger.py +++ /dev/null @@ -1,59 +0,0 @@ -# -*- coding: utf-8 -*- -# streamtologger.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -Stream object that redirects writes to a logger instance. -""" -import logging - - -class StreamToLogger(object): - """ - Fake file-like stream object that redirects writes to a logger instance. - - Credits to: - http://www.electricmonk.nl/log/2011/08/14/\ - redirect-stdout-and-stderr-to-a-logger-in-python/ - """ - def __init__(self, logger, log_level=logging.INFO): - """ - Constructor, defines the logger and level to use to log messages. - - :param logger: logger object to log messages. - :type logger: logging.Handler - :param log_level: the level to use to log messages through the logger. - :type log_level: int - look at logging-levels in 'logging' docs. - """ - self._logger = logger - self._log_level = log_level - - def write(self, data): - """ - Simulates the 'write' method in a file object. - It writes the data receibed in buf to the logger 'self._logger'. - - :param data: data to write to the 'file' - :type data: str - """ - for line in data.rstrip().splitlines(): - self._logger.log(self._log_level, line.rstrip()) - - def flush(self): - """ - Dummy method. Needed to replace the twisted.log output. - """ - pass diff --git a/src/leap/util/tests/__init__.py b/src/leap/util/tests/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/src/leap/util/tests/__init__.py +++ /dev/null diff --git a/src/leap/util/tests/test_is_release_version.py b/src/leap/util/tests/test_is_release_version.py deleted file mode 100644 index 4199f603..00000000 --- a/src/leap/util/tests/test_is_release_version.py +++ /dev/null @@ -1,57 +0,0 @@ -# -*- coding: utf-8 -*- -# test_is_release_version.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -tests for _is_release_version function -""" - -import unittest -from leap.common.testing.basetest import BaseLeapTest -from leap.util import _is_release_version as is_release_version - - -class TestIsReleaseVersion(BaseLeapTest): - """Tests for release version check.""" - - def setUp(self): - pass - - def tearDown(self): - pass - - def test_git_version(self): - version = '0.2.3-12-ge5b50a1' - self.assertFalse(is_release_version(version)) - - def test_release(self): - version = '0.2.4' - self.assertTrue(is_release_version(version)) - - def test_release_candidate(self): - version = '0.2.4-rc1' - self.assertFalse(is_release_version(version)) - - def test_complex_version(self): - version = '12.5.2.4-rc12.dev.alpha1' - self.assertFalse(is_release_version(version)) - - def test_super_high_version(self): - version = '12.5.2.4.45' - self.assertTrue(is_release_version(version)) - - -if __name__ == "__main__": - unittest.main(verbosity=2) diff --git a/src/leap/util/tests/test_leap_log_handler.py b/src/leap/util/tests/test_leap_log_handler.py deleted file mode 100644 index ea509ea8..00000000 --- a/src/leap/util/tests/test_leap_log_handler.py +++ /dev/null @@ -1,118 +0,0 @@ -# -*- coding: utf-8 -*- -# test_leap_log_handler.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -tests for leap_log_handler -""" - -import unittest - -import logging - -from leap.util.leap_log_handler import LeapLogHandler -from leap.common.testing.basetest import BaseLeapTest -from leap.util.pyside_tests_helper import BasicPySlotCase - -from mock import Mock - - -class LeapLogHandlerTest(BaseLeapTest, BasicPySlotCase): - """ - LeapLogHandlerTest's tests. - """ - def _callback(self, *args): - """ - Simple callback to track if a signal was emitted. - """ - self.called = True - self.emitted_msg = args[0][LeapLogHandler.MESSAGE_KEY] - - def setUp(self): - BasicPySlotCase.setUp(self) - - # Create the logger - level = logging.DEBUG - self.logger = logging.getLogger(name='test') - self.logger.setLevel(level) - - # Create the handler - self.leap_handler = LeapLogHandler() - self.leap_handler.setLevel(level) - self.logger.addHandler(self.leap_handler) - - def tearDown(self): - BasicPySlotCase.tearDown(self) - try: - self.leap_handler.new_log.disconnect() - except Exception: - pass - - def test_history_starts_empty(self): - self.assertEqual(self.leap_handler.log_history, []) - - def test_one_log_captured(self): - self.logger.debug('test') - self.assertEqual(len(self.leap_handler.log_history), 1) - - def test_history_records_order(self): - self.logger.debug('test 01') - self.logger.debug('test 02') - self.logger.debug('test 03') - - logs = [] - for message in self.leap_handler.log_history: - logs.append(message[LeapLogHandler.RECORD_KEY].msg) - - self.assertIn('test 01', logs) - self.assertIn('test 02', logs) - self.assertIn('test 03', logs) - - def test_history_messages_order(self): - self.logger.debug('test 01') - self.logger.debug('test 02') - self.logger.debug('test 03') - - logs = [] - for message in self.leap_handler.log_history: - logs.append(message[LeapLogHandler.MESSAGE_KEY]) - - self.assertIn('test 01', logs[0]) - self.assertIn('test 02', logs[1]) - self.assertIn('test 03', logs[2]) - - def test_emits_signal(self): - log_format = '%(name)s - %(levelname)s - %(message)s' - formatter = logging.Formatter(log_format) - get_format = Mock(return_value=formatter) - self.leap_handler._handler._get_format = get_format - - self.leap_handler.new_log.connect(self._callback) - self.logger.debug('test') - - expected_log_msg = "test - DEBUG - test" - - # signal emitted - self.assertTrue(self.called) - - # emitted message - self.assertEqual(self.emitted_msg, expected_log_msg) - - # Mock called - self.assertTrue(get_format.called) - - -if __name__ == "__main__": - unittest.main() diff --git a/src/leap/util/tests/test_streamtologger.py b/src/leap/util/tests/test_streamtologger.py deleted file mode 100644 index 4c98e562..00000000 --- a/src/leap/util/tests/test_streamtologger.py +++ /dev/null @@ -1,122 +0,0 @@ -# -*- coding: utf-8 -*- -# test_streamtologger.py -# Copyright (C) 2013 LEAP -# -# This program 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, either version 3 of the License, or -# (at your option) any later version. -# -# This program 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 this program. If not, see <http://www.gnu.org/licenses/>. -""" -tests for streamtologger -""" - -try: - import unittest2 as unittest -except ImportError: - import unittest - -import logging -import sys - -from leap.util.streamtologger import StreamToLogger -from leap.common.testing.basetest import BaseLeapTest - - -class SimpleLogHandler(logging.Handler): - """ - The simplest log handler that allows to check if the log was - delivered to the handler correctly. - """ - def __init__(self): - logging.Handler.__init__(self) - self._last_log = "" - self._last_log_level = "" - - def emit(self, record): - self._last_log = record.getMessage() - self._last_log_level = record.levelno - - def get_last_log(self): - """ - Returns the last logged message by this handler. - - :return: the last logged message. - :rtype: str - """ - return self._last_log - - def get_last_log_level(self): - """ - Returns the level of the last logged message by this handler. - - :return: the last logged level. - :rtype: str - """ - return self._last_log_level - - -class StreamToLoggerTest(BaseLeapTest): - """ - StreamToLogger's tests. - - NOTE: we may need to find a way to test the use case that an exception - is raised. I couldn't catch the output of an exception because the - test failed if some exception is raised. - """ - def setUp(self): - # Create the logger - level = logging.DEBUG - self.logger = logging.getLogger(name='test') - self.logger.setLevel(level) - - # Simple log handler - self.handler = SimpleLogHandler() - self.logger.addHandler(self.handler) - - # Preserve original values - self._sys_stdout = sys.stdout - self._sys_stderr = sys.stderr - - # Create the handler - sys.stdout = StreamToLogger(self.logger, logging.DEBUG) - sys.stderr = StreamToLogger(self.logger, logging.ERROR) - - def tearDown(self): - # Restore original values - sys.stdout = self._sys_stdout - sys.stderr = self._sys_stderr - - def test_logger_starts_empty(self): - self.assertEqual(self.handler.get_last_log(), '') - - def test_standard_output(self): - message = 'Test string' - print message - - log = self.handler.get_last_log() - log_level = self.handler.get_last_log_level() - - self.assertEqual(log, message) - self.assertEqual(log_level, logging.DEBUG) - - def test_standard_error(self): - message = 'Test string' - sys.stderr.write(message) - - log_level = self.handler.get_last_log_level() - log = self.handler.get_last_log() - - self.assertEqual(log, message) - self.assertEqual(log_level, logging.ERROR) - - -if __name__ == "__main__": - unittest.main(verbosity=2) |