BaseManager._get_process() - Code Metrics - Inspection of "Merge pull request #58 from jacebrowning/release/v..." - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f46766...a9fd17 )

by Jace

created 2016-05-26 18:36 UTC

BaseManager._get_process() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	72

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	8
c	1
b	0
f	0
dl	0
loc	25
ccs	0
cts	0
cp	0
crap	72
rs	4

"""Classes to manage application state."""

import os
import abc
import time
import glob
import platform
import functools
import subprocess
import logging

import psutil
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3


log = logging.getLogger(__name__)

# TODO: delete this after implementing `BaseManager`
# https://github.com/jacebrowning/mine/issues/8
# https://github.com/jacebrowning/mine/issues/9
# pylint: disable=R0903,W0223,E0110


# TODO: enable coverage when a Linux test is implemented
def log_running(func):  # pragma: no cover (manual)
    """Decorator for methods that return application status."""
    @functools.wraps(func)
    def wrapped(self, application):
        """Wrapped method to log if an application is running."""
        log.debug("Determining if %s is running...", application)
        running = func(self, application)
        if running is None:
            status = "Untracked"
        elif running:
            status = "Running"
        else:
            status = "Not running"
        log.info("%s: %s", status, application)
        return running
    return wrapped


# TODO: enable coverage when a Linux test is implemented
def log_starting(func):  # pragma: no cover (manual)
    """Decorator for methods that start an application."""
    @functools.wraps(func)
    def wrapped(self, application):
        """Wrapped method to log that an application is being started."""
        log.info("Starting %s...", application)
        result = func(self, application)
        log.info("Running: %s", application)
        return result
    return wrapped


# TODO: enable coverage when a Linux test is implemented
def log_stopping(func):  # pragma: no cover (manual)
    """Decorator for methods that stop an application."""
    @functools.wraps(func)
    def wrapped(self, application):
        """Wrapped method to log that an application is being stopped."""
        log.info("Stopping %s...", application)
        result = func(self, application)
        log.info("Not running: %s", application)
        return result
    return wrapped


class BaseManager(metaclass=abc.ABCMeta):  # pragma: no cover (abstract)
    """Base application manager."""

    NAME = FRIENDLY = None

    def __str__(self):
        return self.FRIENDLY

    @abc.abstractmethod
    def is_running(self, application):
        """Determine if an application is currently running."""
        raise NotImplementedError

    @abc.abstractmethod
    def start(self, application):
        """Start an application on the current computer."""
        raise NotImplementedError

    @abc.abstractmethod
    def stop(self, application):
        """Stop an application on the current computer."""
        raise NotImplementedError

    @abc.abstractmethod
    def launch(self, path):
        """Open a file for editing."""
        raise NotImplementedError

    @staticmethod
    def _get_process(name):
        """Get a process whose executable path contains an app name."""
        log.debug("Searching for exe path containing '%s'...", name)

        for process in psutil.process_iter():
            try:
                command = ' '.join(process.cmdline()).lower()
                parts = []
                for arg in process.cmdline():
                    parts.extend([p.lower() for p in arg.split(os.sep)])

                if name.lower() in parts:
                    if process.pid == os.getpid():
                        log.debug("Skipped current process: %s", command)
                    elif process.status() == psutil.STATUS_ZOMBIE:
                        log.debug("Skipped zombie process: %s", command)
                    else:
                        log.debug("Found matching process: %s", command)
                        return process

            except psutil.AccessDenied:
                pass  # the process is likely owned by root

        return None


class LinuxManager(BaseManager):  # pragma: no cover (manual)
    """Application manager for Linux."""

    NAME = 'Linux'
    FRIENDLY = NAME

    def is_running(self, application):
        name = application.versions.linux
        if not name:
            return None
        process = self._get_process(name)
        return process is not None

    def start(self, application):
        pass

    def stop(self, application):
        name = application.versions.linux
        process = self._get_process(name)
        if process.is_running():
            process.terminate()

    def launch(self, path):
        log.info("Opening %s...", path)
        return subprocess.call(['xdg-open', path]) == 0


