BaseTest.spawn() - Code Metrics - Inspection of "Workaround for the `MainMenu() takes no arguments..." - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 09a48b...79b39d )

by Dean

created 2016-09-10 03:28 UTC

BaseTest.spawn() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	15
CRAP Score	16.4

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
dl	0
loc	60
ccs	15
cts	25
cp	0.6
rs	4.8
c	1
b	0
f	0
cc	10
crap	16.4

How to fix Long Method 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, 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, ex:
53		if success:
54		continue
55
56		message = ex.message
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:
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

Push — develop ( 09a48b...79b39d )

BaseTest.spawn() D

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like