build.v0x04.flow.Flow.__init__() - Code Metrics - Inspection of "Merge pull request #62 from gleybersonandrade/unit..." - kytos/of_core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 75d007...4d9626 )

by Humberto

created 2020-04-02 17:17 UTC

build.v0x04.flow.Flow.init() A

↳ Parent: build.v0x04.flow

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	4
dl	0
loc	4
ccs	3
cts	3
cp	1
crap	1
rs	10
c	0
b	0
f	0

"""Deal with OpenFlow 1.3 specificities related to flows."""
from itertools import chain

from pyof.foundation.network_types import EtherType
from pyof.v0x04.common.action import ActionOutput as OFActionOutput
from pyof.v0x04.common.action import ActionPopVLAN as OFActionPopVLAN
from pyof.v0x04.common.action import ActionPush as OFActionPush
from pyof.v0x04.common.action import ActionSetField as OFActionSetField
from pyof.v0x04.common.action import ActionType
from pyof.v0x04.common.flow_instructions import (InstructionApplyAction,
                                                 InstructionType)
from pyof.v0x04.common.flow_match import Match as OFMatch
from pyof.v0x04.common.flow_match import (OxmMatchFields, OxmOfbMatchField,
                                          OxmTLV, VlanId)
from pyof.v0x04.controller2switch.flow_mod import FlowMod

from napps.kytos.of_core.flow import (ActionBase, ActionFactoryBase, FlowBase,
                                      FlowStats, MatchBase, PortStats)
from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory

__all__ = ('ActionOutput', 'ActionSetVlan', 'Action', 'Flow', 'FlowStats',
           'PortStats')


class Match(MatchBase):
    """High-level Match for OpenFlow 1.3 match fields."""

    @classmethod
    def from_of_match(cls, of_match):
        """Return an instance from a pyof Match."""
        match = cls()
        match_fields = (MatchFieldFactory.from_of_tlv(tlv)

                        for tlv in of_match.oxm_match_fields)
        for field in match_fields:
            if field is not None:
                setattr(match, field.name, field.value)
        return match

    def as_of_match(self):
        """Create an OF Match with TLVs from instance attributes."""
        oxm_fields = OxmMatchFields()
        for field_name, value in self.__dict__.items():
            if value is not None:
                field = MatchFieldFactory.from_name(field_name, value)
                if field:
                    tlv = field.as_of_tlv()
                    oxm_fields.append(tlv)
        return OFMatch(oxm_match_fields=oxm_fields)


class ActionOutput(ActionBase):

    """Action with an output port."""

    def __init__(self, port):
        """Require an output port.

        Args:
            port (int): Specific port number.
        """
        self.port = port
        self.action_type = 'output'

    @classmethod
    def from_of_action(cls, of_action):
        """Return a high-level ActionOuput instance from pyof ActionOutput."""
        return cls(port=of_action.port.value)

    def as_of_action(self):
        """Return a pyof ActionOuput instance."""
        return OFActionOutput(port=self.port)


class ActionPopVlan(ActionBase):
    """Action to pop the outermost VLAN tag."""

    def __init__(self, *args):  # pylint: disable=unused-argument
        """Initialize the action with the correct action_type."""
        self.action_type = 'pop_vlan'

    @classmethod
    def from_of_action(cls, of_action):
        """Return a high-level ActionPopVlan instance from the pyof class."""
        return cls()

    def as_of_action(self):
        """Return a pyof ActionPopVLAN instance."""
        return OFActionPopVLAN()


