build.flow.FlowBase.__init__() - Code Metrics - Inspection of "Create tags decorator to tests" - kytos/of_core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#69)

by Gleyberson

created 2020-04-16 17:46 UTC

build.flow.FlowBase.init() A

↳ Parent: build.flow

Complexity

Conditions

Size

Total Lines	28
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

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

How to fix Many Parameters

"""High-level abstraction for Flows of multiple OpenFlow versions.

Use common fields of FlowStats/FlowMod of supported OF versions. ``match`` and
``actions`` fields are different, so Flow, Action and Match related classes are
inherited in v0x01 and v0x04 modules.
"""
import json
from abc import ABC, abstractmethod
from hashlib import md5

# Note: FlowModCommand is the same in both v0x01 and v0x04
from pyof.v0x04.controller2switch.flow_mod import FlowModCommand

from napps.kytos.of_core import v0x01, v0x04


class FlowFactory(ABC):  # pylint: disable=too-few-public-methods
    """Choose the correct Flow according to OpenFlow version."""

    @classmethod
    def from_of_flow_stats(cls, of_flow_stats, switch):
        """Return a Flow for the switch OpenFlow version."""
        flow_class = cls.get_class(switch)
        return flow_class.from_of_flow_stats(of_flow_stats, switch)

    @staticmethod
    def get_class(switch):
        """Return the Flow class for the switch OF version."""
        of_version = switch.connection.protocol.version
        if of_version == 0x01:
            return v0x01.flow.Flow
        if of_version == 0x04:
            return v0x04.flow.Flow
        raise NotImplementedError(f'Unsupported OpenFlow version {of_version}')


