kytos.core.napps.base.NApp.download() - Code Metrics - Inspection of "[Feature][Fix] Upgrade dependencies" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#284)

by Vinicius

created 2022-09-08 21:06 UTC

kytos.core.napps.base.NApp.download() A

↳ Parent: kytos.core.napps.base

Complexity

Conditions

Size

Total Lines	18
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	2

Importance

Changes

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

"""Kytos Napps Module."""
import json
import os
import re
import sys
import tarfile
import traceback
import urllib
from abc import ABCMeta, abstractmethod
from pathlib import Path
from random import randint
from threading import Event, Thread

from kytos.core.events import KytosEvent
from kytos.core.logs import NAppLog

__all__ = ('KytosNApp',)

LOG = NAppLog()


class NApp:
    """Class to represent a NApp."""

    # pylint: disable=too-many-arguments
    def __init__(self, username=None, name=None, version=None,
                 repository=None, meta=False):
        self.username = username
        self.name = name
        self.version = version if version else 'latest'
        self.repository = repository
        self.meta = meta
        self.description = None
        self.tags = []
        self.enabled = False
        self.napp_dependencies = []

    def __str__(self):
        return f"{self.username}/{self.name}"

    def __repr__(self):
        return f"NApp({self.username}/{self.name})"

    def __hash__(self):
        return hash(self.id)

    def __eq__(self, other):
        """Compare username/name strings."""
        return isinstance(other, self.__class__) and self.id == other.id

    @property
    def id(self):  # pylint: disable=invalid-name
        """username/name string."""
        return str(self)

    @property
    def uri(self):
        """Return a unique identifier of this NApp."""
        version = self.version if self.version else 'latest'

        if not self._has_valid_repository():
            return ""

        # Use the next line after Diraol fix redirect using ":" for version
        # return "{}/{}:{}".format(self.repository, self.id, version)

        return f"{self.repository}/{self.id}-{version}"

    @property
    def package_url(self):
        """Return a fully qualified URL for a NApp package."""
        if not self.uri:
            return ""
        return f"{self.uri}.napp"

    @classmethod
    def create_from_uri(cls, uri):
        """Return a new NApp instance from an unique identifier."""
        regex = r'^(((https?://|file://)(.+))/)?(.+?)/(.+?)/?(:(.+))?$'
        match = re.match(regex, uri)

        if not match:
            return None

        return cls(username=match.groups()[4],
                   name=match.groups()[5],
                   version=match.groups()[7],
                   repository=match.groups()[1])

    @classmethod
    def create_from_json(cls, filename):
        """Return a new NApp instance from a metadata file."""
        with open(filename, encoding='utf-8') as data_file:
            data = json.loads(data_file.read())

        return cls.create_from_dict(data)

    @classmethod
    def create_from_dict(cls, data):
        """Return a new NApp instance from metadata."""
        napp = cls()

        for attribute, value in data.items():
            setattr(napp, attribute, value)

        return napp

    def as_json(self):
        """Dump all NApp attributes on a json format."""
        return json.dumps(self.__dict__)

    def match(self, pattern):
        """Whether a pattern is present on NApp id, description and tags."""
        try:
            pattern = f'.*{pattern}.*'
            pattern = re.compile(pattern, re.IGNORECASE)
            strings = [self.id, self.description] + self.tags
            return any(pattern.match(string) for string in strings)
        except TypeError:
            return False

    def download(self):
        """Download NApp package from his repository.

        Raises:
            urllib.error.HTTPError: If download is not successful.

        Returns:
            str: Downloaded temp filename.

        """
        if not self.package_url:
            return None

        package_filename = urllib.request.urlretrieve(self.package_url)[0]
        extracted = self._extract(package_filename)
        Path(package_filename).unlink()
        self._update_repo_file(extracted)
        return extracted

    @staticmethod
    def _extract(filename):
        """Extract NApp package to a temporary folder.

        Return:
            pathlib.Path: Temp dir with package contents.
        """
        random_string = str(randint(0, 10**6))
        tmp = '/tmp/kytos-napp-' + Path(filename).stem + '-' + random_string
        os.mkdir(tmp)
        with tarfile.open(filename, 'r:xz') as tar:
            tar.extractall(tmp)
        return Path(tmp)

    def _has_valid_repository(self):
        """Whether this NApp has a valid repository or not."""
        return all([self.username, self.name, self.repository])

    def _update_repo_file(self, destination=None):
        """Create or update the file '.repo' inside NApp package."""
        with open(f"{destination}/.repo", 'w', encoding="utf8") as repo_file:
            repo_file.write(self.repository + '\n')


