ospd.vts.Vts.items() - Code Metrics - Inspection of "Use an iterator to get the vts when get_vts cmd is..." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#216)

by Juan José

created 2020-03-04 11:14 UTC

ospd.vts.Vts.items() A

↳ Parent: ospd.vts

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
rs	10
c	0
b	0
f	0

# Copyright (C) 2020 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-2.0-or-later
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

""" Classes for storing VTs
"""

import multiprocessing
import re

from copy import deepcopy
from typing import (
    Dict,
    Any,
    Type,
    Iterator,
    Iterable,
    Tuple,
)

from ospd.errors import OspdError

DEFAULT_VT_ID_PATTERN = re.compile("[0-9a-zA-Z_\\-:.]{1,80}")


class Vts:
    def __init__(
        self, storage: Type[Dict] = None, vt_id_pattern=DEFAULT_VT_ID_PATTERN,
    ):
        self.storage = storage

        self.vt_id_pattern = vt_id_pattern
        self._vts = None

    def __contains__(self, key: str) -> bool:
        return key in self._vts

    def __iter__(self) -> Iterator[str]:
        if hasattr(self.vts, '__iter__'):
            return self.vts.__iter__()

        # Use iter because python3.5 has no support for
        # iteration over DictProxy.
        return iter(self.vts.keys())

    def __getitem__(self, key):
        return self.vts[key]

    def items(self) -> Iterator[Tuple[str, Dict]]:
        return iter(self.vts.items())

    def __len__(self) -> int:
        return len(self.vts)

    def __init_vts(self):
        if self.storage:
            self._vts = self.storage()
        else:
            self._vts = multiprocessing.Manager().dict()

    @property
    def vts(self) -> Dict[str, Any]:
        if self._vts is None:
            self.__init_vts()

        return self._vts

    def add(
        self,
        vt_id: str,
        name: str = None,
        vt_params: str = None,
        vt_refs: str = None,
        custom: str = None,
        vt_creation_time: str = None,
        vt_modification_time: str = None,
        vt_dependencies: str = None,
        summary: str = None,
        impact: str = None,
        affected: str = None,
        insight: str = None,
        solution: str = None,
        solution_t: str = None,
        solution_m: str = None,
        detection: str = None,
        qod_t: str = None,
        qod_v: str = None,
        severities: str = None,
    ) -> None:
        """ Add a vulnerability test information.

        IMPORTANT: The VT's Data Manager will store the vts collection.
        If the collection is considerably big and it will be consultated
        intensible during a routine, consider to do a deepcopy(), since
        accessing the shared memory in the data manager is very expensive.
        At the end of the routine, the temporal copy must be set to None
        and deleted.
        """
        if not vt_id:
            raise OspdError('Invalid vt_id {}'.format(vt_id))

        if self.vt_id_pattern.fullmatch(vt_id) is None:
            raise OspdError('Invalid vt_id {}'.format(vt_id))

        if vt_id in self.vts:
            raise OspdError('vt_id {} already exists'.format(vt_id))

        if name is None:
            name = ''

        vt = {'name': name}
        if custom is not None:
            vt["custom"] = custom
        if vt_params is not None:
            vt["vt_params"] = vt_params
        if vt_refs is not None:
            vt["vt_refs"] = vt_refs
        if vt_dependencies is not None:
            vt["vt_dependencies"] = vt_dependencies
        if vt_creation_time is not None:
            vt["creation_time"] = vt_creation_time
        if vt_modification_time is not None:
            vt["modification_time"] = vt_modification_time
        if summary is not None:
            vt["summary"] = summary
        if impact is not None:
            vt["impact"] = impact
        if affected is not None:
            vt["affected"] = affected
        if insight is not None:
            vt["insight"] = insight

        if solution is not None:
            vt["solution"] = solution
            if solution_t is not None:
                vt["solution_type"] = solution_t
            if solution_m is not None:
                vt["solution_method"] = solution_m

        if detection is not None:
            vt["detection"] = detection

        if qod_t is not None:
            vt["qod_type"] = qod_t
        elif qod_v is not None:
            vt["qod"] = qod_v

        if severities is not None:
            vt["severities"] = severities

        self.vts[vt_id] = vt

    def get(self, vt_id: str) -> Dict[str, Any]:
        return self.vts.get(vt_id)

    def keys(self) -> Iterable[str]:
        return self.vts.keys()

    def clear(self) -> None:
        self._vts.clear()
        self._vts = None

    def copy(self) -> "Vts":
        copy = Vts(self.storage, vt_id_pattern=self.vt_id_pattern)
        copy._vts = deepcopy(self._vts)  # pylint: disable=protected-access
        return copy


