BaseTest.find_python_executable() - Code Metrics - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

BaseTest.find_python_executable() F
last analyzed 2018-03-05 01:11 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	14.8108

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	33
ccs	13
cts	19
cp	0.6842
rs	3.1764
cc	11
crap	14.8108

How to fix Complexity

from plugin.core.helpers.variable import merge

from subprocess import Popen
import json
import logging
import os
import subprocess
import sys

CURRENT_PATH = os.path.abspath(__file__)
HOST_PATH = os.path.join(os.path.dirname(CURRENT_PATH), 'host.py')

log = logging.getLogger(__name__)


class BaseTest(object):
    name = None
    optional = False

    @classmethod
    def run(cls, search_paths):
        metadata = {}

        message = None
        success = None

        # Retrieve names of test functions
        names = [
            name for name in dir(cls)
            if name.startswith('test_')
        ]

        if not names:
            return cls.build_failure('No tests defined')

        # Run tests
        for name in names:
            # Ensure function exists
            if not hasattr(cls, name):
                return cls.build_failure('Unable to find function: %r' % name)

            # Run test
            try:
                result = cls.spawn(name, search_paths)

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

                # Test successful
                message = None
                success = True
            except Exception as ex:

                if success:
                    continue

                message = ex.message

                success = False

        if not success:
            # Trigger event
            cls.on_failure(message)

            # Build result
            return cls.build_failure(message)

        # Trigger event
        cls.on_success(metadata)

        # Build result
        return cls.build_success(metadata)

    @classmethod
    def spawn(cls, name, search_paths):
        # Find path to python executable
        python_exe = cls.find_python_executable()

        if not python_exe:
            raise Exception('Unable to find python executable')

        # Ensure test host exists
        if not os.path.exists(HOST_PATH):
            raise Exception('Unable to find "host.py" script')

        # Build test process arguments
        args = [
            python_exe, HOST_PATH,
            '--module', cls.__module__,
            '--name', name,

            '--search-paths="%s"' % (
                ';'.join(search_paths)
            ),
        ]

        # Spawn test (in sub-process)
        log.debug('Starting test: %s:%s', cls.__module__, name)

        process = Popen(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        # Wait for test to complete
        stdout, stderr = process.communicate()

        if stderr:
            log.debug('Test returned messages:\n%s', stderr.replace("\r\n", "\n"))

        # Parse output
        result = None

        if stdout:
            try:
                result = json.loads(stdout)
            except Exception as ex:

                log.warn('Invalid output returned %r - %s', stdout, ex, exc_info=True)

        # Build result
        if process.returncode != 0:
            # Test failed
            if result and result.get('message'):
                if result.get('traceback'):
                    log.info('%s - %s', result['message'], result['traceback'])

                raise Exception(result['message'])

            raise Exception('Unknown error (code: %s)' % process.returncode)

        # Test successful
        return result

    @classmethod
    def find_python_executable(cls):
        candidates = [sys.executable]

        # Add candidates based on the script path in `sys.argv`
        if sys.argv and len(sys.argv) > 0 and os.path.exists(sys.argv[0]):
            bootstrap_path = sys.argv[0]
            resources_pos = bootstrap_path.lower().find('resources')

            if resources_pos > 0:
                pms_path = bootstrap_path[:resources_pos]

                cls._add_python_home_candidates(candidates, pms_path)

        # Add candidates relative to `PLEX_MEDIA_SERVER_HOME`
        pms_home = os.environ.get('PLEX_MEDIA_SERVER_HOME')

        if pms_home and os.path.exists(pms_home):
            cls._add_python_home_candidates(candidates, pms_home)

        # Add candidates relative to `PYTHONHOME`
        python_home = os.environ.get('PYTHONHOME')

        if python_home and os.path.exists(python_home):
            candidates.append(os.path.join(python_home, 'bin', 'python'))

        # Use first candidate that exists
        for path in candidates:
            if os.path.exists(path):
                return path

        log.warn('Unable to find python executable', extra={'candidates': candidates})
        return None

    @staticmethod
    def _add_python_home_candidates(candidates, path):
        # Windows
        candidates.append(os.path.join(path, 'PlexScriptHost.exe'))

        # *nix
        candidates.append(os.path.join(path, 'Plex Script Host'))
        candidates.append(os.path.join(path, 'Resources', 'Plex Script Host'))
        candidates.append(os.path.join(path, 'Resources', 'Python', 'bin', 'python'))

    #
    # Events
    #

    @classmethod
    def on_failure(cls, message):
        pass

    @classmethod
    def on_success(cls, metadata):
        pass

    #
    # Helpers
    #

    @classmethod
    def build_exception(cls, message, exc_info=None):
        if exc_info is None:
            exc_info = sys.exc_info()

        return cls.build_failure(
            message,
            exc_info=exc_info
        )

    @classmethod
    def build_failure(cls, message, **kwargs):
        result = {
            'success': False,
            'message': message
        }

        # Merge extra attributes
        merge(result, kwargs)

        return result

    @staticmethod
    def build_success(metadata):
        return {
            'success': True,
            'metadata': metadata
        }


1	1	from plugin.core.helpers.variable import merge
2
3	1	from subprocess import Popen
4	1	import json
5	1	import logging
6	1	import os
7	1	import subprocess
8	1	import sys
9
10	1	CURRENT_PATH = os.path.abspath(__file__)
11	1	HOST_PATH = os.path.join(os.path.dirname(CURRENT_PATH), 'host.py')
12
13	1	log = logging.getLogger(__name__)
14
15
16	1	class BaseTest(object):
17	1	name = None
18	1	optional = False
19
20	1	@classmethod
21		def run(cls, search_paths):
22	1	metadata = {}
23
24	1	message = None
25	1	success = None
26
27		# Retrieve names of test functions
28	1	names = [
29		name for name in dir(cls)
30		if name.startswith('test_')
31		]
32
33	1	if not names:
34		return cls.build_failure('No tests defined')
35
36		# Run tests
37	1	for name in names:
38		# Ensure function exists
39	1	if not hasattr(cls, name):
40		return cls.build_failure('Unable to find function: %r' % name)
41
42		# Run test
43	1	try:
44	1	result = cls.spawn(name, search_paths)
45
46		# Merge test result into `metadata`
47	1	merge(metadata, result, recursive=True)
48
49		# Test successful
50	1	message = None
51	1	success = True
52		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
53		if success:
54		continue
55
56		message = ex.message
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Instance of `Exception` does not seem to have a member named `message`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
57		success = False
58
59	1	if not success:
60		# Trigger event
61		cls.on_failure(message)
62
63		# Build result
64		return cls.build_failure(message)
65
66		# Trigger event
67	1	cls.on_success(metadata)
68
69		# Build result
70	1	return cls.build_success(metadata)
71
72	1	@classmethod
73		def spawn(cls, name, search_paths):
74		# Find path to python executable
75	1	python_exe = cls.find_python_executable()
76
77	1	if not python_exe:
78		raise Exception('Unable to find python executable')
79
80		# Ensure test host exists
81	1	if not os.path.exists(HOST_PATH):
82		raise Exception('Unable to find "host.py" script')
83
84		# Build test process arguments
85	1	args = [
86		python_exe, HOST_PATH,
87		'--module', cls.__module__,
88		'--name', name,
89
90		'--search-paths="%s"' % (
91		';'.join(search_paths)
92		),
93		]
94
95		# Spawn test (in sub-process)
96	1	log.debug('Starting test: %s:%s', cls.__module__, name)
97
98	1	process = Popen(
99		args,
100		stdout=subprocess.PIPE,
101		stderr=subprocess.PIPE
102		)
103
104		# Wait for test to complete
105	1	stdout, stderr = process.communicate()
106
107	1	if stderr:
108		log.debug('Test returned messages:\n%s', stderr.replace("\r\n", "\n"))
109
110		# Parse output
111	1	result = None
112
113	1	if stdout:
114	1	try:
115	1	result = json.loads(stdout)
116		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
117		log.warn('Invalid output returned %r - %s', stdout, ex, exc_info=True)
118
119		# Build result
120	1	if process.returncode != 0:
121		# Test failed
122		if result and result.get('message'):
123		if result.get('traceback'):
124		log.info('%s - %s', result['message'], result['traceback'])
125
126		raise Exception(result['message'])
127
128		raise Exception('Unknown error (code: %s)' % process.returncode)
129
130		# Test successful
131	1	return result
132
133	1	@classmethod
134		def find_python_executable(cls):
135	1	candidates = [sys.executable]
136
137		# Add candidates based on the script path in `sys.argv`
138	1	if sys.argv and len(sys.argv) > 0 and os.path.exists(sys.argv[0]):
139	1	bootstrap_path = sys.argv[0]
140	1	resources_pos = bootstrap_path.lower().find('resources')
141
142	1	if resources_pos > 0:
143		pms_path = bootstrap_path[:resources_pos]
144
145		cls._add_python_home_candidates(candidates, pms_path)
146
147		# Add candidates relative to `PLEX_MEDIA_SERVER_HOME`
148	1	pms_home = os.environ.get('PLEX_MEDIA_SERVER_HOME')
149
150	1	if pms_home and os.path.exists(pms_home):
151		cls._add_python_home_candidates(candidates, pms_home)
152
153		# Add candidates relative to `PYTHONHOME`
154	1	python_home = os.environ.get('PYTHONHOME')
155
156	1	if python_home and os.path.exists(python_home):
157		candidates.append(os.path.join(python_home, 'bin', 'python'))
158
159		# Use first candidate that exists
160	1	for path in candidates:
161	1	if os.path.exists(path):
162	1	return path
163
164		log.warn('Unable to find python executable', extra={'candidates': candidates})
165		return None
166
167	1	@staticmethod
168		def _add_python_home_candidates(candidates, path):
169		# Windows
170		candidates.append(os.path.join(path, 'PlexScriptHost.exe'))
171
172		# *nix
173		candidates.append(os.path.join(path, 'Plex Script Host'))
174		candidates.append(os.path.join(path, 'Resources', 'Plex Script Host'))
175		candidates.append(os.path.join(path, 'Resources', 'Python', 'bin', 'python'))
176
177		#
178		# Events
179		#
180
181	1	@classmethod
182		def on_failure(cls, message):
183		pass
184
185	1	@classmethod
186		def on_success(cls, metadata):
187	1	pass
188
189		#
190		# Helpers
191		#
192
193	1	@classmethod
194	1	def build_exception(cls, message, exc_info=None):
195		if exc_info is None:
196		exc_info = sys.exc_info()
197
198		return cls.build_failure(
199		message,
200		exc_info=exc_info
201		)
202
203	1	@classmethod
204		def build_failure(cls, message, **kwargs):
205		result = {
206		'success': False,
207		'message': message
208		}
209
210		# Merge extra attributes
211		merge(result, kwargs)
212
213		return result
214
215	1	@staticmethod
216		def build_success(metadata):
217	1	return {
218		'success': True,
219		'metadata': metadata
220		}
221

fuzeman / Plex-Trakt-Scrobbler

BaseTest.find_python_executable() F last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

BaseTest.find_python_executable() F
last analyzed 2018-03-05 01:11 UTC