build.v0x04.flow.ActionPopVlan.from_of_action() - Code Metrics - kytos/of_core - Measure and Improve Code Quality continuously with Scrutinizer

build.v0x04.flow.ActionPopVlan.from_of_action() A
last analyzed 2021-05-26 15:17 UTC

↳ Parent: build.v0x04.flow

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.125

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	2
dl	0
loc	4
ccs	1
cts	2
cp	0.5
crap	1.125
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 ActionSetQueue as OFActionSetQueue
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', 'ActionSetQueue', 'ActionPushVlan',
           'ActionPopVlan', '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 ActionSetQueue(ActionBase):
    """Action to set a queue for the packet."""

    def __init__(self, queue_id):
        """Require the id of the queue.

        Args:
            queue_id (int): Queue ID.
        """
        self.queue_id = queue_id
        self.action_type = 'set_queue'

    @classmethod
    def from_of_action(cls, of_action):
        """Return a high-level ActionSetQueue instance from pyof class."""
        return cls(queue_id=of_action.queue_id.value)

    def as_of_action(self):
        """Return a pyof ActionSetQueue instance."""
        return OFActionSetQueue(queue_id=self.queue_id)


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,
        'set_queue': ActionSetQueue,
        OFActionOutput: ActionOutput,
        OFActionSetField: ActionSetVlan,
        OFActionPush: ActionPushVlan,
        OFActionPopVLAN: ActionPopVlan,
        OFActionSetQueue: ActionSetQueue
    }


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

kytos / of_core

build.v0x04.flow.ActionPopVlan.from_of_action() A last analyzed 2021-05-26 15:17 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

build.v0x04.flow.ActionPopVlan.from_of_action() A
last analyzed 2021-05-26 15:17 UTC