1			# Copyright (C) 2020 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: GPL-2.0-or-later
4			#
5			# This program is free software; you can redistribute it and/or
6			# modify it under the terms of the GNU General Public License
7			# as published by the Free Software Foundation; either version 2
8			# of the License, or (at your option) any later version.
9			#
10			# This program is distributed in the hope that it will be useful,
11			# but WITHOUT ANY WARRANTY; without even the implied warranty of
12			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			# GNU General Public License for more details.
14			#
15			# You should have received a copy of the GNU General Public License
16			# along with this program; if not, write to the Free Software
17			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18
19			""" Classes for storing VTs
20			"""
21
22			import multiprocessing
23			import re
24
25			from copy import deepcopy
26			from typing import (
27			Dict,
28			Any,
29			Type,
30			Iterator,
31			Iterable,
32			Tuple,
33			)
34
35			from ospd.errors import OspdError
36
37			DEFAULT_VT_ID_PATTERN = re.compile("[0-9a-zA-Z_\\-:.]{1,80}")
38
39
40			class Vts:
41			def __init__(
42			self, storage: Type[Dict] = None, vt_id_pattern=DEFAULT_VT_ID_PATTERN,
43			):
44			self.storage = storage
45
46			self.vt_id_pattern = vt_id_pattern
47			self._vts = None
48
49			def __contains__(self, key: str) -> bool:
50			return key in self._vts
51
52			def __iter__(self) -> Iterator[str]:
53			if hasattr(self.vts, '__iter__'):
54			return self.vts.__iter__()
55
56			# Use iter because python3.5 has no support for
57			# iteration over DictProxy.
58			return iter(self.vts.keys())
59
60			def __getitem__(self, key):
61			return self.vts[key]
62
63			def items(self) -> Iterator[Tuple[str, Dict]]:
64			return iter(self.vts.items())
65
66			def __len__(self) -> int:
67			return len(self.vts)
68
69			def __init_vts(self):
70			if self.storage:
71			self._vts = self.storage()
72			else:
73			self._vts = multiprocessing.Manager().dict()
74
75			@property
76			def vts(self) -> Dict[str, Any]:
77			if self._vts is None:
78			self.__init_vts()
79
80			return self._vts
81
82			def add(
83			self,
84			vt_id: str,
85			name: str = None,
86			vt_params: str = None,
87			vt_refs: str = None,
88			custom: str = None,
89			vt_creation_time: str = None,
90			vt_modification_time: str = None,
91			vt_dependencies: str = None,
92			summary: str = None,
93			impact: str = None,
94			affected: str = None,
95			insight: str = None,
96			solution: str = None,
97			solution_t: str = None,
98			solution_m: str = None,
99			detection: str = None,
100			qod_t: str = None,
101			qod_v: str = None,
102			severities: str = None,
103			) -> None:
104			""" Add a vulnerability test information.
105
106			IMPORTANT: The VT's Data Manager will store the vts collection.
107			If the collection is considerably big and it will be consultated
108			intensible during a routine, consider to do a deepcopy(), since
109			accessing the shared memory in the data manager is very expensive.
110			At the end of the routine, the temporal copy must be set to None
111			and deleted.
112			"""
113			if not vt_id:
114			raise OspdError('Invalid vt_id {}'.format(vt_id))
115
116			if self.vt_id_pattern.fullmatch(vt_id) is None:
117			raise OspdError('Invalid vt_id {}'.format(vt_id))
118
119			if vt_id in self.vts:
120			raise OspdError('vt_id {} already exists'.format(vt_id))
121
122			if name is None:
123			name = ''
124
125			vt = {'name': name}
126			if custom is not None:
127			vt["custom"] = custom
128			if vt_params is not None:
129			vt["vt_params"] = vt_params
130			if vt_refs is not None:
131			vt["vt_refs"] = vt_refs
132			if vt_dependencies is not None:
133			vt["vt_dependencies"] = vt_dependencies
134			if vt_creation_time is not None:
135			vt["creation_time"] = vt_creation_time
136			if vt_modification_time is not None:
137			vt["modification_time"] = vt_modification_time
138			if summary is not None:
139			vt["summary"] = summary
140			if impact is not None:
141			vt["impact"] = impact
142			if affected is not None:
143			vt["affected"] = affected
144			if insight is not None:
145			vt["insight"] = insight
146
147			if solution is not None:
148			vt["solution"] = solution
149			if solution_t is not None:
150			vt["solution_type"] = solution_t
151			if solution_m is not None:
152			vt["solution_method"] = solution_m
153
154			if detection is not None:
155			vt["detection"] = detection
156
157			if qod_t is not None:
158			vt["qod_type"] = qod_t
159			elif qod_v is not None:
160			vt["qod"] = qod_v
161
162			if severities is not None:
163			vt["severities"] = severities
164
165			self.vts[vt_id] = vt
166
167			def get(self, vt_id: str) -> Dict[str, Any]:
168			return self.vts.get(vt_id)
169
170			def keys(self) -> Iterable[str]:
171			return self.vts.keys()
172
173			def clear(self) -> None:
174			self._vts.clear()
175			self._vts = None
176
177			def copy(self) -> "Vts":
178			copy = Vts(self.storage, vt_id_pattern=self.vt_id_pattern)
179			copy._vts = deepcopy(self._vts) # pylint: disable=protected-access
180			return copy
181

greenbone / ospd

Pull Request — master (#216)

ospd.vts.Vts.items() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like