build.flow.FlowBase.__init__() - Code Metrics - kytos/of_core - Measure and Improve Code Quality continuously with Scrutinizer

build.flow.FlowBase.init() A
last analyzed 2021-05-26 15:17 UTC

↳ Parent: build.flow

Complexity

Conditions

Size

Total Lines	29
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	1

Importance

Changes

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

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=0x0, 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
                            (0xff is valid only for the delete command)
            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.

        """
        version = self.switch.connection.protocol.version
        if version == 0x01:
            return json.dumps(self.as_dict(include_id),
                              cls=v0x01.utils.JSONEncoderOF10,
                              sort_keys=sort_keys)
        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)

    def as_of_strict_delete_flow_mod(self):
        """Return an OpenFlow strict delete FlowMod."""
        return self._as_of_flow_mod(FlowModCommand.OFPFC_DELETE_STRICT)

    @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
        flow_mod.table_id = self.table_id
        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))

    def __eq__(self, other):
        include_id = False
        if not isinstance(other, self.__class__):
            raise ValueError(f'Error comparing flows: {other} is not '
                             f'an instance of {self.__class__}')

        return self.as_dict(include_id) == other.as_dict(include_id)


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, in_phy_port=None,
                 ip_dscp=None, ip_ecn=None, udp_src=None, udp_dst=None,
                 sctp_src=None, sctp_dst=None, icmpv4_type=None,
                 icmpv4_code=None, arp_op=None, arp_spa=None, arp_tpa=None,
                 arp_sha=None, arp_tha=None, ipv6_src=None, ipv6_dst=None,
                 ipv6_flabel=None, icmpv6_type=None, icmpv6_code=None,
                 nd_tar=None, nd_sll=None, nd_tll=None, mpls_lab=None,
                 mpls_tc=None, mpls_bos=None, pbb_isid=None, v6_hdr=None,
                 metadata=None, tun_id=None):
        """Make it possible to set all attributes from the constructor."""
        # pylint: disable=too-many-arguments
        # pylint: disable=too-many-locals
        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
        self.in_phy_port = in_phy_port
        self.ip_dscp = ip_dscp
        self.ip_ecn = ip_ecn
        self.udp_src = udp_src
        self.udp_dst = udp_dst
        self.sctp_src = sctp_src
        self.sctp_dst = sctp_dst
        self.icmpv4_type = icmpv4_type
        self.icmpv4_code = icmpv4_code
        self.arp_op = arp_op
        self.arp_spa = arp_spa
        self.arp_tpa = arp_tpa
        self.arp_sha = arp_sha
        self.arp_tha = arp_tha
        self.ipv6_src = ipv6_src
        self.ipv6_dst = ipv6_dst
        self.ipv6_flabel = ipv6_flabel
        self.icmpv6_type = icmpv6_type
        self.icmpv6_code = icmpv6_code
        self.nd_tar = nd_tar
        self.nd_sll = nd_sll
        self.nd_tll = nd_tll
        self.mpls_lab = mpls_lab
        self.mpls_tc = mpls_tc
        self.mpls_bos = mpls_bos
        self.pbb_isid = pbb_isid
        self.v6_hdr = v6_hdr
        self.metadata = metadata
        self.tun_id = tun_id

    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	1	import json
8	1	from abc import ABC, abstractmethod
9	1	from hashlib import md5
10
11		# Note: FlowModCommand is the same in both v0x01 and v0x04
12	1	from pyof.v0x04.controller2switch.flow_mod import FlowModCommand
13
14	1	from napps.kytos.of_core import v0x01, v0x04
15
16
17	1	class FlowFactory(ABC): # pylint: disable=too-few-public-methods
18		"""Choose the correct Flow according to OpenFlow version."""
19
20	1	@classmethod
21		def from_of_flow_stats(cls, of_flow_stats, switch):
22		"""Return a Flow for the switch OpenFlow version."""
23	1	flow_class = cls.get_class(switch)
24	1	return flow_class.from_of_flow_stats(of_flow_stats, switch)
25
26	1	@staticmethod
27		def get_class(switch):
28		"""Return the Flow class for the switch OF version."""
29	1	of_version = switch.connection.protocol.version
30	1	if of_version == 0x01:
31	1	return v0x01.flow.Flow
32	1	if of_version == 0x04:
33	1	return v0x04.flow.Flow
34		raise NotImplementedError(f'Unsupported OpenFlow version {of_version}')
35
36
37	1	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	1	of_version = None
47
48		# Subclasses must set their version-specific classes
49	1	_action_factory = None
50	1	_flow_mod_class = None
51	1	_match_class = None
52
53	1	def __init__(self, switch, table_id=0x0, 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		(0xff is valid only for the delete command)
63		match (\|match\|): Match object.
64		priority (int): Priority level of flow entry.
65		idle_timeout (int): Idle time before discarding, in seconds.
66		hard_timeout (int): Max time before discarding, in seconds.
67		cookie (int): Opaque controller-issued identifier.
68		actions (\|list_of_actions\|): List of actions to apply.
69		stats (Stats): Latest flow statistics.
70		"""
71		# pylint: disable=too-many-arguments,too-many-locals
72	1	self.switch = switch
73	1	self.table_id = table_id
74		# Disable not-callable error as subclasses set a class
75	1	self.match = match or self._match_class() # pylint: disable=E1102
76	1	self.priority = priority
77	1	self.idle_timeout = idle_timeout
78	1	self.hard_timeout = hard_timeout
79	1	self.cookie = cookie
80	1	self.actions = actions or []
81	1	self.stats = stats or FlowStats() # pylint: disable=E1102
82
83	1	@property
84		def id(self): # pylint: disable=invalid-name
85		"""Return this flow unique identifier.
86
87		Calculate an md5 hash based on this object's modified json string. The
88		json for ID calculation excludes ``stats`` attribute that changes over
89		time.
90
91		Returns:
92		str: Flow unique identifier (md5sum).
93
94		"""
95	1	flow_str = self.as_json(sort_keys=True, include_id=False)
96	1	md5sum = md5()
97	1	md5sum.update(flow_str.encode('utf-8'))
98	1	return md5sum.hexdigest()
99
100	1	def as_dict(self, include_id=True):
101		"""Return the Flow as a serializable Python dictionary.
102
103		Args:
104		include_id (bool): Default is ``True``. Internally, it is set to
105		``False`` when calculating the flow ID that is based in this
106		dictionary's JSON string.
107
108		Returns:
109		dict: Serializable dictionary.
110
111		"""
112	1	flow_dict = {
113		'switch': self.switch.id,
114		'table_id': self.table_id,
115		'match': self.match.as_dict(),
116		'priority': self.priority,
117		'idle_timeout': self.idle_timeout,
118		'hard_timeout': self.hard_timeout,
119		'cookie': self.cookie,
120		'actions': [action.as_dict() for action in self.actions]}
121	1	if include_id:
122		# Avoid infinite recursion
123	1	flow_dict['id'] = self.id
124		# Remove statistics that change over time
125	1	flow_dict['stats'] = self.stats.as_dict()
126
127	1	return flow_dict
128
129	1	@classmethod
130		def from_dict(cls, flow_dict, switch):
131		"""Return an instance with values from ``flow_dict``."""
132	1	flow = cls(switch)
133
134		# Set attributes found in ``flow_dict``
135	1	for attr_name, attr_value in flow_dict.items():
136	1	if attr_name in vars(flow):
137	1	setattr(flow, attr_name, attr_value)
138
139	1	flow.switch = switch
140	1	if 'stats' in flow_dict:
141	1	flow.stats = FlowStats.from_dict(flow_dict['stats'])
142
143		# Version-specific attributes
144	1	if 'match' in flow_dict:
145	1	flow.match = cls._match_class.from_dict(flow_dict['match'])
146	1	if 'actions' in flow_dict:
147	1	flow.actions = []
148	1	for action_dict in flow_dict['actions']:
149	1	action = cls._action_factory.from_dict(action_dict)
150	1	if action:
151	1	flow.actions.append(action)
152
153	1	return flow
154
155	1	def as_json(self, sort_keys=False, include_id=True):
156		"""Return the representation of a flow in JSON format.
157
158		Args:
159		sort_keys (bool): ``False`` by default (Python's default). Sorting
160		is used, for example, to calculate the flow ID.
161		include_id (bool): ``True`` by default. Internally, the ID is not
162		included while calculating it.
163
164		Returns:
165		string: Flow JSON string representation.
166
167		"""
168	1	version = self.switch.connection.protocol.version
169	1	if version == 0x01:
170	1	return json.dumps(self.as_dict(include_id),
171		cls=v0x01.utils.JSONEncoderOF10,
172		sort_keys=sort_keys)
173	1	return json.dumps(self.as_dict(include_id), sort_keys=sort_keys)
174
175	1	def as_of_add_flow_mod(self):
176		"""Return an OpenFlow add FlowMod."""
177	1	return self._as_of_flow_mod(FlowModCommand.OFPFC_ADD)
178
179	1	def as_of_delete_flow_mod(self):
180		"""Return an OpenFlow delete FlowMod."""
181	1	return self._as_of_flow_mod(FlowModCommand.OFPFC_DELETE)
182
183	1	def as_of_strict_delete_flow_mod(self):
184		"""Return an OpenFlow strict delete FlowMod."""
185		return self._as_of_flow_mod(FlowModCommand.OFPFC_DELETE_STRICT)
186
187	1	@abstractmethod
188		def _as_of_flow_mod(self, command):
189		"""Return a pyof FlowMod with given ``command``."""
190		# Disable not-callable error as subclasses will set a class
191	1	flow_mod = self._flow_mod_class() # pylint: disable=E1102
192	1	flow_mod.match = self.match.as_of_match()
193	1	flow_mod.cookie = self.cookie
194	1	flow_mod.command = command
195	1	flow_mod.idle_timeout = self.idle_timeout
196	1	flow_mod.hard_timeout = self.hard_timeout
197	1	flow_mod.priority = self.priority
198	1	flow_mod.table_id = self.table_id
199	1	return flow_mod
200
201	1	@staticmethod
202	1	@abstractmethod
203		def _get_of_actions(of_flow_stats):
204		"""Return pyof actions from pyof FlowStats."""
205
206	1	@classmethod
207		def from_of_flow_stats(cls, of_flow_stats, switch):
208		"""Create a flow with latest stats based on pyof FlowStats."""
209		of_actions = cls._get_of_actions(of_flow_stats)
210		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...
211		for of_action in of_actions)
212		non_none_actions = [action for action in actions if action]
213		return cls(switch,
214		table_id=of_flow_stats.table_id.value,
215		match=cls._match_class.from_of_match(of_flow_stats.match),
216		priority=of_flow_stats.priority.value,
217		idle_timeout=of_flow_stats.idle_timeout.value,
218		hard_timeout=of_flow_stats.hard_timeout.value,
219		cookie=of_flow_stats.cookie.value,
220		actions=non_none_actions,
221		stats=FlowStats.from_of_flow_stats(of_flow_stats))
222
223	1	def __eq__(self, other):
224	1	include_id = False
225	1	if not isinstance(other, self.__class__):
226	1	raise ValueError(f'Error comparing flows: {other} is not '
227		f'an instance of {self.__class__}')
228
229	1	return self.as_dict(include_id) == other.as_dict(include_id)
230
231
232	1	class ActionBase(ABC):
233		"""Base class for a flow action."""
234
235	1	def as_dict(self):
236		"""Return a dict that can be dumped as JSON."""
237	1	return vars(self)
238
239	1	@classmethod
240		def from_dict(cls, action_dict):
241		"""Return an action instance from attributes in a dictionary."""
242	1	action = cls(None)
243	1	for attr_name, value in action_dict.items():
244	1	if hasattr(action, attr_name):
245	1	setattr(action, attr_name, value)
246	1	return action
247
248	1	@abstractmethod
249		def as_of_action(self):
250		"""Return a pyof action to be used by a FlowMod."""
251
252	1	@classmethod
253	1	@abstractmethod
254		def from_of_action(cls, of_action):
255		"""Return an action from a pyof action."""
256
257
258	1	class ActionFactoryBase(ABC):
259		"""Deal with different implementations of ActionBase."""
260
261		# key: action_type or pyof class, value: ActionBase child
262	1	_action_class = {
263		'output': None,
264		'set_vlan': None,
265		# pyof class: ActionBase child
266		}
267
268	1	@classmethod
269		def from_dict(cls, action_dict):
270		"""Build the proper Action from a dictionary.
271
272		Args:
273		action_dict (dict): Action attributes.
274		"""
275	1	action_type = action_dict.get('action_type')
276	1	action_class = cls._action_class[action_type]
277	1	return action_class.from_dict(action_dict) if action_class else None
278
279	1	@classmethod
280		def from_of_action(cls, of_action):
281		"""Build the proper Action from a pyof action.
282
283		Args:
284		of_action (pyof action): Action from python-openflow.
285		"""
286		of_class = type(of_action)
287		action_class = cls._action_class.get(of_class)
288		return action_class.from_of_action(of_action) if action_class else None
289
290
291	1	class MatchBase: # pylint: disable=too-many-instance-attributes
292		"""Base class with common high-level Match fields."""
293
294	1	def __init__(self, in_port=None, dl_src=None, dl_dst=None, dl_vlan=None,
295		dl_vlan_pcp=None, dl_type=None, nw_proto=None, nw_src=None,
296		nw_dst=None, tp_src=None, tp_dst=None, in_phy_port=None,
297		ip_dscp=None, ip_ecn=None, udp_src=None, udp_dst=None,
298		sctp_src=None, sctp_dst=None, icmpv4_type=None,
299		icmpv4_code=None, arp_op=None, arp_spa=None, arp_tpa=None,
300		arp_sha=None, arp_tha=None, ipv6_src=None, ipv6_dst=None,
301		ipv6_flabel=None, icmpv6_type=None, icmpv6_code=None,
302		nd_tar=None, nd_sll=None, nd_tll=None, mpls_lab=None,
303		mpls_tc=None, mpls_bos=None, pbb_isid=None, v6_hdr=None,
304		metadata=None, tun_id=None):
305		"""Make it possible to set all attributes from the constructor."""
306		# pylint: disable=too-many-arguments
307		# pylint: disable=too-many-locals
308	1	self.in_port = in_port
309	1	self.dl_src = dl_src
310	1	self.dl_dst = dl_dst
311	1	self.dl_vlan = dl_vlan
312	1	self.dl_vlan_pcp = dl_vlan_pcp
313	1	self.dl_type = dl_type
314	1	self.nw_proto = nw_proto
315	1	self.nw_src = nw_src
316	1	self.nw_dst = nw_dst
317	1	self.tp_src = tp_src
318	1	self.tp_dst = tp_dst
319	1	self.in_phy_port = in_phy_port
320	1	self.ip_dscp = ip_dscp
321	1	self.ip_ecn = ip_ecn
322	1	self.udp_src = udp_src
323	1	self.udp_dst = udp_dst
324	1	self.sctp_src = sctp_src
325	1	self.sctp_dst = sctp_dst
326	1	self.icmpv4_type = icmpv4_type
327	1	self.icmpv4_code = icmpv4_code
328	1	self.arp_op = arp_op
329	1	self.arp_spa = arp_spa
330	1	self.arp_tpa = arp_tpa
331	1	self.arp_sha = arp_sha
332	1	self.arp_tha = arp_tha
333	1	self.ipv6_src = ipv6_src
334	1	self.ipv6_dst = ipv6_dst
335	1	self.ipv6_flabel = ipv6_flabel
336	1	self.icmpv6_type = icmpv6_type
337	1	self.icmpv6_code = icmpv6_code
338	1	self.nd_tar = nd_tar
339	1	self.nd_sll = nd_sll
340	1	self.nd_tll = nd_tll
341	1	self.mpls_lab = mpls_lab
342	1	self.mpls_tc = mpls_tc
343	1	self.mpls_bos = mpls_bos
344	1	self.pbb_isid = pbb_isid
345	1	self.v6_hdr = v6_hdr
346	1	self.metadata = metadata
347	1	self.tun_id = tun_id
348
349	1	def as_dict(self):
350		"""Return a dictionary excluding ``None`` values."""
351	1	return {k: v for k, v in self.__dict__.items() if v is not None}
352
353	1	@classmethod
354		def from_dict(cls, match_dict):
355		"""Return a Match instance from a dictionary."""
356	1	match = cls()
357	1	for key, value in match_dict.items():
358	1	if key in match.__dict__:
359	1	setattr(match, key, value)
360	1	return match
361
362	1	@classmethod
363	1	@abstractmethod
364		def from_of_match(cls, of_match):
365		"""Return a Match instance from a pyof Match."""
366
367	1	@abstractmethod
368		def as_of_match(self):
369		"""Return a python-openflow Match."""
370
371
372	1	class Stats:
373		"""Simple class to store statistics as attributes and values."""
374
375	1	def as_dict(self):
376		"""Return a dict excluding attributes with ``None`` value."""
377	1	return {attribute: value
378		for attribute, value in vars(self).items()
379		if value is not None}
380
381	1	@classmethod
382		def from_dict(cls, stats_dict):
383		"""Return a statistics object from a dictionary."""
384	1	stats = cls()
385	1	cls._update(stats, stats_dict.items())
386	1	return stats
387
388	1	@classmethod
389		def from_of_flow_stats(cls, of_stats):
390		"""Create an instance from a pyof FlowStats."""
391		stats = cls()
392		stats.update(of_stats)
393		return stats
394
395	1	def update(self, of_stats):
396		"""Given a pyof stats object, update attributes' values.
397
398		Avoid object creation and memory leak. pyof values are GenericType
399		instances whose native values can be accessed by `.value`.
400		"""
401		# Generator for GenericType values
402		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...
403		for attr_name, gen_type in vars(of_stats).items()
404		if attr_name in vars(self))
405		self._update(self, attr_name_value)
406
407	1	@staticmethod
408		def _update(obj, iterable):
409		"""From attribute name and value pairs, update ``obj``."""
410	1	for attr_name, value in iterable:
411		if hasattr(obj, attr_name):
412		setattr(obj, attr_name, value)
413
414
415	1	class FlowStats(Stats):
416		"""Common fields for 1.0 and 1.3 FlowStats."""
417
418	1	def __init__(self):
419		"""Initialize all statistics as ``None``."""
420	1	self.byte_count = None
421	1	self.duration_sec = None
422	1	self.duration_nsec = None
423	1	self.packet_count = None
424
425
426	1	class PortStats(Stats): # pylint: disable=too-many-instance-attributes
427		"""Common fields for 1.0 and 1.3 PortStats."""
428
429	1	def __init__(self):
430		"""Initialize all statistics as ``None``."""
431		self.rx_packets = None
432		self.tx_packets = None
433		self.rx_bytes = None
434		self.tx_bytes = None
435		self.rx_dropped = None
436		self.tx_dropped = None
437		self.rx_errors = None
438		self.tx_errors = None
439		self.rx_frame_err = None
440		self.rx_over_err = None
441		self.rx_crc_err = None
442		self.collisions = None
443

kytos / of_core

build.flow.FlowBase.__init__() A last analyzed 2021-05-26 15:17 UTC

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
last analyzed 2021-05-26 15:17 UTC