class FlowBase(ABC):  # pylint: disable=too-many-instance-attributes
    """Class to abstract a Flow to switches.

    This class represents a Flow installed or to be installed inside the
    switch. A flow, in this case is represented by a Match object and a set of
    actions that should occur in case any match happen.
    """

    # of_version number: 0x01, 0x04
    of_version = None

    # Subclasses must set their version-specific classes
    _action_factory = None
    _flow_mod_class = None
    _match_class = None

    def __init__(self, switch, table_id=0xff, match=None, priority=0x8000,
                 idle_timeout=0, hard_timeout=0, cookie=0, actions=None,
                 stats=None):
        """Assign parameters to attributes.

        Args:
            switch (kytos.core.switch.Switch): Switch ID is used to uniquely
                identify a flow.
            table_id (int): The index of a single table or 0xff for all tables.
            match (|match|): Match object.
            priority (int): Priority level of flow entry.
            idle_timeout (int): Idle time before discarding, in seconds.
            hard_timeout (int): Max time before discarding, in seconds.
            cookie (int): Opaque controller-issued identifier.
            actions (|list_of_actions|): List of actions to apply.
            stats (Stats): Latest flow statistics.
        """
        # pylint: disable=too-many-arguments,too-many-locals
        self.switch = switch
        self.table_id = table_id
        # Disable not-callable error as subclasses set a class
        self.match = match or self._match_class()  # pylint: disable=E1102
        self.priority = priority
        self.idle_timeout = idle_timeout
        self.hard_timeout = hard_timeout
        self.cookie = cookie
        self.actions = actions or []
        self.stats = stats or FlowStats()  # pylint: disable=E1102

    @property
    def id(self):  # pylint: disable=invalid-name
        """Return this flow unique identifier.

        Calculate an md5 hash based on this object's modified json string. The
        json for ID calculation excludes ``stats`` attribute that changes over
        time.

        Returns:
            str: Flow unique identifier (md5sum).

        """
        flow_str = self.as_json(sort_keys=True, include_id=False)
        md5sum = md5()
        md5sum.update(flow_str.encode('utf-8'))
        return md5sum.hexdigest()

    def as_dict(self, include_id=True):
        """Return the Flow as a serializable Python dictionary.

        Args:
            include_id (bool): Default is ``True``. Internally, it is set to
                ``False`` when calculating the flow ID that is based in this
                dictionary's JSON string.

        Returns:
            dict: Serializable dictionary.

        """
        flow_dict = {
            'switch': self.switch.id,
            'table_id': self.table_id,
            'match': self.match.as_dict(),
            'priority': self.priority,
            'idle_timeout': self.idle_timeout,
            'hard_timeout': self.hard_timeout,
            'cookie': self.cookie,
            'actions': [action.as_dict() for action in self.actions]}
        if include_id:
            # Avoid infinite recursion
            flow_dict['id'] = self.id
            # Remove statistics that change over time
            flow_dict['stats'] = self.stats.as_dict()

        return flow_dict

    @classmethod
    def from_dict(cls, flow_dict, switch):
        """Return an instance with values from ``flow_dict``."""
        flow = cls(switch)

        # Set attributes found in ``flow_dict``
        for attr_name, attr_value in flow_dict.items():
            if attr_name in vars(flow):
                setattr(flow, attr_name, attr_value)

        flow.switch = switch
        if 'stats' in flow_dict:
            flow.stats = FlowStats.from_dict(flow_dict['stats'])

        # Version-specific attributes
        if 'match' in flow_dict:
            flow.match = cls._match_class.from_dict(flow_dict['match'])
        if 'actions' in flow_dict:
            flow.actions = []
            for action_dict in flow_dict['actions']:
                action = cls._action_factory.from_dict(action_dict)
                if action:
                    flow.actions.append(action)

        return flow

    def as_json(self, sort_keys=False, include_id=True):
        """Return the representation of a flow in JSON format.

        Args:
            sort_keys (bool): ``False`` by default (Python's default). Sorting
                is used, for example, to calculate the flow ID.
            include_id (bool): ``True`` by default. Internally, the ID is not
                included while calculating it.

        Returns:
            string: Flow JSON string representation.

        """
        return json.dumps(self.as_dict(include_id), sort_keys=sort_keys)

    def as_of_add_flow_mod(self):
        """Return an OpenFlow add FlowMod."""
        return self._as_of_flow_mod(FlowModCommand.OFPFC_ADD)

    def as_of_delete_flow_mod(self):
        """Return an OpenFlow delete FlowMod."""
        return self._as_of_flow_mod(FlowModCommand.OFPFC_DELETE)

    @abstractmethod
    def _as_of_flow_mod(self, command):
        """Return a pyof FlowMod with given ``command``."""
        # Disable not-callable error as subclasses will set a class
        flow_mod = self._flow_mod_class()  # pylint: disable=E1102
        flow_mod.match = self.match.as_of_match()
        flow_mod.cookie = self.cookie
        flow_mod.command = command
        flow_mod.idle_timeout = self.idle_timeout
        flow_mod.hard_timeout = self.hard_timeout
        flow_mod.priority = self.priority
        return flow_mod

    @staticmethod
    @abstractmethod
    def _get_of_actions(of_flow_stats):
        """Return pyof actions from pyof FlowStats."""

    @classmethod
    def from_of_flow_stats(cls, of_flow_stats, switch):
        """Create a flow with latest stats based on pyof FlowStats."""
        of_actions = cls._get_of_actions(of_flow_stats)
        actions = (cls._action_factory.from_of_action(of_action)

                   for of_action in of_actions)
        non_none_actions = [action for action in actions if action]
        return cls(switch,
                   table_id=of_flow_stats.table_id.value,
                   match=cls._match_class.from_of_match(of_flow_stats.match),
                   priority=of_flow_stats.priority.value,
                   idle_timeout=of_flow_stats.idle_timeout.value,
                   hard_timeout=of_flow_stats.hard_timeout.value,
                   cookie=of_flow_stats.cookie.value,
                   actions=non_none_actions,
                   stats=FlowStats.from_of_flow_stats(of_flow_stats))