class KytosNApp(Thread, metaclass=ABCMeta):
    """Base class for any KytosNApp to be developed."""

    def __init__(self, controller, **kwargs):
        """Contructor of KytosNapps.

        Go through all of the instance methods and selects those that have
        the events attribute, then creates a dict containing the event_name
        and the list of methods that are responsible for handling such event.

        At the end, the setup method is called as a complement of the init
        process.
        """
        Thread.__init__(self, daemon=False)
        self.controller = controller
        self.username = None  # loaded from json
        self.name = None      # loaded from json
        self.meta = False     # loaded from json
        self._load_json()

        # Force a listener with a private method.
        self._listeners = {
            'kytos/core.shutdown': [self._shutdown_handler],
            'kytos/core.shutdown.' + self.napp_id: [self._shutdown_handler]}

        self.__event = Event()
        #: int: Seconds to sleep before next call to :meth:`execute`. If
        #: negative, run :meth:`execute` only once.
        self.__interval = -1
        self.setup()

        #: Add non-private methods that listen to events.
        handler_methods = [getattr(self, method_name) for method_name in
                           dir(self) if method_name[0] != '_' and
                           callable(getattr(self, method_name)) and
                           hasattr(getattr(self, method_name), 'events')]

        # Building the listeners dictionary
        for method in handler_methods:
            for event_name in method.events:
                if event_name not in self._listeners:
                    self._listeners[event_name] = []
                self._listeners[event_name].append(method)

    @property
    def napp_id(self):
        """username/name string."""
        return f"{self.username}/{self.name}"

    def listeners(self):
        """Return all listeners registered."""
        return list(self._listeners.keys())

    def _load_json(self):
        """Update object attributes based on kytos.json."""
        current_file = sys.modules[self.__class__.__module__].__file__
        json_path = os.path.join(os.path.dirname(current_file), 'kytos.json')
        with open(json_path, encoding='utf-8') as data_file:
            data = json.loads(data_file.read())

        for attribute, value in data.items():
            setattr(self, attribute, value)

    def execute_as_loop(self, interval):
        """Run :meth:`execute` within a loop. Call this method during setup.

        By calling this method, the application does not need to worry about
        loop details such as sleeping and stopping the loop when
        :meth:`shutdown` is called. Just call this method during :meth:`setup`
        and implement :meth:`execute` as a single execution.

        Args:
            interval (int): Seconds between each call to :meth:`execute`.
        """
        self.__interval = interval

    # pylint: disable=broad-except
    def run(self):
        """Call the execute method, looping as needed.

        It should not be overriden.
        """
        self.notify_loaded()
        LOG.info(f"Running NApp: {self}")
        try:
            self.execute()
        except Exception:
            traceback_str = traceback.format_exc(chain=False)
            LOG.error(f"NApp: {self} unhandled exception {traceback_str}")
        while self.__interval > 0 and not self.__event.is_set():
            self.__event.wait(self.__interval)
            try:
                self.execute()
            except Exception:
                traceback_str = traceback.format_exc(chain=False)
                LOG.error(f"NApp: {self} unhandled exception {traceback_str}")

    def notify_loaded(self):
        """Inform this NApp has been loaded."""
        name = f'{self.username}/{self.name}.loaded'
        event = KytosEvent(name=name, content={})
        self.controller.buffers.app.put(event)

    # all listeners receive event
    def _shutdown_handler(self, event):  # pylint: disable=unused-argument
        """Listen shutdown event from kytos.

        This method listens the kytos/core.shutdown event and call the shutdown
        method from napp subclass implementation.

        Paramters
            event (:class:`KytosEvent`): event to be listened.
        """
        if not self.__event.is_set():
            self.__event.set()
            self.shutdown()

    @abstractmethod
    def setup(self):
        """Replace the 'init' method for the KytosApp subclass.

        The setup method is automatically called on the NApp __init__().
        Users aren't supposed to call this method directly.
        """

    @abstractmethod
    def execute(self):
        """Execute in a loop until 'kytos/core.shutdown' is received.

        The execute method is called by KytosNApp class.
        Users shouldn't call this method directly.
        """

    @abstractmethod
    def shutdown(self):
        """Run before the app is unloaded and the controller, stopped.

        The user implementation of this method should do the necessary routine
        for the user App and also it is a good moment to break the loop of the
        execute method if it is in a loop.

        This methods is not going to be called directly, it is going to be
        called by the _shutdown_handler method when a KytosShutdownEvent is
        sent.
        """