class MacManager(BaseManager):  # pragma: no cover (manual)
    """Application manager for OS X."""

    NAME = 'Darwin'
    FRIENDLY = 'Mac'

    @log_running
    def is_running(self, application):
        name = application.versions.mac
        if not name:
            return None
        process = self._get_process(name)
        return process is not None

    @log_starting
    def start(self, application):
        name = application.versions.mac
        path = None
        for base in (".",
                     "~/Applications",
                     "/Applications",
                     "/Applications/*"):
            pattern = os.path.expanduser(os.path.join(base, name))
            log.debug("Glob pattern: %s", pattern)
            paths = glob.glob(pattern)
            if paths:
                path = paths[0]
                log.debug("Match: %s", path)
                break
        else:
            assert path, "Not found: {}".format(application)
        return self._start_app(path)

    @log_stopping
    def stop(self, application):
        name = application.versions.mac
        process = self._get_process(name)
        if process and process.is_running():
            process.terminate()
            time.sleep(0.1)

    @staticmethod
    def _start_app(path):
        """Start an application from it's .app directory."""
        assert os.path.exists(path), path
        process = psutil.Popen(['open', path])
        time.sleep(0.1)
        return process

    def launch(self, path):
        log.info("opening %s...", path)
        return subprocess.call(['open', path]) == 0


class WindowsManager(BaseManager):  # pragma: no cover (manual)
    """Application manager for Windows."""

    NAME = 'Windows'
    FRIENDLY = NAME

    def is_running(self, application):
        pass

    def start(self, application):
        pass

    def stop(self, application):
        pass

    def launch(self, path):
        log.info("starting %s...", path)
        os.startfile(path)  # pylint: disable=no-member
        return True


def get_manager(name=None):
    """Return an application manager for the current operating system."""
    log.info("Detecting the current system...")
    name = name or platform.system()
    manager = {
        WindowsManager.NAME: WindowsManager,
        MacManager.NAME: MacManager,
        LinuxManager.NAME: LinuxManager,
    }[name]()
    log.info("Current system: %s", manager)
    return manager


Push — master ( f46766...a9fd17 )

BaseManager._get_process() D

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing init.py files

1		"""Classes to manage application state."""
2
3	1	import os
4	1	import abc
5	1	import time
6	1	import glob
7	1	import platform
8	1	import functools
9	1	import subprocess
10	1	import logging
11
12	1	import psutil
		0 ignored issues – show Configuration introduced 2015-12-30 22:06 UTC by Report Bug Copy Issue Report The import `psutil` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
