mine.manager.WindowsManager.stop() - Code Metrics - jacebrowning/mine - Measure and Improve Code Quality continuously with Scrutinizer

mine.manager.WindowsManager.stop() A
last analyzed 2020-12-17 23:19 UTC

↳ Parent: mine.manager

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	2
dl	0
loc	2
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

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

import abc
import functools
import glob
import os
import platform
import subprocess
import time
from typing import List

import log
import psutil


# 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)
    @functools.wraps(func)
    def wrapped(self, application):
        log.debug("Determining if %s is running...", application)
        running = func(self, application)
        if running is None:
            status = "Application untracked"
        elif running:
            status = "Application running on current machine"
        else:
            status = "Application not running on current machine"
        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)
    @functools.wraps(func)
    def wrapped(self, application):
        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)
    @functools.wraps(func)
    def wrapped(self, application):
        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 = ''

    IGNORED_APPLICATION_NAMES: List[str] = []

    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")
                    break
            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",
        "slack helper.app",
        "garcon.appex",
        "musiccacheextension",
        "podcastswidget",
    ]

    @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/*",
            "/System/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):
        # pylint: disable=no-member
        log.info("starting %s...", path)
        os.startfile(path)  # type: ignore
        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 = {  # type: ignore
        WindowsManager.NAME: WindowsManager,
        MacManager.NAME: MacManager,
        LinuxManager.NAME: LinuxManager,
    }[name]()
    log.info("Current system: %s", manager)
    return manager


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

jacebrowning / mine

mine.manager.WindowsManager.stop() A last analyzed 2020-12-17 23:19 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

mine.manager.WindowsManager.stop() A
last analyzed 2020-12-17 23:19 UTC