class ActionPushVlan(ActionBase):
    """Action to push a VLAN tag."""

    def __init__(self, tag_type):
        """Require a tag_type for the VLAN."""
        self.action_type = 'push_vlan'
        self.tag_type = tag_type

    @classmethod
    def from_of_action(cls, of_action):
        """Return a high level ActionPushVlan instance from pyof ActionPush."""
        if of_action.ethertype.value == EtherType.VLAN_QINQ:
            return cls(tag_type='s')
        return cls(tag_type='c')

    def as_of_action(self):
        """Return a pyof ActionPush instance."""
        if self.tag_type == 's':
            return OFActionPush(action_type=ActionType.OFPAT_PUSH_VLAN,
                                ethertype=EtherType.VLAN_QINQ)
        return OFActionPush(action_type=ActionType.OFPAT_PUSH_VLAN,
                            ethertype=EtherType.VLAN)


class ActionSetVlan(ActionBase):
    """Action to set VLAN ID."""

    def __init__(self, vlan_id):
        """Require a VLAN ID."""
        self.vlan_id = vlan_id
        self.action_type = 'set_vlan'

    @classmethod
    def from_of_action(cls, of_action):
        """Return high-level ActionSetVlan object from pyof ActionSetField."""
        vlan_id = int.from_bytes(of_action.field.oxm_value, 'big') & 4095
        return cls(vlan_id)

    def as_of_action(self):
        """Return a pyof ActionSetField instance."""
        tlv = OxmTLV()
        tlv.oxm_field = OxmOfbMatchField.OFPXMT_OFB_VLAN_VID
        oxm_value = self.vlan_id | VlanId.OFPVID_PRESENT
        tlv.oxm_value = oxm_value.to_bytes(2, 'big')
        return OFActionSetField(field=tlv)


class Action(ActionFactoryBase):
    """An action to be executed once a flow is activated.

    This class behavies like a factory but has no "Factory" suffix for end-user
    usability issues.
    """

    # Set v0x04 classes for action types and pyof classes
    _action_class = {
        'output': ActionOutput,
        'set_vlan': ActionSetVlan,
        'push_vlan': ActionPushVlan,
        'pop_vlan': ActionPopVlan,
        OFActionOutput: ActionOutput,
        OFActionSetField: ActionSetVlan,
        OFActionPush: ActionPushVlan,
        OFActionPopVLAN: ActionPopVlan
    }


class Flow(FlowBase):
    """High-level flow representation for OpenFlow 1.0.

    This is a subclass that only deals with 1.3 flow actions.
    """

    _action_factory = Action
    _flow_mod_class = FlowMod
    _match_class = Match

    def __init__(self, *args, cookie_mask=0, **kwargs):
        """Require a cookie mask."""
        super().__init__(*args, **kwargs)
        self.cookie_mask = cookie_mask

    @staticmethod
    def _get_of_actions(of_flow_stats):
        """Return the pyof actions from pyof ``FlowStats.instructions``."""
        # Add list of high-level actions
        # Filter action instructions
        apply_actions = InstructionType.OFPIT_APPLY_ACTIONS
        of_instructions = (ins for ins in of_flow_stats.instructions

                           if ins.instruction_type == apply_actions)
        # Get actions from a list of actions
        return chain.from_iterable(ins.actions for ins in of_instructions)

    def _as_of_flow_mod(self, command):
        """Return pyof FlowMod with a ``command`` to add or delete a flow.

        Actions become items of the ``instructions`` attribute.
        """
        of_flow_mod = super()._as_of_flow_mod(command)
        of_flow_mod.cookie_mask = self.cookie_mask
        of_actions = [action.as_of_action() for action in self.actions]
        of_instruction = InstructionApplyAction(actions=of_actions)
        of_flow_mod.instructions = [of_instruction]
        return of_flow_mod