13
14
15	1	log = logging.getLogger(__name__)
16
17		# TODO: delete this after implementing `BaseManager`
18		# https://github.com/jacebrowning/mine/issues/8
19		# https://github.com/jacebrowning/mine/issues/9
20		# pylint: disable=R0903,W0223,E0110
21
22
23		# TODO: enable coverage when a Linux test is implemented
24		def log_running(func): # pragma: no cover (manual)
25		"""Decorator for methods that return application status."""
26		@functools.wraps(func)
27		def wrapped(self, application):
28		"""Wrapped method to log if an application is running."""
29		log.debug("Determining if %s is running...", application)
30		running = func(self, application)
31		if running is None:
32		status = "Untracked"
33		elif running:
34		status = "Running"
35		else:
36		status = "Not running"
37		log.info("%s: %s", status, application)
38		return running
39		return wrapped
40
41
42		# TODO: enable coverage when a Linux test is implemented
43		def log_starting(func): # pragma: no cover (manual)
44		"""Decorator for methods that start an application."""
45		@functools.wraps(func)
46		def wrapped(self, application):
47		"""Wrapped method to log that an application is being started."""
48		log.info("Starting %s...", application)
49		result = func(self, application)
50		log.info("Running: %s", application)
51		return result
52		return wrapped
53
54
55		# TODO: enable coverage when a Linux test is implemented
56		def log_stopping(func): # pragma: no cover (manual)
57		"""Decorator for methods that stop an application."""
58		@functools.wraps(func)
59		def wrapped(self, application):
60		"""Wrapped method to log that an application is being stopped."""
61		log.info("Stopping %s...", application)
62		result = func(self, application)
63		log.info("Not running: %s", application)
64		return result
65		return wrapped
66
67
68		class BaseManager(metaclass=abc.ABCMeta): # pragma: no cover (abstract)
69		"""Base application manager."""
70
71		NAME = FRIENDLY = None
72
73		def __str__(self):
74		return self.FRIENDLY
75
76		@abc.abstractmethod
77		def is_running(self, application):
78		"""Determine if an application is currently running."""
79		raise NotImplementedError
80
81		@abc.abstractmethod
82		def start(self, application):
83		"""Start an application on the current computer."""
84		raise NotImplementedError
85
86		@abc.abstractmethod
87		def stop(self, application):
88		"""Stop an application on the current computer."""
89		raise NotImplementedError
90
91		@abc.abstractmethod
92		def launch(self, path):
93		"""Open a file for editing."""
94		raise NotImplementedError
95
96		@staticmethod
97		def _get_process(name):
98		"""Get a process whose executable path contains an app name."""
99		log.debug("Searching for exe path containing '%s'...", name)
100
101		for process in psutil.process_iter():
102		try:
103		command = ' '.join(process.cmdline()).lower()
104		parts = []
105		for arg in process.cmdline():
106		parts.extend([p.lower() for p in arg.split(os.sep)])
107
108		if name.lower() in parts:
109		if process.pid == os.getpid():
110		log.debug("Skipped current process: %s", command)
111		elif process.status() == psutil.STATUS_ZOMBIE:
112		log.debug("Skipped zombie process: %s", command)
113		else:
114		log.debug("Found matching process: %s", command)
115		return process
116
117		except psutil.AccessDenied:
118		pass # the process is likely owned by root
119
120		return None
121
122
123		class LinuxManager(BaseManager): # pragma: no cover (manual)
124		"""Application manager for Linux."""
125
126		NAME = 'Linux'
127		FRIENDLY = NAME
128
129		def is_running(self, application):
130		name = application.versions.linux
131		if not name:
132		return None
133		process = self._get_process(name)
134		return process is not None
135
136		def start(self, application):
137		pass
138
139		def stop(self, application):
140		name = application.versions.linux
141		process = self._get_process(name)
142		if process.is_running():
143		process.terminate()
144
145		def launch(self, path):
146		log.info("Opening %s...", path)
147		return subprocess.call(['xdg-open', path]) == 0
148
149
150		class MacManager(BaseManager): # pragma: no cover (manual)
151		"""Application manager for OS X."""
152
153		NAME = 'Darwin'
154		FRIENDLY = 'Mac'
155
156		@log_running
157		def is_running(self, application):
158		name = application.versions.mac
159		if not name:
160		return None
161		process = self._get_process(name)
162		return process is not None
163
164		@log_starting
165		def start(self, application):
166		name = application.versions.mac
167		path = None
168		for base in (".",
169		"~/Applications",
170		"/Applications",
171		"/Applications/*"):
172		pattern = os.path.expanduser(os.path.join(base, name))
173		log.debug("Glob pattern: %s", pattern)
174		paths = glob.glob(pattern)
175		if paths:
176		path = paths[0]
177		log.debug("Match: %s", path)
178		break
179		else:
180		assert path, "Not found: {}".format(application)
181		return self._start_app(path)
182
183		@log_stopping
184		def stop(self, application):
185		name = application.versions.mac
186		process = self._get_process(name)
187		if process and process.is_running():
188		process.terminate()
189		time.sleep(0.1)
190
191		@staticmethod
192		def _start_app(path):
193		"""Start an application from it's .app directory."""
194		assert os.path.exists(path), path
195		process = psutil.Popen(['open', path])
196		time.sleep(0.1)
197		return process
198
199		def launch(self, path):
200		log.info("opening %s...", path)
201		return subprocess.call(['open', path]) == 0
202
203
204		class WindowsManager(BaseManager): # pragma: no cover (manual)
205		"""Application manager for Windows."""
206
207		NAME = 'Windows'
208		FRIENDLY = NAME
209
210		def is_running(self, application):
211		pass
212
213		def start(self, application):
214		pass
215
216		def stop(self, application):
217		pass
218
219		def launch(self, path):
220		log.info("starting %s...", path)
221		os.startfile(path) # pylint: disable=no-member
222		return True
223
224
225	1	def get_manager(name=None):
226		"""Return an application manager for the current operating system."""
227	1	log.info("Detecting the current system...")
228	1	name = name or platform.system()
229	1	manager = {
230		WindowsManager.NAME: WindowsManager,
231		MacManager.NAME: MacManager,
232		LinuxManager.NAME: LinuxManager,
233		}[name]()
234	1	log.info("Current system: %s", manager)
235		return manager
236

jacebrowning / mine

Push — master ( f46766...a9fd17 )

BaseManager._get_process() D

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files