LibrariesManager

Complexity

Total Complexity

39

Size/Duplication

Total Lines	246
Duplicated Lines	0 %

Test Coverage

Coverage

64.21%

9 Methods

Rating	Name	Duplication	Size	Complexity
C	_insert_paths_unix()	0	34	7
A	_insert_paths_windows()	0	18	2
B	setup()	0	33	4
B	_libraries_path()	0	29	5
B	reset()	0	25	4
A	_insert_paths()	0	20	4
A	_insert_architecture_paths()	0	17	3
A	_cache_libraries()	0	13	2
D	test()	0	46	8

from plugin.core.configuration import Configuration

The name configuration does not seem to exist in module plugin.core.

Unable to import 'plugin.core.configuration' (invalid syntax (<string>, line 38))

from plugin.core.environment import Environment
from plugin.core.helpers.variable import merge
from plugin.core.libraries.cache import CacheManager
from plugin.core.libraries.constants import CONTENTS_PATH, NATIVE_DIRECTORIES, UNICODE_MAP
from plugin.core.libraries.helpers import PathHelper, StorageHelper, SystemHelper
from plugin.core.libraries.tests import LIBRARY_TESTS
from plugin.core.logger.handlers.error_reporter import RAVEN

The name error_reporter does not seem to exist in module plugin.core.logger.handlers.

Unable to import 'plugin.core.logger.handlers.error_reporter' (invalid syntax (<string>, line 133))

import logging
import os
import sys

log = logging.getLogger(__name__)


class LibrariesManager(object):
    @classmethod
    def setup(cls, cache=False):
        """Setup native library directories

        :param cache: Enable native library caching
        :type cache: bool
        """

        # Use `cache` value from advanced configuration
        cache = Configuration.advanced['libraries'].get_boolean('cache', cache)

        # Retrieve libraries path (and cache libraries, if enabled)
        libraries_path = cls._libraries_path(cache)

        if not libraries_path:
            return

        log.info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))

        # Remove current native library directories from `sys.path`
        cls.reset()

        # Insert platform specific library paths
        cls._insert_paths(libraries_path)

        # Display library paths in logfile
        for path in sys.path:
            path = os.path.abspath(path)

            if StorageHelper.is_framework_path(path):
                continue

            log.info('[PATH] %s', StorageHelper.to_relative_path(path))

    @staticmethod
    def test():
        """Test native libraries to ensure they can be correctly loaded"""
        log.info('Testing native library support...')

        metadata = {}

        for test in LIBRARY_TESTS:
            # Run tests
            result = test.run()

The Class LList does not seem to have a member named run.

The Class PyOpenSSL does not seem to have a member named run.

The Class Apsw does not seem to have a member named run.