1			"""Deal with OpenFlow 1.3 specificities related to flows."""
2	1		from itertools import chain
3
4	1		from pyof.foundation.network_types import EtherType
5	1		from pyof.v0x04.common.action import ActionOutput as OFActionOutput
6	1		from pyof.v0x04.common.action import ActionPopVLAN as OFActionPopVLAN
7	1		from pyof.v0x04.common.action import ActionPush as OFActionPush
8	1		from pyof.v0x04.common.action import ActionSetField as OFActionSetField
9	1		from pyof.v0x04.common.action import ActionType
10	1		from pyof.v0x04.common.flow_instructions import (InstructionApplyAction,
11			InstructionType)
12	1		from pyof.v0x04.common.flow_match import Match as OFMatch
13	1		from pyof.v0x04.common.flow_match import (OxmMatchFields, OxmOfbMatchField,
14			OxmTLV, VlanId)
15	1		from pyof.v0x04.controller2switch.flow_mod import FlowMod
16
17	1		from napps.kytos.of_core.flow import (ActionBase, ActionFactoryBase, FlowBase,
18			FlowStats, MatchBase, PortStats)
19	1		from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory
20
21	1		__all__ = ('ActionOutput', 'ActionSetVlan', 'Action', 'Flow', 'FlowStats',
22			'PortStats')
23
24
25	1		class Match(MatchBase):
26			"""High-level Match for OpenFlow 1.3 match fields."""
27
28	1		@classmethod
29			def from_of_match(cls, of_match):
30			"""Return an instance from a pyof Match."""
31	1		match = cls()
32	1		match_fields = (MatchFieldFactory.from_of_tlv(tlv)
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-02-18 13:53 UTC by Report Bug Copy Issue Report The variable `tlv` does not seem to be defined. Loading history...
33			for tlv in of_match.oxm_match_fields)
34	1		for field in match_fields:
35	1		if field is not None:
36	1		setattr(match, field.name, field.value)
37	1		return match
38
39	1		def as_of_match(self):
40			"""Create an OF Match with TLVs from instance attributes."""
41	1		oxm_fields = OxmMatchFields()
42	1		for field_name, value in self.__dict__.items():
43	1		if value is not None:
44	1		field = MatchFieldFactory.from_name(field_name, value)
45	1		if field:
46	1		tlv = field.as_of_tlv()
47	1		oxm_fields.append(tlv)
48	1		return OFMatch(oxm_match_fields=oxm_fields)
49
50
51	1	View Code Duplication	class ActionOutput(ActionBase):
			0 ignored issues – show Duplication introduced 2019-02-18 13:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
