BaseManager._get_process() - Code Metrics - Inspection of "Release v1.3" - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#78)

by Jace

created 2017-03-13 13:54 UTC

BaseManager._get_process() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	110

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	10
dl	0
loc	33
ccs	0
cts	0
cp	0
crap	110
rs	3.1304
c	2
b	0
f	0

How to fix Complexity

"""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


# 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

    IGNORED_APPLICATION_NAMES = [
    ]

    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

    @classmethod
    def _get_process(cls, 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)])
            except psutil.AccessDenied:
                continue  # the process is likely owned by root

            if name.lower() not in parts:
                continue

            if process.pid == os.getpid():
                log.debug("Skipped current process: %s", command)
                continue

            if process.status() == psutil.STATUS_ZOMBIE:
                log.debug("Skipped zombie process: %s", command)
                continue

            log.debug("Found matching process: %s", command)
            for ignored in cls.IGNORED_APPLICATION_NAMES:
                if ignored.lower() in parts:
                    log.debug("But skipped due to ignored name")
                else:
                    return process

        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'

    IGNORED_APPLICATION_NAMES = [
        "iTunesHelper.app",
    ]

    @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()

    @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(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


Pull Request — master (#78)

BaseManager._get_process() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Complexity

Complexity

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
21
22		# TODO: enable coverage when a Linux test is implemented
23		def log_running(func): # pragma: no cover (manual)
24		"""Decorator for methods that return application status."""
25		@functools.wraps(func)
26		def wrapped(self, application):
27		"""Wrapped method to log if an application is running."""
28		log.debug("Determining if %s is running...", application)
29		running = func(self, application)
30		if running is None:
31		status = "Untracked"
32		elif running:
33		status = "Running"
34		else:
35		status = "Not running"
36		log.info("%s: %s", status, application)
37		return running
38		return wrapped
39
40
41		# TODO: enable coverage when a Linux test is implemented
42		def log_starting(func): # pragma: no cover (manual)
43		"""Decorator for methods that start an application."""
44		@functools.wraps(func)
45		def wrapped(self, application):
46		"""Wrapped method to log that an application is being started."""
47		log.info("Starting %s...", application)
48		result = func(self, application)
49		log.info("Running: %s", application)
50		return result
51		return wrapped
52
53
54		# TODO: enable coverage when a Linux test is implemented
55		def log_stopping(func): # pragma: no cover (manual)
56		"""Decorator for methods that stop an application."""
57		@functools.wraps(func)
58		def wrapped(self, application):
59		"""Wrapped method to log that an application is being stopped."""
60		log.info("Stopping %s...", application)
61		result = func(self, application)
62		log.info("Not running: %s", application)
63		return result
64		return wrapped
65
66
67		class BaseManager(metaclass=abc.ABCMeta): # pragma: no cover (abstract)
68		"""Base application manager."""
69
70		NAME = FRIENDLY = None
71
72		IGNORED_APPLICATION_NAMES = [
73		]
74
75		def __str__(self):
76		return self.FRIENDLY
77
78		@abc.abstractmethod
79		def is_running(self, application):
80		"""Determine if an application is currently running."""
81		raise NotImplementedError
82
83		@abc.abstractmethod
84		def start(self, application):
85		"""Start an application on the current computer."""
86		raise NotImplementedError
87
88		@abc.abstractmethod
89		def stop(self, application):
90		"""Stop an application on the current computer."""
91		raise NotImplementedError
92
93		@abc.abstractmethod
94		def launch(self, path):
95		"""Open a file for editing."""
96		raise NotImplementedError
97
98		@classmethod
99		def _get_process(cls, name):
100		"""Get a process whose executable path contains an app name."""
101		log.debug("Searching for exe path containing '%s'...", name)
102
103		for process in psutil.process_iter():
104		try:
105		command = ' '.join(process.cmdline()).lower()
106		parts = []
107		for arg in process.cmdline():
108		parts.extend([p.lower() for p in arg.split(os.sep)])
109		except psutil.AccessDenied:
110		continue # the process is likely owned by root
111
112		if name.lower() not in parts:
113		continue
114
115		if process.pid == os.getpid():
116		log.debug("Skipped current process: %s", command)
117		continue
118
119		if process.status() == psutil.STATUS_ZOMBIE:
120		log.debug("Skipped zombie process: %s", command)
121		continue
122
123		log.debug("Found matching process: %s", command)
124		for ignored in cls.IGNORED_APPLICATION_NAMES:
125		if ignored.lower() in parts:
126		log.debug("But skipped due to ignored name")
127		else:
128		return process
129
130		return None
131
132
133		class LinuxManager(BaseManager): # pragma: no cover (manual)
134		"""Application manager for Linux."""
135
136		NAME = 'Linux'
137		FRIENDLY = NAME
138
139		def is_running(self, application):
140		name = application.versions.linux
141		if not name:
142		return None
143		process = self._get_process(name)
144		return process is not None
145
146		def start(self, application):
147		pass
148
149		def stop(self, application):
150		name = application.versions.linux
151		process = self._get_process(name)
152		if process.is_running():
153		process.terminate()
154
155		def launch(self, path):
156		log.info("Opening %s...", path)
157		return subprocess.call(['xdg-open', path]) == 0
158
159
160		class MacManager(BaseManager): # pragma: no cover (manual)
161		"""Application manager for OS X."""
162
163		NAME = 'Darwin'
164		FRIENDLY = 'Mac'
165
166		IGNORED_APPLICATION_NAMES = [
167		"iTunesHelper.app",
168		]
169
170		@log_running
171		def is_running(self, application):
172		name = application.versions.mac
173		if not name:
174		return None
175		process = self._get_process(name)
176		return process is not None
177
178		@log_starting
179		def start(self, application):
180		name = application.versions.mac
181		path = None
182		for base in (".",
183		"~/Applications",
184		"/Applications",
185		"/Applications/*"):
186		pattern = os.path.expanduser(os.path.join(base, name))
187		log.debug("Glob pattern: %s", pattern)
188		paths = glob.glob(pattern)
189		if paths:
190		path = paths[0]
191		log.debug("Match: %s", path)
192		break
193		else:
194		assert path, "Not found: {}".format(application)
195		return self._start_app(path)
196
197		@log_stopping
198		def stop(self, application):
199		name = application.versions.mac
200		process = self._get_process(name)
201		if process and process.is_running():
202		process.terminate()
203
204		@staticmethod
205		def _start_app(path):
206		"""Start an application from it's .app directory."""
207		assert os.path.exists(path), path
208		process = psutil.Popen(['open', path])
209		time.sleep(1)
210		return process
211
212		def launch(self, path):
213		log.info("opening %s...", path)
214		return subprocess.call(['open', path]) == 0
215
216
217		class WindowsManager(BaseManager): # pragma: no cover (manual)
218		"""Application manager for Windows."""
219
220		NAME = 'Windows'
221		FRIENDLY = NAME
222
223	1	def is_running(self, application):
224		pass
225	1
226	1	def start(self, application):
227	1	pass
228
229		def stop(self, application):
230		pass
231
232	1	def launch(self, path):
233	1	log.info("starting %s...", path)
234		os.startfile(path) # pylint: disable=no-member
235		return True
236
237
238		def get_manager(name=None):
239		"""Return an application manager for the current operating system."""
240		log.info("Detecting the current system...")
241		name = name or platform.system()
242		manager = {
243		WindowsManager.NAME: WindowsManager,
244		MacManager.NAME: MacManager,
245		LinuxManager.NAME: LinuxManager,
246		}[name]()
247		log.info("Current system: %s", manager)
248		return manager
249

jacebrowning / mine

Pull Request — master (#78)

BaseManager._get_process() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Complexity

Complexity

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files