The Class Cryptography does not seem to have a member named run.

            if not result.get('success'):
                # Write message to logfile
                log.error('%s: unavailable - %s', test.name, result.get('message'), exc_info=result.get('exc_info'))

                if not test.optional:
                    return

                continue

            # Test successful
            t_metadata = result.get('metadata') or {}
            t_versions = t_metadata.get('versions')

            if t_versions:
                if len(t_versions) > 1:
                    log.info('%s: available (%s)', test.name, ', '.join([
                        '%s: %s' % (key, value)
                        for key, value in t_versions.items()
                    ]))
                else:
                    key = t_versions.keys()[0]

                    log.info('%s: available (%s)', test.name, t_versions[key])
            else:
                log.info('%s: available', test.name)

            # Merge result into `metadata`
            merge(metadata, t_metadata, recursive=True)

        # Include versions in error reports
        versions = metadata.get('versions') or {}

        RAVEN.tags.update(dict([
            ('%s.version' % key, value)
            for key, value in versions.items()
        ]))

    @classmethod
    def reset(cls):
        """Remove all the native library directives from `sys.path`"""

        for path in sys.path:
            path = os.path.abspath(path)

            if not path.lower().startswith(CONTENTS_PATH.lower()):
                continue

            # Convert to relative path
            path_rel = os.path.relpath(path, CONTENTS_PATH)

            # Take the first two fragments
            path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])

            # Convert to unix-style separators (/)
            path_rel = path_rel.replace('\\', '/')

            # Ignore non-native library directories
            if path_rel not in NATIVE_DIRECTORIES:
                continue

            # Remove from `sys.path`
            PathHelper.remove(path)

    @classmethod
    def _libraries_path(cls, cache=False):
        """Retrieve the native libraries base directory (and cache the libraries if enabled)

        :param cache: Enable native library caching
        :type cache: bool
        """

        # Use specified libraries path (from "advanced.ini')
        libraries_path = Configuration.advanced['libraries'].get('libraries_path')

        if libraries_path and os.path.exists(libraries_path):
            log.info('Using libraries at %r', StorageHelper.to_relative_path(libraries_path))
            RAVEN.tags.update({'libraries.source': 'custom'})
            return libraries_path

        # Use system libraries (if bundled libraries have been disabled in "advanced.ini")
        if not Configuration.advanced['libraries'].get_boolean('bundled', True):
            log.info('Bundled libraries have been disabled, using system libraries')
            RAVEN.tags.update({'libraries.source': 'system'})
            return None

        # Cache libraries (if enabled)
        if cache:
            RAVEN.tags.update({'libraries.source': 'cache'})
            return cls._cache_libraries()

        RAVEN.tags.update({'libraries.source': 'bundle'})
        return Environment.path.libraries

    @classmethod
    def _cache_libraries(cls):
        cache_path = Configuration.advanced['libraries'].get('cache_path')

        # Try cache libraries to `cache_path`
        libraries_path = CacheManager.sync(cache_path)

        if not libraries_path:
            log.info('Unable to cache libraries, using bundled libraries directly')
            return Environment.path.libraries

        log.info('Cached libraries to %r', StorageHelper.to_relative_path(libraries_path))
        return libraries_path

    @classmethod
    def _insert_paths(cls, libraries_path):
        # Retrieve system details
        system = SystemHelper.name()
        architecture = SystemHelper.architecture()

        if not architecture:
            return

        log.debug('System: %r, Architecture: %r', system, architecture)

        # Insert architecture specific libraries
        architectures = [architecture]

        if architecture == 'i686':
            # Fallback to i386
            architectures.append('i386')

        for arch in architectures + ['universal']:
            cls._insert_architecture_paths(libraries_path, system, arch)

    @classmethod
    def _insert_architecture_paths(cls, libraries_path, system, architecture):
        architecture_path = os.path.join(libraries_path, system, architecture)

        if not os.path.exists(architecture_path):
            return

        # Architecture libraries
        PathHelper.insert(libraries_path, system, architecture)

        # System libraries
        if system == 'Windows':
            # Windows libraries (VC++ specific)
            cls._insert_paths_windows(libraries_path, system, architecture)
        else:
            # Darwin/FreeBSD/Linux libraries
            cls._insert_paths_unix(libraries_path, system, architecture)

    @staticmethod
    def _insert_paths_unix(libraries_path, system, architecture):
        # UCS specific libraries
        ucs = UNICODE_MAP.get(sys.maxunicode)
        log.debug('UCS: %r', ucs)

        if ucs:
            PathHelper.insert(libraries_path, system, architecture, ucs)

        # CPU specific libraries
        cpu_type = SystemHelper.cpu_type()
        page_size = SystemHelper.page_size()

        log.debug('CPU Type: %r', cpu_type)
        log.debug('Page Size: %r', page_size)

        if cpu_type:
            PathHelper.insert(libraries_path, system, architecture, cpu_type)

            if page_size:
                PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size))

        # UCS + CPU specific libraries
        if cpu_type and ucs:
            PathHelper.insert(libraries_path, system, architecture, cpu_type, ucs)

            if page_size:
                PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size), ucs)

        # Include attributes in error reports
        RAVEN.tags.update({
            'cpu.type': cpu_type,
            'memory.page_size': page_size,
            'python.ucs': ucs
        })

    @staticmethod
    def _insert_paths_windows(libraries_path, system, architecture):
        vcr = SystemHelper.vcr_version() or 'vc12'  # Assume "vc12" if call fails
        ucs = UNICODE_MAP.get(sys.maxunicode)

        log.debug('VCR: %r, UCS: %r', vcr, ucs)

        # VC++ libraries
        PathHelper.insert(libraries_path, system, architecture, vcr)

        # UCS libraries
        if ucs:
            PathHelper.insert(libraries_path, system, architecture, vcr, ucs)

        # Include attributes in error reports
        RAVEN.tags.update({
            'python.ucs': ucs,
            'vcr.version': vcr
        })