52			"""Action with an output port."""
53
54	1		def __init__(self, port):
55			"""Require an output port.
56
57			Args:
58			port (int): Specific port number.
59			"""
60	1		self.port = port
61	1		self.action_type = 'output'
62
63	1		@classmethod
64			def from_of_action(cls, of_action):
65			"""Return a high-level ActionOuput instance from pyof ActionOutput."""
66	1		return cls(port=of_action.port.value)
67
68	1		def as_of_action(self):
69			"""Return a pyof ActionOuput instance."""
70			return OFActionOutput(port=self.port)
71
72
73	1		class ActionPopVlan(ActionBase):
74			"""Action to pop the outermost VLAN tag."""
75
76	1		def __init__(self, *args): # pylint: disable=unused-argument
77			"""Initialize the action with the correct action_type."""
78			self.action_type = 'pop_vlan'
79
80	1		@classmethod
81			def from_of_action(cls, of_action):
82			"""Return a high-level ActionPopVlan instance from the pyof class."""
83			return cls()
84
85	1		def as_of_action(self):
86			"""Return a pyof ActionPopVLAN instance."""
87			return OFActionPopVLAN()
88
89
90	1		class ActionPushVlan(ActionBase):
91			"""Action to push a VLAN tag."""
92
93	1		def __init__(self, tag_type):
94			"""Require a tag_type for the VLAN."""
95			self.action_type = 'push_vlan'
96			self.tag_type = tag_type
97
98	1		@classmethod
99			def from_of_action(cls, of_action):
100			"""Return a high level ActionPushVlan instance from pyof ActionPush."""
101			if of_action.ethertype.value == EtherType.VLAN_QINQ:
102			return cls(tag_type='s')
103			return cls(tag_type='c')
104
105	1		def as_of_action(self):
106			"""Return a pyof ActionPush instance."""
107			if self.tag_type == 's':
108			return OFActionPush(action_type=ActionType.OFPAT_PUSH_VLAN,
109			ethertype=EtherType.VLAN_QINQ)
110			return OFActionPush(action_type=ActionType.OFPAT_PUSH_VLAN,
111			ethertype=EtherType.VLAN)
112
113
114	1		class ActionSetVlan(ActionBase):
115			"""Action to set VLAN ID."""
116
117	1		def __init__(self, vlan_id):
118			"""Require a VLAN ID."""
119	1		self.vlan_id = vlan_id
120	1		self.action_type = 'set_vlan'
121
122	1		@classmethod
123			def from_of_action(cls, of_action):
124			"""Return high-level ActionSetVlan object from pyof ActionSetField."""
125			vlan_id = int.from_bytes(of_action.field.oxm_value, 'big') & 4095
126			return cls(vlan_id)
127
128	1		def as_of_action(self):
129			"""Return a pyof ActionSetField instance."""
130	1		tlv = OxmTLV()
131	1		tlv.oxm_field = OxmOfbMatchField.OFPXMT_OFB_VLAN_VID
132	1		oxm_value = self.vlan_id \| VlanId.OFPVID_PRESENT
133	1		tlv.oxm_value = oxm_value.to_bytes(2, 'big')
134	1		return OFActionSetField(field=tlv)
135
136
137	1		class Action(ActionFactoryBase):
138			"""An action to be executed once a flow is activated.
139
140			This class behavies like a factory but has no "Factory" suffix for end-user
141			usability issues.
142			"""
143
144			# Set v0x04 classes for action types and pyof classes
145	1		_action_class = {
146			'output': ActionOutput,
147			'set_vlan': ActionSetVlan,
148			'push_vlan': ActionPushVlan,
149			'pop_vlan': ActionPopVlan,
150			OFActionOutput: ActionOutput,
151			OFActionSetField: ActionSetVlan,
152			OFActionPush: ActionPushVlan,
153			OFActionPopVLAN: ActionPopVlan
154			}
155
156
157	1		class Flow(FlowBase):
158			"""High-level flow representation for OpenFlow 1.0.
159
160			This is a subclass that only deals with 1.3 flow actions.
161			"""
162
163	1		_action_factory = Action
164	1		_flow_mod_class = FlowMod
165	1		_match_class = Match
166
167	1		def __init__(self, args, cookie_mask=0, *kwargs):
168			"""Require a cookie mask."""
169	1		super().__init__(args, *kwargs)
170	1		self.cookie_mask = cookie_mask
171
172	1		@staticmethod
173			def _get_of_actions(of_flow_stats):
174			"""Return the pyof actions from pyof ``FlowStats.instructions``."""
175			# Add list of high-level actions
176			# Filter action instructions
177	1		apply_actions = InstructionType.OFPIT_APPLY_ACTIONS
178	1		of_instructions = (ins for ins in of_flow_stats.instructions
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-02-18 13:53 UTC by Report Bug Copy Issue Report The variable `ins` does not seem to be defined. Loading history...
179			if ins.instruction_type == apply_actions)
180			# Get actions from a list of actions
181	1		return chain.from_iterable(ins.actions for ins in of_instructions)
182
183	1		def _as_of_flow_mod(self, command):
184			"""Return pyof FlowMod with a ``command`` to add or delete a flow.
185
186			Actions become items of the ``instructions`` attribute.
187			"""
188	1		of_flow_mod = super()._as_of_flow_mod(command)
189	1		of_flow_mod.cookie_mask = self.cookie_mask
190	1		of_actions = [action.as_of_action() for action in self.actions]
191	1		of_instruction = InstructionApplyAction(actions=of_actions)
192	1		of_flow_mod.instructions = [of_instruction]
193			return of_flow_mod
194

kytos / of_core

Push — master ( 75d007...4d9626 )

build.v0x04.flow.Flow.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

build.v0x04.flow.Flow.init() A