NAppsManager.install() - Code Metrics - Inspection of "Improving NappManager to install napp dependencies" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#376)

by macartur

created 2017-04-26 15:28 UTC

NAppsManager.install() C

↳ Parent: NAppsManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	5
Bugs	2	Features	0

Metric	Value
c	5
b	2
f	0
dl	0
loc	38
rs	5.5
cc	7

"""Manage Network Application files."""
import json
import logging
import shutil
import urllib
from pathlib import Path

from kytos.core.napps import NApp

log = logging.getLogger(__name__)


class NAppsManager:
    """Deal with NApps at filesystem level and ask Kytos to (un)load NApps."""

    def __init__(self, controller):
        """Need the controller for configuration paths and (un)loading NApps.

        Args:
            controller (kytos.Controller): Controller to (un)load NApps.
        """
        self._config = controller.options

        self._controller = controller

        self._enabled = Path(self._config.napps)
        self._installed = self._enabled / '.installed'

    def enable(self, napp_uri):
        """Enable a NApp if not already enabled.

        Raises:
            FileNotFoundError: If NApp is not installed.
            PermissionError: No filesystem permission to enable NApp.
        """
        napp = NApp.create_from_uri(napp_uri)
        enabled = self._enabled / napp.username / napp.name
        installed = self._installed / napp.username / napp.name

        if not installed.is_dir():
            log.error("Failed to enable NApp %s. NApp not installed.",
                      napp.id)
        elif not enabled.exists():
            self._create_module(enabled.parent)
            try:
                # Create symlink
                enabled.symlink_to(installed)
                if self._controller is not None:
                    self._controller.load_napp(napp.username, napp.name)
                log.info("NApp enabled: %s", napp.id)
            except FileExistsError:
                pass  # OK, NApp was already enabled
            except PermissionError:
                log.error("Failed to enable NApp %s. Permission denied.",
                          napp.id)

    def disable(self, napp_uri):
        """Disable a NApp if it is enabled."""
        napp = NApp.create_from_uri(napp_uri)
        enabled = self._enabled / napp.username / napp.name
        try:
            enabled.unlink()
            log.info("NApp disabled: %s", napp.id)
            if self._controller is not None:
                self._controller.unload_napp(napp.username, napp.name)
        except FileNotFoundError:
            pass  # OK, it was already disabled

    def install(self, napp_uri, enable=True):
        """Install and enable a NApp from his repository.

        By default, install procedure will also enable the NApp. If you only
        want to install and keep NApp disabled, please use enable=False.
        """
        napp = NApp.create_from_uri(napp_uri)

        if napp in self.list():
            log.warning("Unable to install NApp %s. Already installed.", napp)
            return False

        if not napp.repository:
            options = self._controller.options
            napp.repository = json.loads(options.napps_repositories)[0]

        pkg_folder = None
        try:
            pkg_folder = napp.download()
            napp_folder = self._get_local_folder(napp, pkg_folder)
            dst = self._installed / napp.username / napp.name
            self._create_module(dst.parent)
            shutil.move(str(napp_folder), str(dst))
        finally:
            if pkg_folder and pkg_folder.exists():
                shutil.rmtree(str(pkg_folder))

        log.info("New NApp installed: %s", napp)

        napp = NApp.create_from_json(dst/'kytos.json')
        log.info('Installing Napp Dependencies:')
        for napp in napp.napp_dependencies:
            self.install(napp, enable)

        if enable:
            return self.enable(napp_uri)

        return True

    def uninstall(self, napp_uri):
        """Remove a NApp from filesystem, if existent."""
        napp = NApp.create_from_uri(napp_uri)

        if self.is_enabled(napp_uri):
            log.warning("Unable to uninstall NApp %s. NApp currently in use.",
                        napp)
            return False

        if self.is_installed(napp_uri):
            installed = self._installed / napp.username / napp.name
            if installed.is_symlink():
                installed.unlink()
            else:
                shutil.rmtree(str(installed))
            log.info("NApp uninstalled: %s", napp)
        else:
            log.warning("Unable to uninstall NApp %s. Already uninstalled.",
                        napp)
        return True

    def is_enabled(self, napp_uri):
        """Whether a NApp is enabled or not on this controller."""
        napp = NApp.create_from_uri(napp_uri)
        return napp in self.list_enabled()

    def is_installed(self, napp_uri):
        """Whether a NApp is installed or not on this controller."""
        napp = NApp.create_from_uri(napp_uri)
        return napp in self.list()

    def list(self):
        """List all NApps on this controller."""
        disabled = self.list_disabled()
        enabled = self.list_enabled()
        return enabled + disabled

    def list_enabled(self):
        """List all enabled NApps on this controller."""
        enabled = self._list_all(self._enabled)
        for napp in enabled:
            napp.enabled = True
        return enabled

    def list_disabled(self):
        """List all disabled NApps on this controller."""
        installed = set(self._list_all(self._installed))
        enabled = set(self.list_enabled())
        return list(installed - enabled)

    def search(self, pattern, use_cache=False):
        """Search for NApps in NApp repositories matching a pattern."""
        # ISSUE #347, we need to loop here over all repositories
        repo = eval(self._config.napps_repositories)[0]  # noqa

        if use_cache:
            # ISSUE #346, we should use cache here
            pass

        result = urllib.request.urlretrieve("{}/.database".format(repo))[0]
        with open(result, 'r') as fp:
            napps_json = json.load(fp)

        napps = [NApp.create_from_dict(napp_json) for napp_json in napps_json]
        return [napp for napp in napps if napp.match(pattern)]

    @staticmethod
    def _list_all(napps_dir):
        """List all NApps found in ``napps_dir``."""
        if not napps_dir.exists():
            log.warning("NApps dir (%s) doesn't exist.", napps_dir)
            return []

        jsons = napps_dir.glob('*/*/kytos.json')
        return [NApp.create_from_json(j) for j in jsons]

    @staticmethod
    def _create_module(folder):
        """Create module folder with empty __init__.py if it doesn't exist.

        Args:
            folder (pathlib.Path): Module path.
        """
        if not folder.exists():
            folder.mkdir()
            (folder / '__init__.py').touch()

    @staticmethod
    def _get_local_folder(napp, root=None):
        """Return local NApp root folder.

        Search for kytos.json in _./_ folder and _./user/napp_.

        Args:
            root (pathlib.Path): Where to begin searching.

        Raises:
            FileNotFoundError: If there is no such local NApp.

        Return:
            pathlib.Path: NApp root folder.
        """
        if root is None:
            root = Path()
        for folders in ['.'], [napp.username, napp.name]:
            kytos_json = root / Path(*folders) / 'kytos.json'
            if kytos_json.exists():
                with kytos_json.open() as f:
                    meta = json.load(f)
                    if meta['username'] == napp.username and \
                            meta['name'] == napp.name:
                        return kytos_json.parent
        raise FileNotFoundError('kytos.json not found.')