class ActionBase(ABC):
    """Base class for a flow action."""

    def as_dict(self):
        """Return a dict that can be dumped as JSON."""
        return vars(self)

    @classmethod
    def from_dict(cls, action_dict):
        """Return an action instance from attributes in a dictionary."""
        action = cls(None)
        for attr_name, value in action_dict.items():
            if hasattr(action, attr_name):
                setattr(action, attr_name, value)
        return action

    @abstractmethod
    def as_of_action(self):
        """Return a pyof action to be used by a FlowMod."""

    @classmethod
    @abstractmethod
    def from_of_action(cls, of_action):
        """Return an action from a pyof action."""


class ActionFactoryBase(ABC):
    """Deal with different implementations of ActionBase."""

    # key: action_type or pyof class, value: ActionBase child
    _action_class = {
        'output': None,
        'set_vlan': None,
        # pyof class: ActionBase child
    }

    @classmethod
    def from_dict(cls, action_dict):
        """Build the proper Action from a dictionary.

        Args:
            action_dict (dict): Action attributes.
        """
        action_type = action_dict.get('action_type')
        action_class = cls._action_class[action_type]
        return action_class.from_dict(action_dict) if action_class else None

    @classmethod
    def from_of_action(cls, of_action):
        """Build the proper Action from a pyof action.

        Args:
            of_action (pyof action): Action from python-openflow.
        """
        of_class = type(of_action)
        action_class = cls._action_class.get(of_class)
        return action_class.from_of_action(of_action) if action_class else None


class MatchBase:  # pylint: disable=too-many-instance-attributes
    """Base class with common high-level Match fields."""

    def __init__(self, in_port=None, dl_src=None, dl_dst=None, dl_vlan=None,
                 dl_vlan_pcp=None, dl_type=None, nw_proto=None, nw_src=None,
                 nw_dst=None, tp_src=None, tp_dst=None):
        """Make it possible to set all attributes from the constructor."""
        # pylint: disable=too-many-arguments
        self.in_port = in_port
        self.dl_src = dl_src
        self.dl_dst = dl_dst
        self.dl_vlan = dl_vlan
        self.dl_vlan_pcp = dl_vlan_pcp
        self.dl_type = dl_type
        self.nw_proto = nw_proto
        self.nw_src = nw_src
        self.nw_dst = nw_dst
        self.tp_src = tp_src
        self.tp_dst = tp_dst

    def as_dict(self):
        """Return a dictionary excluding ``None`` values."""
        return {k: v for k, v in self.__dict__.items() if v is not None}

    @classmethod
    def from_dict(cls, match_dict):
        """Return a Match instance from a dictionary."""
        match = cls()
        for key, value in match_dict.items():
            if key in match.__dict__:
                setattr(match, key, value)
        return match

    @classmethod
    @abstractmethod
    def from_of_match(cls, of_match):
        """Return a Match instance from a pyof Match."""

    @abstractmethod
    def as_of_match(self):
        """Return a python-openflow Match."""


class Stats:
    """Simple class to store statistics as attributes and values."""

    def as_dict(self):
        """Return a dict excluding attributes with ``None`` value."""
        return {attribute: value
                for attribute, value in vars(self).items()
                if value is not None}

    @classmethod
    def from_dict(cls, stats_dict):
        """Return a statistics object from a dictionary."""
        stats = cls()
        cls._update(stats, stats_dict.items())
        return stats

    @classmethod
    def from_of_flow_stats(cls, of_stats):
        """Create an instance from a pyof FlowStats."""
        stats = cls()
        stats.update(of_stats)
        return stats

    def update(self, of_stats):
        """Given a pyof stats object, update attributes' values.

        Avoid object creation and memory leak. pyof values are GenericType
        instances whose native values can be accessed by `.value`.
        """
        # Generator for GenericType values
        attr_name_value = ((attr_name, gen_type.value)

                           for attr_name, gen_type in vars(of_stats).items()
                           if attr_name in vars(self))
        self._update(self, attr_name_value)

    @staticmethod
    def _update(obj, iterable):
        """From attribute name and value pairs, update ``obj``."""
        for attr_name, value in iterable:
            if hasattr(obj, attr_name):
                setattr(obj, attr_name, value)