1		"""Kytos Napps Module."""
2	1	import json
3	1	import os
4	1	import re
5	1	import sys
6	1	import tarfile
7	1	import traceback
8	1	import urllib
9	1	from abc import ABCMeta, abstractmethod
10	1	from pathlib import Path
11	1	from random import randint
12	1	from threading import Event, Thread
13
14	1	from kytos.core.events import KytosEvent
15	1	from kytos.core.logs import NAppLog
16
17	1	__all__ = ('KytosNApp',)
18
19	1	LOG = NAppLog()
20
21
22	1	class NApp:
23		"""Class to represent a NApp."""
24
25		# pylint: disable=too-many-arguments
26	1	def __init__(self, username=None, name=None, version=None,
27		repository=None, meta=False):
28	1	self.username = username
29	1	self.name = name
30	1	self.version = version if version else 'latest'
31	1	self.repository = repository
32	1	self.meta = meta
33	1	self.description = None
34	1	self.tags = []
35	1	self.enabled = False
36	1	self.napp_dependencies = []
37
38	1	def __str__(self):
39	1	return f"{self.username}/{self.name}"
40
41	1	def __repr__(self):
42	1	return f"NApp({self.username}/{self.name})"
43
44	1	def __hash__(self):
45		return hash(self.id)
46
47	1	def __eq__(self, other):
48		"""Compare username/name strings."""
49		return isinstance(other, self.__class__) and self.id == other.id
50
51	1	@property
52	1	def id(self): # pylint: disable=invalid-name
53		"""username/name string."""
54	1	return str(self)
55
56	1	@property
57	1	def uri(self):
58		"""Return a unique identifier of this NApp."""
59	1	version = self.version if self.version else 'latest'
60
61	1	if not self._has_valid_repository():
62	1	return ""
63
64		# Use the next line after Diraol fix redirect using ":" for version
65		# return "{}/{}:{}".format(self.repository, self.id, version)
66
67	1	return f"{self.repository}/{self.id}-{version}"
68
69	1	@property
70	1	def package_url(self):
71		"""Return a fully qualified URL for a NApp package."""
72	1	if not self.uri:
73	1	return ""
74	1	return f"{self.uri}.napp"
75
76	1	@classmethod
77	1	def create_from_uri(cls, uri):
78		"""Return a new NApp instance from an unique identifier."""
79	1	regex = r'^(((https?://\|file://)(.+))/)?(.+?)/(.+?)/?(:(.+))?$'
80	1	match = re.match(regex, uri)
81
82	1	if not match:
83	1	return None
84
85	1	return cls(username=match.groups()[4],
86		name=match.groups()[5],
87		version=match.groups()[7],
88		repository=match.groups()[1])
89
90	1	@classmethod
91	1	def create_from_json(cls, filename):
92		"""Return a new NApp instance from a metadata file."""
93	1	with open(filename, encoding='utf-8') as data_file:
94	1	data = json.loads(data_file.read())
95
96	1	return cls.create_from_dict(data)
97
98	1	@classmethod
99	1	def create_from_dict(cls, data):
100		"""Return a new NApp instance from metadata."""
101	1	napp = cls()
102
103	1	for attribute, value in data.items():
104	1	setattr(napp, attribute, value)
105
106	1	return napp
107
108	1	def as_json(self):
109		"""Dump all NApp attributes on a json format."""
110		return json.dumps(self.__dict__)
111
112	1	def match(self, pattern):
113		"""Whether a pattern is present on NApp id, description and tags."""
114	1	try:
115	1	pattern = f'.{pattern}.'
116	1	pattern = re.compile(pattern, re.IGNORECASE)
117	1	strings = [self.id, self.description] + self.tags
118	1	return any(pattern.match(string) for string in strings)
119		except TypeError:
120		return False
121
122	1	def download(self):
123		"""Download NApp package from his repository.
124
125		Raises:
126		urllib.error.HTTPError: If download is not successful.
127
128		Returns:
129		str: Downloaded temp filename.
130
131		"""
132	1	if not self.package_url:
133	1	return None
134
135	1	package_filename = urllib.request.urlretrieve(self.package_url)[0]
136	1	extracted = self._extract(package_filename)
137	1	Path(package_filename).unlink()
138	1	self._update_repo_file(extracted)
139	1	return extracted
140
141	1	@staticmethod
142	1	def _extract(filename):
143		"""Extract NApp package to a temporary folder.
144
145		Return:
146		pathlib.Path: Temp dir with package contents.
147		"""
148	1	random_string = str(randint(0, 10**6))
149	1	tmp = '/tmp/kytos-napp-' + Path(filename).stem + '-' + random_string
150	1	os.mkdir(tmp)
151	1	with tarfile.open(filename, 'r:xz') as tar:
152	1	tar.extractall(tmp)
153	1	return Path(tmp)
154
155	1	def _has_valid_repository(self):
156		"""Whether this NApp has a valid repository or not."""
157	1	return all([self.username, self.name, self.repository])
158
159	1	def _update_repo_file(self, destination=None):
160		"""Create or update the file '.repo' inside NApp package."""
161	1	with open(f"{destination}/.repo", 'w', encoding="utf8") as repo_file:
162	1	repo_file.write(self.repository + '\n')
163
164
165	1	class KytosNApp(Thread, metaclass=ABCMeta):
166		"""Base class for any KytosNApp to be developed."""
167
168	1	def __init__(self, controller, **kwargs):
169		"""Contructor of KytosNapps.
170
171		Go through all of the instance methods and selects those that have
172		the events attribute, then creates a dict containing the event_name
173		and the list of methods that are responsible for handling such event.
174
175		At the end, the setup method is called as a complement of the init
176		process.
177		"""
178	1	Thread.__init__(self, daemon=False)
179	1	self.controller = controller
180	1	self.username = None # loaded from json
181	1	self.name = None # loaded from json
182	1	self.meta = False # loaded from json
183	1	self._load_json()
184
185		# Force a listener with a private method.
186	1	self._listeners = {
187		'kytos/core.shutdown': [self._shutdown_handler],
188		'kytos/core.shutdown.' + self.napp_id: [self._shutdown_handler]}
189
190	1	self.__event = Event()
191		#: int: Seconds to sleep before next call to :meth:`execute`. If
192		#: negative, run :meth:`execute` only once.
193	1	self.__interval = -1
194	1	self.setup()
195
196		#: Add non-private methods that listen to events.
197	1	handler_methods = [getattr(self, method_name) for method_name in
198		dir(self) if method_name[0] != '_' and
199		callable(getattr(self, method_name)) and
200		hasattr(getattr(self, method_name), 'events')]
201
202		# Building the listeners dictionary
203	1	for method in handler_methods:
204	1	for event_name in method.events:
205		if event_name not in self._listeners:
206		self._listeners[event_name] = []
207		self._listeners[event_name].append(method)
208
209	1	@property
210	1	def napp_id(self):
211		"""username/name string."""
212	1	return f"{self.username}/{self.name}"
213
214	1	def listeners(self):
215		"""Return all listeners registered."""
216		return list(self._listeners.keys())
217
218	1	def _load_json(self):
219		"""Update object attributes based on kytos.json."""
220	1	current_file = sys.modules[self.__class__.__module__].__file__
221	1	json_path = os.path.join(os.path.dirname(current_file), 'kytos.json')
222	1	with open(json_path, encoding='utf-8') as data_file:
223	1	data = json.loads(data_file.read())
224
225	1	for attribute, value in data.items():
226	1	setattr(self, attribute, value)
227
228	1	def execute_as_loop(self, interval):
229		"""Run :meth:`execute` within a loop. Call this method during setup.
230
231		By calling this method, the application does not need to worry about
232		loop details such as sleeping and stopping the loop when
233		:meth:`shutdown` is called. Just call this method during :meth:`setup`
234		and implement :meth:`execute` as a single execution.
235
236		Args:
237		interval (int): Seconds between each call to :meth:`execute`.
238		"""
239	1	self.__interval = interval
240
241		# pylint: disable=broad-except
242	1	def run(self):
243		"""Call the execute method, looping as needed.
244
245		It should not be overriden.
246		"""
247	1	self.notify_loaded()
248	1	LOG.info(f"Running NApp: {self}")
249	1	try:
250	1	self.execute()
251		except Exception:
252		traceback_str = traceback.format_exc(chain=False)
253		LOG.error(f"NApp: {self} unhandled exception {traceback_str}")
254	1	while self.__interval > 0 and not self.__event.is_set():
255	1	self.__event.wait(self.__interval)
256	1	try:
257	1	self.execute()
258		except Exception:
259		traceback_str = traceback.format_exc(chain=False)
260		LOG.error(f"NApp: {self} unhandled exception {traceback_str}")
261
262	1	def notify_loaded(self):
263		"""Inform this NApp has been loaded."""
264	1	name = f'{self.username}/{self.name}.loaded'
265	1	event = KytosEvent(name=name, content={})
266	1	self.controller.buffers.app.put(event)
267
268		# all listeners receive event
269	1	def _shutdown_handler(self, event): # pylint: disable=unused-argument
270		"""Listen shutdown event from kytos.
271
272		This method listens the kytos/core.shutdown event and call the shutdown
273		method from napp subclass implementation.
274
275		Paramters
276		event (:class:`KytosEvent`): event to be listened.
277		"""
278	1	if not self.__event.is_set():
279	1	self.__event.set()
280	1	self.shutdown()
281
282	1	@abstractmethod
283	1	def setup(self):
284		"""Replace the 'init' method for the KytosApp subclass.
285
286		The setup method is automatically called on the NApp __init__().
287		Users aren't supposed to call this method directly.
288		"""
289
290	1	@abstractmethod
291	1	def execute(self):
292		"""Execute in a loop until 'kytos/core.shutdown' is received.
293
294		The execute method is called by KytosNApp class.
295		Users shouldn't call this method directly.
296		"""
297
298	1	@abstractmethod
299	1	def shutdown(self):
300		"""Run before the app is unloaded and the controller, stopped.
301
302		The user implementation of this method should do the necessary routine
303		for the user App and also it is a good moment to break the loop of the
304		execute method if it is in a loop.
305
306		This methods is not going to be called directly, it is going to be
307		called by the _shutdown_handler method when a KytosShutdownEvent is
308		sent.
309		"""
310

kytos-ng / kytos

Pull Request — master (#284)

kytos.core.napps.base.NApp.download() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like