1			"""Manage Network Application files."""
2			import json
3			import logging
4			import shutil
5			import urllib
6			from pathlib import Path
7
8			from kytos.core.napps import NApp
9
10			log = logging.getLogger(__name__)
11
12
13			class NAppsManager:
14			"""Deal with NApps at filesystem level and ask Kytos to (un)load NApps."""
15
16			def __init__(self, controller):
17			"""Need the controller for configuration paths and (un)loading NApps.
18
19			Args:
20			controller (kytos.Controller): Controller to (un)load NApps.
21			"""
22			self._config = controller.options
23
24			self._controller = controller
25
26			self._enabled = Path(self._config.napps)
27			self._installed = self._enabled / '.installed'
28
29			def enable(self, napp_uri):
30			"""Enable a NApp if not already enabled.
31
32			Raises:
33			FileNotFoundError: If NApp is not installed.
34			PermissionError: No filesystem permission to enable NApp.
35			"""
36			napp = NApp.create_from_uri(napp_uri)
37			enabled = self._enabled / napp.username / napp.name
38			installed = self._installed / napp.username / napp.name
39
40			if not installed.is_dir():
41			log.error("Failed to enable NApp %s. NApp not installed.",
42			napp.id)
43			elif not enabled.exists():
44			self._create_module(enabled.parent)
45			try:
46			# Create symlink
47			enabled.symlink_to(installed)
48			if self._controller is not None:
49			self._controller.load_napp(napp.username, napp.name)
50			log.info("NApp enabled: %s", napp.id)
51			except FileExistsError:
52			pass # OK, NApp was already enabled
53			except PermissionError:
54			log.error("Failed to enable NApp %s. Permission denied.",
55			napp.id)
56
57			def disable(self, napp_uri):
58			"""Disable a NApp if it is enabled."""
59			napp = NApp.create_from_uri(napp_uri)
60			enabled = self._enabled / napp.username / napp.name
61			try:
62			enabled.unlink()
63			log.info("NApp disabled: %s", napp.id)
64			if self._controller is not None:
65			self._controller.unload_napp(napp.username, napp.name)
66			except FileNotFoundError:
67			pass # OK, it was already disabled
68
69			def install(self, napp_uri, enable=True):
70			"""Install and enable a NApp from his repository.
71
72			By default, install procedure will also enable the NApp. If you only
73			want to install and keep NApp disabled, please use enable=False.
74			"""
75			napp = NApp.create_from_uri(napp_uri)
76
77			if napp in self.list():
78			log.warning("Unable to install NApp %s. Already installed.", napp)
79			return False
80
81			if not napp.repository:
82			options = self._controller.options
83			napp.repository = json.loads(options.napps_repositories)[0]
84
85			pkg_folder = None
86			try:
87			pkg_folder = napp.download()
88			napp_folder = self._get_local_folder(napp, pkg_folder)
89			dst = self._installed / napp.username / napp.name
90			self._create_module(dst.parent)
91			shutil.move(str(napp_folder), str(dst))
92			finally:
93			if pkg_folder and pkg_folder.exists():
94			shutil.rmtree(str(pkg_folder))
95
96			log.info("New NApp installed: %s", napp)
97
98			napp = NApp.create_from_json(dst/'kytos.json')
99			log.info('Installing Napp Dependencies:')
100			for napp in napp.napp_dependencies:
101			self.install(napp, enable)
102
103			if enable:
104			return self.enable(napp_uri)
105
106			return True
107
108			def uninstall(self, napp_uri):
109			"""Remove a NApp from filesystem, if existent."""
110			napp = NApp.create_from_uri(napp_uri)
111
112			if self.is_enabled(napp_uri):
113			log.warning("Unable to uninstall NApp %s. NApp currently in use.",
114			napp)
115			return False
116
117			if self.is_installed(napp_uri):
118			installed = self._installed / napp.username / napp.name
119			if installed.is_symlink():
120			installed.unlink()
121			else:
122			shutil.rmtree(str(installed))
123			log.info("NApp uninstalled: %s", napp)
124			else:
125			log.warning("Unable to uninstall NApp %s. Already uninstalled.",
126			napp)
127			return True
128
129			def is_enabled(self, napp_uri):
130			"""Whether a NApp is enabled or not on this controller."""
131			napp = NApp.create_from_uri(napp_uri)
132			return napp in self.list_enabled()
133
134			def is_installed(self, napp_uri):
135			"""Whether a NApp is installed or not on this controller."""
136			napp = NApp.create_from_uri(napp_uri)
137			return napp in self.list()
138
139			def list(self):
140			"""List all NApps on this controller."""
141			disabled = self.list_disabled()
142			enabled = self.list_enabled()
143			return enabled + disabled
144
145			def list_enabled(self):
146			"""List all enabled NApps on this controller."""
147			enabled = self._list_all(self._enabled)
148			for napp in enabled:
149			napp.enabled = True
150			return enabled
151
152			def list_disabled(self):
153			"""List all disabled NApps on this controller."""
154			installed = set(self._list_all(self._installed))
155			enabled = set(self.list_enabled())
156			return list(installed - enabled)
157
158			def search(self, pattern, use_cache=False):
159			"""Search for NApps in NApp repositories matching a pattern."""
160			# ISSUE #347, we need to loop here over all repositories
161			repo = eval(self._config.napps_repositories)[0] # noqa
162
163			if use_cache:
164			# ISSUE #346, we should use cache here
165			pass
166
167			result = urllib.request.urlretrieve("{}/.database".format(repo))[0]
168			with open(result, 'r') as fp:
169			napps_json = json.load(fp)
170
171			napps = [NApp.create_from_dict(napp_json) for napp_json in napps_json]
172			return [napp for napp in napps if napp.match(pattern)]
173
174			@staticmethod
175			def _list_all(napps_dir):
176			"""List all NApps found in ``napps_dir``."""
177			if not napps_dir.exists():
178			log.warning("NApps dir (%s) doesn't exist.", napps_dir)
179			return []
180
181			jsons = napps_dir.glob('//kytos.json')
182			return [NApp.create_from_json(j) for j in jsons]
183
184			@staticmethod
185			def _create_module(folder):
186			"""Create module folder with empty __init__.py if it doesn't exist.
187
188			Args:
189			folder (pathlib.Path): Module path.
190			"""
191			if not folder.exists():
192			folder.mkdir()
193			(folder / '__init__.py').touch()
194
195			@staticmethod
196			def _get_local_folder(napp, root=None):
197			"""Return local NApp root folder.
198
199			Search for kytos.json in _./_ folder and _./user/napp_.
200
201			Args:
202			root (pathlib.Path): Where to begin searching.
203
204			Raises:
205			FileNotFoundError: If there is no such local NApp.
206
207			Return:
208			pathlib.Path: NApp root folder.
209			"""
210			if root is None:
211			root = Path()
212			for folders in ['.'], [napp.username, napp.name]:
213			kytos_json = root / Path(*folders) / 'kytos.json'
214			if kytos_json.exists():
215			with kytos_json.open() as f:
216			meta = json.load(f)
217			if meta['username'] == napp.username and \
218			meta['name'] == napp.name:
219			return kytos_json.parent
220			raise FileNotFoundError('kytos.json not found.')
221

kytos / kytos

Pull Request — master (#376)

NAppsManager.install() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like