class FlowStats(Stats):
    """Common fields for 1.0 and 1.3 FlowStats."""

    def __init__(self):
        """Initialize all statistics as ``None``."""
        self.byte_count = None
        self.duration_sec = None
        self.duration_nsec = None
        self.packet_count = None


class PortStats(Stats):  # pylint: disable=too-many-instance-attributes
    """Common fields for 1.0 and 1.3 PortStats."""

    def __init__(self):
        """Initialize all statistics as ``None``."""
        self.rx_packets = None
        self.tx_packets = None
        self.rx_bytes = None
        self.tx_bytes = None
        self.rx_dropped = None
        self.tx_dropped = None
        self.rx_errors = None
        self.tx_errors = None
        self.rx_frame_err = None
        self.rx_over_err = None
        self.rx_crc_err = None
        self.collisions = None


1			"""High-level abstraction for Flows of multiple OpenFlow versions.
2
3			Use common fields of FlowStats/FlowMod of supported OF versions. ``match`` and
4			``actions`` fields are different, so Flow, Action and Match related classes are
5			inherited in v0x01 and v0x04 modules.
6			"""
7			import json
8			from abc import ABC, abstractmethod
9			from hashlib import md5
10
11			# Note: FlowModCommand is the same in both v0x01 and v0x04
12			from pyof.v0x04.controller2switch.flow_mod import FlowModCommand
13
14			from napps.kytos.of_core import v0x01, v0x04
15
16
17			class FlowFactory(ABC): # pylint: disable=too-few-public-methods
18			"""Choose the correct Flow according to OpenFlow version."""
19
20			@classmethod
21			def from_of_flow_stats(cls, of_flow_stats, switch):
22			"""Return a Flow for the switch OpenFlow version."""
23			flow_class = cls.get_class(switch)
24			return flow_class.from_of_flow_stats(of_flow_stats, switch)
25
26			@staticmethod
27			def get_class(switch):
28			"""Return the Flow class for the switch OF version."""
29			of_version = switch.connection.protocol.version
30			if of_version == 0x01:
31			return v0x01.flow.Flow
32			if of_version == 0x04:
33			return v0x04.flow.Flow
34			raise NotImplementedError(f'Unsupported OpenFlow version {of_version}')
35
36
37			class FlowBase(ABC): # pylint: disable=too-many-instance-attributes
38			"""Class to abstract a Flow to switches.
39
40			This class represents a Flow installed or to be installed inside the
41			switch. A flow, in this case is represented by a Match object and a set of
42			actions that should occur in case any match happen.
43			"""
44
45			# of_version number: 0x01, 0x04
46			of_version = None
47
48			# Subclasses must set their version-specific classes
49			_action_factory = None
50			_flow_mod_class = None
51			_match_class = None
52
53			def __init__(self, switch, table_id=0xff, match=None, priority=0x8000,
54			idle_timeout=0, hard_timeout=0, cookie=0, actions=None,
55			stats=None):
56			"""Assign parameters to attributes.
57
58			Args:
59			switch (kytos.core.switch.Switch): Switch ID is used to uniquely
60			identify a flow.
61			table_id (int): The index of a single table or 0xff for all tables.
62			match (\|match\|): Match object.
63			priority (int): Priority level of flow entry.
64			idle_timeout (int): Idle time before discarding, in seconds.
65			hard_timeout (int): Max time before discarding, in seconds.
66			cookie (int): Opaque controller-issued identifier.
67			actions (\|list_of_actions\|): List of actions to apply.
68			stats (Stats): Latest flow statistics.
69			"""
70			# pylint: disable=too-many-arguments,too-many-locals
71			self.switch = switch
72			self.table_id = table_id
73			# Disable not-callable error as subclasses set a class
74			self.match = match or self._match_class() # pylint: disable=E1102
75			self.priority = priority
76			self.idle_timeout = idle_timeout
77			self.hard_timeout = hard_timeout
78			self.cookie = cookie
79			self.actions = actions or []
80			self.stats = stats or FlowStats() # pylint: disable=E1102
81
82			@property
83			def id(self): # pylint: disable=invalid-name
84			"""Return this flow unique identifier.
85
86			Calculate an md5 hash based on this object's modified json string. The
87			json for ID calculation excludes ``stats`` attribute that changes over
88			time.
89
90			Returns:
91			str: Flow unique identifier (md5sum).
92
93			"""
94			flow_str = self.as_json(sort_keys=True, include_id=False)
95			md5sum = md5()
96			md5sum.update(flow_str.encode('utf-8'))
97			return md5sum.hexdigest()
98
99			def as_dict(self, include_id=True):
100			"""Return the Flow as a serializable Python dictionary.
101
102			Args:
103			include_id (bool): Default is ``True``. Internally, it is set to
104			``False`` when calculating the flow ID that is based in this
105			dictionary's JSON string.
106
107			Returns:
108			dict: Serializable dictionary.
109
110			"""
111			flow_dict = {
112			'switch': self.switch.id,
113			'table_id': self.table_id,
114			'match': self.match.as_dict(),
115			'priority': self.priority,
116			'idle_timeout': self.idle_timeout,
117			'hard_timeout': self.hard_timeout,
118			'cookie': self.cookie,
119			'actions': [action.as_dict() for action in self.actions]}
120			if include_id:
121			# Avoid infinite recursion
122			flow_dict['id'] = self.id
123			# Remove statistics that change over time
124			flow_dict['stats'] = self.stats.as_dict()
125
126			return flow_dict
127
128			@classmethod
129			def from_dict(cls, flow_dict, switch):
130			"""Return an instance with values from ``flow_dict``."""
131			flow = cls(switch)
132
133			# Set attributes found in ``flow_dict``
134			for attr_name, attr_value in flow_dict.items():
135			if attr_name in vars(flow):
136			setattr(flow, attr_name, attr_value)
137
138			flow.switch = switch
139			if 'stats' in flow_dict:
140			flow.stats = FlowStats.from_dict(flow_dict['stats'])
141
142			# Version-specific attributes
143			if 'match' in flow_dict:
144			flow.match = cls._match_class.from_dict(flow_dict['match'])
145			if 'actions' in flow_dict:
146			flow.actions = []
147			for action_dict in flow_dict['actions']:
148			action = cls._action_factory.from_dict(action_dict)
149			if action:
150			flow.actions.append(action)
151
152			return flow
153
154			def as_json(self, sort_keys=False, include_id=True):
155			"""Return the representation of a flow in JSON format.
156
157			Args:
158			sort_keys (bool): ``False`` by default (Python's default). Sorting
159			is used, for example, to calculate the flow ID.
160			include_id (bool): ``True`` by default. Internally, the ID is not
161			included while calculating it.
162
163			Returns:
164			string: Flow JSON string representation.
165
166			"""
167			return json.dumps(self.as_dict(include_id), sort_keys=sort_keys)
168
169			def as_of_add_flow_mod(self):
170			"""Return an OpenFlow add FlowMod."""
171			return self._as_of_flow_mod(FlowModCommand.OFPFC_ADD)
172
173			def as_of_delete_flow_mod(self):
174			"""Return an OpenFlow delete FlowMod."""
175			return self._as_of_flow_mod(FlowModCommand.OFPFC_DELETE)
176
177			@abstractmethod
178			def _as_of_flow_mod(self, command):
179			"""Return a pyof FlowMod with given ``command``."""
180			# Disable not-callable error as subclasses will set a class
181			flow_mod = self._flow_mod_class() # pylint: disable=E1102
182			flow_mod.match = self.match.as_of_match()
183			flow_mod.cookie = self.cookie
184			flow_mod.command = command
185			flow_mod.idle_timeout = self.idle_timeout
186			flow_mod.hard_timeout = self.hard_timeout
187			flow_mod.priority = self.priority
188			return flow_mod
189
190			@staticmethod
191			@abstractmethod
192			def _get_of_actions(of_flow_stats):
193			"""Return pyof actions from pyof FlowStats."""
194
195			@classmethod
196			def from_of_flow_stats(cls, of_flow_stats, switch):
197			"""Create a flow with latest stats based on pyof FlowStats."""
198			of_actions = cls._get_of_actions(of_flow_stats)
199			actions = (cls._action_factory.from_of_action(of_action)
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-02-18 13:53 UTC by Report Bug Copy Issue Report The variable `of_action` does not seem to be defined. Loading history...
200			for of_action in of_actions)
201			non_none_actions = [action for action in actions if action]
202			return cls(switch,
203			table_id=of_flow_stats.table_id.value,
204			match=cls._match_class.from_of_match(of_flow_stats.match),
205			priority=of_flow_stats.priority.value,
206			idle_timeout=of_flow_stats.idle_timeout.value,
207			hard_timeout=of_flow_stats.hard_timeout.value,
208			cookie=of_flow_stats.cookie.value,
209			actions=non_none_actions,
210			stats=FlowStats.from_of_flow_stats(of_flow_stats))
211
212
213			class ActionBase(ABC):
214			"""Base class for a flow action."""
215
216			def as_dict(self):
217			"""Return a dict that can be dumped as JSON."""
218			return vars(self)
219
220			@classmethod
221			def from_dict(cls, action_dict):
222			"""Return an action instance from attributes in a dictionary."""
223			action = cls(None)
224			for attr_name, value in action_dict.items():
225			if hasattr(action, attr_name):
226			setattr(action, attr_name, value)
227			return action
228
229			@abstractmethod
230			def as_of_action(self):
231			"""Return a pyof action to be used by a FlowMod."""
232
233			@classmethod
234			@abstractmethod
235			def from_of_action(cls, of_action):
236			"""Return an action from a pyof action."""
237
238
239			class ActionFactoryBase(ABC):
240			"""Deal with different implementations of ActionBase."""
241
242			# key: action_type or pyof class, value: ActionBase child
243			_action_class = {
244			'output': None,
245			'set_vlan': None,
246			# pyof class: ActionBase child
247			}
248
249			@classmethod
250			def from_dict(cls, action_dict):
251			"""Build the proper Action from a dictionary.
252
253			Args:
254			action_dict (dict): Action attributes.
255			"""
256			action_type = action_dict.get('action_type')
257			action_class = cls._action_class[action_type]
258			return action_class.from_dict(action_dict) if action_class else None
259
260			@classmethod
261			def from_of_action(cls, of_action):
262			"""Build the proper Action from a pyof action.
263
264			Args:
265			of_action (pyof action): Action from python-openflow.
266			"""
267			of_class = type(of_action)
268			action_class = cls._action_class.get(of_class)
269			return action_class.from_of_action(of_action) if action_class else None
270
271
272			class MatchBase: # pylint: disable=too-many-instance-attributes
273			"""Base class with common high-level Match fields."""
274
275			def __init__(self, in_port=None, dl_src=None, dl_dst=None, dl_vlan=None,
276			dl_vlan_pcp=None, dl_type=None, nw_proto=None, nw_src=None,
277			nw_dst=None, tp_src=None, tp_dst=None):
278			"""Make it possible to set all attributes from the constructor."""
279			# pylint: disable=too-many-arguments
280			self.in_port = in_port
281			self.dl_src = dl_src
282			self.dl_dst = dl_dst
283			self.dl_vlan = dl_vlan
284			self.dl_vlan_pcp = dl_vlan_pcp
285			self.dl_type = dl_type
286			self.nw_proto = nw_proto
287			self.nw_src = nw_src
288			self.nw_dst = nw_dst
289			self.tp_src = tp_src
290			self.tp_dst = tp_dst
291
292			def as_dict(self):
293			"""Return a dictionary excluding ``None`` values."""
294			return {k: v for k, v in self.__dict__.items() if v is not None}
295
296			@classmethod
297			def from_dict(cls, match_dict):
298			"""Return a Match instance from a dictionary."""
299			match = cls()
300			for key, value in match_dict.items():
301			if key in match.__dict__:
302			setattr(match, key, value)
303			return match
304
305			@classmethod
306			@abstractmethod
307			def from_of_match(cls, of_match):
308			"""Return a Match instance from a pyof Match."""
309
310			@abstractmethod
311			def as_of_match(self):
312			"""Return a python-openflow Match."""
313
314
315			class Stats:
316			"""Simple class to store statistics as attributes and values."""
317
318			def as_dict(self):
319			"""Return a dict excluding attributes with ``None`` value."""
320			return {attribute: value
321			for attribute, value in vars(self).items()
322			if value is not None}
323
324			@classmethod
325			def from_dict(cls, stats_dict):
326			"""Return a statistics object from a dictionary."""
327			stats = cls()
328			cls._update(stats, stats_dict.items())
329			return stats
330
331			@classmethod
332			def from_of_flow_stats(cls, of_stats):
333			"""Create an instance from a pyof FlowStats."""
334			stats = cls()
335			stats.update(of_stats)
336			return stats
337
338			def update(self, of_stats):
339			"""Given a pyof stats object, update attributes' values.
340
341			Avoid object creation and memory leak. pyof values are GenericType
342			instances whose native values can be accessed by `.value`.
343			"""
344			# Generator for GenericType values
345			attr_name_value = ((attr_name, gen_type.value)
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-02-18 13:53 UTC by Report Bug Copy Issue Report The variable `gen_type` does not seem to be defined. Loading history... Comprehensibility Best Practice introduced 2019-02-18 13:53 UTC by Report Bug Copy Issue Report The variable `attr_name` does not seem to be defined. Loading history...
346			for attr_name, gen_type in vars(of_stats).items()
347			if attr_name in vars(self))
348			self._update(self, attr_name_value)
349
350			@staticmethod
351			def _update(obj, iterable):
352			"""From attribute name and value pairs, update ``obj``."""
353			for attr_name, value in iterable:
354			if hasattr(obj, attr_name):
355			setattr(obj, attr_name, value)
356
357
358			class FlowStats(Stats):
359			"""Common fields for 1.0 and 1.3 FlowStats."""
360
361			def __init__(self):
362			"""Initialize all statistics as ``None``."""
363			self.byte_count = None
364			self.duration_sec = None
365			self.duration_nsec = None
366			self.packet_count = None
367
368
369			class PortStats(Stats): # pylint: disable=too-many-instance-attributes
370			"""Common fields for 1.0 and 1.3 PortStats."""
371
372			def __init__(self):
373			"""Initialize all statistics as ``None``."""
374			self.rx_packets = None
375			self.tx_packets = None
376			self.rx_bytes = None
377			self.tx_bytes = None
378			self.rx_dropped = None
379			self.tx_dropped = None
380			self.rx_errors = None
381			self.tx_errors = None
382			self.rx_frame_err = None
383			self.rx_over_err = None
384			self.rx_crc_err = None
385			self.collisions = None
386

kytos / of_core

Pull Request — master (#69)

build.flow.FlowBase.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Many Parameters

Many Parameters

Duplication Side-by-Side

Filter issues like

build.flow.FlowBase.init() A